LISTEN/NOTIFY funcionality

Signed-off-by: chandr-andr (Kiselev Aleksandr) <chandr@chandr.net>

Author: chandr-andr

python/psqlpy/_internal/__init__.pyi

@@ -1799,15 +1799,15 @@ class Listener:
         - `channel`: name of the channel.
         """
 
-    async def listen(self: Self) -> None:
+    def listen(self: Self) -> None:
         """Start listening.
 
         Start actual listening.
         In the background it creates task in Rust event loop.
         You must save returned Future to the array.
         """
 
-    async def abort_listen(self: Self) -> None:
+    def abort_listen(self: Self) -> None:
         """Abort listen.
 
         If `listen()` method was called, stop listening,

src/common.rs

@@ -4,7 +4,10 @@ use pyo3::{
 };
 
 use crate::{
-    driver::connection::InnerConnection, exceptions::rust_errors::RustPSQLDriverPyResult, query_result::{PSQLDriverPyQueryResult, PSQLDriverSinglePyQueryResult}, value_converter::{convert_parameters, PythonDTO, QueryParameter}
+    driver::connection::PsqlpyConnection,
+    exceptions::rust_errors::RustPSQLDriverPyResult,
+    query_result::{PSQLDriverPyQueryResult, PSQLDriverSinglePyQueryResult},
+    value_converter::{convert_parameters, PythonDTO, QueryParameter},
 };
 
 /// Add new module to the parent one.
@@ -52,7 +55,7 @@ pub trait ObjectQueryTrait {
     ) -> impl std::future::Future<Output = RustPSQLDriverPyResult<()>> + Send;
 }
 
-impl ObjectQueryTrait for InnerConnection {
+impl ObjectQueryTrait for PsqlpyConnection {
     async fn psqlpy_query_one(
         &self,
         querystring: String,
@@ -128,6 +131,6 @@ impl ObjectQueryTrait for InnerConnection {
     }
 
     async fn psqlpy_query_simple(&self, querystring: String) -> RustPSQLDriverPyResult<()> {
-        Ok(self.batch_execute(querystring.as_str()).await?)
+        self.batch_execute(querystring.as_str()).await
     }
 }

src/driver/connection.rs

@@ -4,7 +4,9 @@ use futures_util::pin_mut;
 use postgres_types::ToSql;
 use pyo3::{buffer::PyBuffer, pyclass, pymethods, Py, PyAny, PyErr, Python};
 use std::{collections::HashSet, sync::Arc, vec};
-use tokio_postgres::{binary_copy::BinaryCopyInWriter, Client, CopyInSink, Row, Statement, ToStatement};
+use tokio_postgres::{
+    binary_copy::BinaryCopyInWriter, Client, CopyInSink, Row, Statement, ToStatement,
+};
 
 use crate::{
     exceptions::rust_errors::{RustPSQLDriverError, RustPSQLDriverPyResult},
@@ -20,108 +22,114 @@ use super::{
     transaction_options::{IsolationLevel, ReadVariant, SynchronousCommit},
 };
 
-pub enum InnerConnection {
+#[allow(clippy::module_name_repetitions)]
+pub enum PsqlpyConnection {
     PoolConn(Object),
     SingleConn(Client),
 }
 
-impl InnerConnection {
-    pub async fn prepare_cached(
-        &self,
-        query: &str
-    ) -> RustPSQLDriverPyResult<Statement> {
+impl PsqlpyConnection {
+    /// Prepare cached statement.
+    ///
+    /// # Errors
+    /// May return Err if cannot prepare statement.
+    pub async fn prepare_cached(&self, query: &str) -> RustPSQLDriverPyResult<Statement> {
         match self {
-            InnerConnection::PoolConn(pconn) => {
-                return Ok(pconn.prepare_cached(query).await?)
-            }
-            InnerConnection::SingleConn(sconn) => {
-                return Ok(sconn.prepare(query).await?)
-            }
+            PsqlpyConnection::PoolConn(pconn) => return Ok(pconn.prepare_cached(query).await?),
+            PsqlpyConnection::SingleConn(sconn) => return Ok(sconn.prepare(query).await?),
         }
     }
 
+    /// Prepare cached statement.
+    ///
+    /// # Errors
+    /// May return Err if cannot execute statement.
     pub async fn query<T>(
         &self,
         statement: &T,
         params: &[&(dyn ToSql + Sync)],
     ) -> RustPSQLDriverPyResult<Vec<Row>>
-    where T: ?Sized + ToStatement {
+    where
+        T: ?Sized + ToStatement,
+    {
         match self {
-            InnerConnection::PoolConn(pconn) => {
-                return Ok(pconn.query(statement, params).await?)
-            }
-            InnerConnection::SingleConn(sconn) => {
+            PsqlpyConnection::PoolConn(pconn) => return Ok(pconn.query(statement, params).await?),
+            PsqlpyConnection::SingleConn(sconn) => {
                 return Ok(sconn.query(statement, params).await?)
             }
         }
     }
 
+    /// Prepare cached statement.
+    ///
+    /// # Errors
+    /// May return Err if cannot execute statement.
     pub async fn batch_execute(&self, query: &str) -> RustPSQLDriverPyResult<()> {
         match self {
-            InnerConnection::PoolConn(pconn) => {
-                return Ok(pconn.batch_execute(query).await?)
-            }
-            InnerConnection::SingleConn(sconn) => {
-                return Ok(sconn.batch_execute(query).await?)
-            }
+            PsqlpyConnection::PoolConn(pconn) => return Ok(pconn.batch_execute(query).await?),
+            PsqlpyConnection::SingleConn(sconn) => return Ok(sconn.batch_execute(query).await?),
         }
     }
 
+    /// Prepare cached statement.
+    ///
+    /// # Errors
+    /// May return Err if cannot execute statement.
     pub async fn query_one<T>(
         &self,
         statement: &T,
         params: &[&(dyn ToSql + Sync)],
     ) -> RustPSQLDriverPyResult<Row>
-    where T: ?Sized + ToStatement
+    where
+        T: ?Sized + ToStatement,
     {
         match self {
-            InnerConnection::PoolConn(pconn) => {
+            PsqlpyConnection::PoolConn(pconn) => {
                 return Ok(pconn.query_one(statement, params).await?)
             }
-            InnerConnection::SingleConn(sconn) => {
+            PsqlpyConnection::SingleConn(sconn) => {
                 return Ok(sconn.query_one(statement, params).await?)
             }
         }
     }
 
-    pub async fn copy_in<T, U>(
-        &self,
-        statement: &T
-    ) -> RustPSQLDriverPyResult<CopyInSink<U>>
+    /// Prepare cached statement.
+    ///
+    /// # Errors
+    /// May return Err if cannot execute copy data.
+    pub async fn copy_in<T, U>(&self, statement: &T) -> RustPSQLDriverPyResult<CopyInSink<U>>
     where
         T: ?Sized + ToStatement,
-        U: Buf + 'static + Send
+        U: Buf + 'static + Send,
     {
         match self {
-            InnerConnection::PoolConn(pconn) => {
-                return Ok(pconn.copy_in(statement).await?)
-            }
-            InnerConnection::SingleConn(sconn) => {
-                return Ok(sconn.copy_in(statement).await?)
-            }
+            PsqlpyConnection::PoolConn(pconn) => return Ok(pconn.copy_in(statement).await?),
+            PsqlpyConnection::SingleConn(sconn) => return Ok(sconn.copy_in(statement).await?),
         }
     }
 }
 
 #[pyclass(subclass)]
 #[derive(Clone)]
 pub struct Connection {
-    db_client: Option<Arc<InnerConnection>>,
+    db_client: Option<Arc<PsqlpyConnection>>,
     db_pool: Option<Pool>,
 }
 
 impl Connection {
     #[must_use]
-    pub fn new(db_client: Option<Arc<InnerConnection>>, db_pool: Option<Pool>) -> Self {
+    pub fn new(db_client: Option<Arc<PsqlpyConnection>>, db_pool: Option<Pool>) -> Self {
         Connection { db_client, db_pool }
     }
 
-    pub fn db_client(&self) -> Option<Arc<InnerConnection>> {
-        return self.db_client.clone()
+    #[must_use]
+    pub fn db_client(&self) -> Option<Arc<PsqlpyConnection>> {
+        self.db_client.clone()
     }
 
+    #[must_use]
     pub fn db_pool(&self) -> Option<Pool> {
-        return self.db_pool.clone()
+        self.db_pool.clone()
     }
 }
 
@@ -151,7 +159,7 @@ impl Connection {
                 .await??;
             pyo3::Python::with_gil(|gil| {
                 let mut self_ = self_.borrow_mut(gil);
-                self_.db_client = Some(Arc::new(InnerConnection::PoolConn(db_connection)));
+                self_.db_client = Some(Arc::new(PsqlpyConnection::PoolConn(db_connection)));
             });
             return Ok(self_);
         }
@@ -277,7 +285,7 @@ impl Connection {
         let db_client = pyo3::Python::with_gil(|gil| self_.borrow(gil).db_client.clone());
 
         if let Some(db_client) = db_client {
-            return Ok(db_client.batch_execute(&querystring).await?);
+            return db_client.batch_execute(&querystring).await;
         }
 
         Err(RustPSQLDriverError::ConnectionClosedError)

src/driver/connection_pool.rs

@@ -11,7 +11,10 @@ use crate::{
 };
 
 use super::{
-    common_options::{ConnRecyclingMethod, LoadBalanceHosts, SslMode, TargetSessionAttrs}, connection::{Connection, InnerConnection}, listener::core::Listener, utils::{build_connection_config, build_manager, build_tls}
+    common_options::{ConnRecyclingMethod, LoadBalanceHosts, SslMode, TargetSessionAttrs},
+    connection::{Connection, PsqlpyConnection},
+    listener::core::Listener,
+    utils::{build_connection_config, build_manager, build_tls},
 };
 
 /// Make new connection pool.
@@ -210,7 +213,8 @@ pub struct ConnectionPool {
 }
 
 impl ConnectionPool {
-    #[must_use] pub fn build(
+    #[must_use]
+    pub fn build(
         pool: Pool,
         pg_config: Config,
         ca_file: Option<String>,
@@ -497,9 +501,9 @@ impl ConnectionPool {
         Connection::new(None, Some(self.pool.clone()))
     }
 
-    pub fn listener(
-        self_: pyo3::Py<Self>,
-    ) -> RustPSQLDriverPyResult<Listener> {
+    #[must_use]
+    #[allow(clippy::needless_pass_by_value)]
+    pub fn listener(self_: pyo3::Py<Self>) -> Listener {
         let (pg_config, ca_file, ssl_mode) = pyo3::Python::with_gil(|gil| {
             let b_gil = self_.borrow(gil);
             (
@@ -509,7 +513,7 @@ impl ConnectionPool {
             )
         });
 
-        Ok(Listener::new(pg_config, ca_file, ssl_mode))
+        Listener::new(pg_config, ca_file, ssl_mode)
     }
 
     /// Return new single connection.
@@ -524,7 +528,10 @@ impl ConnectionPool {
             })
             .await??;
 
-        Ok(Connection::new(Some(Arc::new(InnerConnection::PoolConn(db_connection))), None))
+        Ok(Connection::new(
+            Some(Arc::new(PsqlpyConnection::PoolConn(db_connection))),
+            None,
+        ))
     }
 
     /// Close connection pool.

src/driver/cursor.rs

@@ -11,7 +11,7 @@ use crate::{
     runtime::rustdriver_future,
 };
 
-use super::connection::InnerConnection;
+use super::connection::PsqlpyConnection;
 
 /// Additional implementation for the `Object` type.
 #[allow(clippy::ref_option)]
@@ -28,7 +28,7 @@ trait CursorObjectTrait {
     async fn cursor_close(&self, closed: &bool, cursor_name: &str) -> RustPSQLDriverPyResult<()>;
 }
 
-impl CursorObjectTrait for InnerConnection {
+impl CursorObjectTrait for PsqlpyConnection {
     /// Start the cursor.
     ///
     /// Execute `DECLARE` command with parameters.
@@ -90,7 +90,7 @@ impl CursorObjectTrait for InnerConnection {
 
 #[pyclass(subclass)]
 pub struct Cursor {
-    db_transaction: Option<Arc<InnerConnection>>,
+    db_transaction: Option<Arc<PsqlpyConnection>>,
     querystring: String,
     parameters: Option<Py<PyAny>>,
     cursor_name: String,
@@ -104,7 +104,7 @@ pub struct Cursor {
 impl Cursor {
     #[must_use]
     pub fn new(
-        db_transaction: Arc<InnerConnection>,
+        db_transaction: Arc<PsqlpyConnection>,
         querystring: String,
         parameters: Option<Py<PyAny>>,
         cursor_name: String,