Added connection class

Author: chandr-andr

README.md

@@ -52,6 +52,7 @@ async def main() -> None:
     # rust does it instead.
 
 ```
+Please take into account that each new execute gets new connection from connection pool. 
 
 ## Query parameters
 You can pass parameters into queries.  
@@ -65,7 +66,37 @@ Any placeholder must be marked with `$< num>`.
     )
 ```
 
-Support all types from Python is in development, but there are support for all basic types. More you can find here: ... 
+## Connection
+You can work with connection instead of DatabasePool.
+```python
+from typing import Any
+import asyncio
+
+from rust_psql_driver import PSQLPool
+
+
+db_pool = PSQLPool(
+    username="postgres",
+    password="pg_password",
+    host="localhost",
+    port=5432,
+    db_name="postgres",
+    max_db_pool_size=2,
+)
+
+async def main() -> None:
+    await db_pool.startup()
+
+    connection = await db_pool.connection()
+
+    res: list[dict[str, Any]] = await connection.execute(
+        "SELECT * FROM users",
+    )
+
+    print(res)
+    # You don't need to close connection by yourself,
+    # rust does it instead.
+```
 
 ## Transactions
 Of course it's possible to use transactions with this driver.  
@@ -85,7 +116,8 @@ db_pool = PSQLPool()
 async def main() -> None:
     await db_pool.startup()
 
-    transaction = await db_pool.transaction(
+    connection = await db_pool.connection()
+    transaction = await connection.transaction(
         isolation_level=IsolationLevel.Serializable,
     )
 
@@ -112,7 +144,8 @@ db_pool = PSQLPool()
 async def main() -> None:
     await db_pool.startup()
 
-    transaction = await db_pool.transaction(
+    connection = await db_pool.connection()
+    transaction = await connection.transaction(
         isolation_level=IsolationLevel.Serializable,
     )
 
@@ -146,7 +179,8 @@ db_pool = PSQLPool()
 async def main() -> None:
     await db_pool.startup()
 
-    transaction = await db_pool.transaction(
+    connection = await db_pool.connection()
+    transaction = await connection.transaction(
         isolation_level=IsolationLevel.Serializable,
     )
 
@@ -173,7 +207,8 @@ db_pool = PSQLPool()
 async def main() -> None:
     await db_pool.startup()
 
-    transaction = await db_pool.transaction(
+    connection = await db_pool.connection()
+    transaction = await connection.transaction(
         isolation_level=IsolationLevel.Serializable,
     )
 
@@ -204,7 +239,8 @@ db_pool = PSQLPool()
 async def main() -> None:
     await db_pool.startup()
 
-    transaction = await db_pool.transaction(
+    connection = await db_pool.connection()
+    transaction = await connection.transaction(
         isolation_level=IsolationLevel.Serializable,
     )
 

python/psql_rust_driver/_internal/__init__.pyi

@@ -206,6 +206,57 @@ class Transaction:
             await transaction.release_savepoint
         ```
         """
+    
+
+class Connection:
+    """Connection from Database Connection Pool.
+    
+    It can be created only from connection pool.
+    """
+    
+    async def execute(
+        self: Self,
+        querystring: str,
+        parameters: List[Any] | None = None,
+    ) -> QueryResult:
+        """Execute the query.
+        
+        Querystring can contain `$<number>` parameters
+        for converting them in the driver side.
+
+        ### Parameters:
+        - `querystring`: querystring to execute.
+        - `parameters`: list of parameters to pass in the query.
+
+        ### Example:
+        ```python
+        import asyncio
+
+        from psql_rust_driver import PSQLPool, QueryResult
+
+
+        async def main() -> None:
+            db_pool = PSQLPool()
+            await db_pool.startup()
+
+            connection = await db_pool.connection()
+            query_result: QueryResult = await connection.execute(
+                "SELECT username FROM users WHERE id = $1",
+                [100],
+            )
+            dict_result: List[Dict[Any, Any]] = query_result.result()
+        ```
+        """
+    
+    async def transaction(
+        self,
+        isolation_level: IsolationLevel | None = IsolationLevel.ReadCommitted,
+    ) -> Transaction:
+        """Create new transaction.
+
+        ### Parameters:
+        - `isolation_level`: configure isolation level of the transaction.
+        """
 
 
 class PSQLPool:
@@ -284,15 +335,8 @@ class PSQLPool:
         """
         ...
 
-    async def transaction(
-        self,
-        isolation_level: IsolationLevel | None = IsolationLevel.ReadCommitted,
-    ) -> Transaction:
-        """Create new transaction.
+    async def connection(self: Self) -> Connection:
+        """Create new connection.
         
-        It acquires new connection from the database pool
-        and make it acts as transaction.
-
-        ### Parameters:
-        - `isolation_level`: configure isolation level of the transaction.
+        It acquires new connection from the database pool.
         """

src/driver/connection.rs

@@ -0,0 +1,118 @@
+use deadpool_postgres::Object;
+use pyo3::{pyclass, pymethods, PyAny, Python};
+use std::{collections::HashSet, sync::Arc, vec};
+use tokio_postgres::types::ToSql;
+
+use crate::{
+    common::rustengine_future,
+    exceptions::rust_errors::RustPSQLDriverPyResult,
+    query_result::PSQLDriverPyQueryResult,
+    value_converter::{convert_parameters, PythonDTO},
+};
+
+use super::{
+    transaction::{RustTransaction, Transaction},
+    transaction_options::IsolationLevel,
+};
+
+pub struct RustConnection {
+    pub db_client: Arc<tokio::sync::RwLock<Object>>,
+}
+
+impl RustConnection {
+    /// Execute querystring with parameters.
+    ///
+    /// Method doesn't acquire lock on database connection.
+    /// It prepares and caches querystring in the inner Object object.
+    ///
+    /// Then execute the query.
+    ///
+    /// # Errors:
+    /// May return Err Result if:
+    /// 1) Can not create/retrieve prepared statement
+    /// 2) Can not execute statement
+    pub async fn inner_execute<'a>(
+        &'a self,
+        querystring: String,
+        parameters: Vec<PythonDTO>,
+    ) -> RustPSQLDriverPyResult<PSQLDriverPyQueryResult> {
+        let db_client_arc = self.db_client.clone();
+
+        let db_client_guard = db_client_arc.read().await;
+
+        let mut vec_parameters: Vec<&(dyn ToSql + Sync)> = Vec::with_capacity(parameters.len());
+        for param in parameters.iter() {
+            vec_parameters.push(param);
+        }
+
+        let statement: tokio_postgres::Statement =
+            db_client_guard.prepare_cached(&querystring).await?;
+
+        let result = db_client_guard
+            .query(&statement, &vec_parameters.into_boxed_slice())
+            .await?;
+
+        Ok(PSQLDriverPyQueryResult::new(result))
+    }
+
+    pub fn inner_transaction<'a>(&'a self, isolation_level: Option<IsolationLevel>) -> Transaction {
+        let inner_transaction = RustTransaction {
+            db_client: self.db_client.clone(),
+            is_started: Arc::new(tokio::sync::RwLock::new(false)),
+            is_done: Arc::new(tokio::sync::RwLock::new(false)),
+            rollback_savepoint: Arc::new(tokio::sync::RwLock::new(HashSet::new())),
+            isolation_level: isolation_level,
+        };
+
+        Transaction {
+            transaction: Arc::new(tokio::sync::RwLock::new(inner_transaction)),
+        }
+    }
+}
+
+#[pyclass]
+pub struct Connection(pub Arc<tokio::sync::RwLock<RustConnection>>);
+
+#[pymethods]
+impl Connection {
+    /// Execute querystring with parameters.
+    ///
+    /// It converts incoming parameters to rust readable
+    /// and then execute the query with them.
+    ///
+    /// # Errors:
+    ///
+    /// May return Err Result if:
+    /// 1) Cannot convert python parameters
+    /// 2) Cannot execute querystring.
+    pub fn execute<'a>(
+        &'a self,
+        py: Python<'a>,
+        querystring: String,
+        parameters: Option<&'a PyAny>,
+    ) -> RustPSQLDriverPyResult<&PyAny> {
+        let connection_arc = self.0.clone();
+        let mut params: Vec<PythonDTO> = vec![];
+        if let Some(parameters) = parameters {
+            params = convert_parameters(parameters)?
+        }
+
+        rustengine_future(py, async move {
+            let connection_guard = connection_arc.read().await;
+            Ok(connection_guard.inner_execute(querystring, params).await?)
+        })
+    }
+
+    pub fn transaction<'a>(
+        &'a self,
+        py: Python<'a>,
+        isolation_level: Option<IsolationLevel>,
+    ) -> RustPSQLDriverPyResult<&PyAny> {
+        let connection_arc = self.0.clone();
+
+        rustengine_future(py, async move {
+            let connection_guard = connection_arc.read().await;
+            Ok(connection_guard.inner_transaction(isolation_level))
+        })
+    }
+}

src/driver/connection_pool.rs

@@ -11,7 +11,10 @@ use crate::{
     value_converter::{convert_parameters, PythonDTO},
 };
 
-use super::transaction_options::IsolationLevel;
+use super::{
+    connection::{Connection, RustConnection},
+    transaction_options::IsolationLevel,
+};
 
 /// PSQLPool for internal use only.
 ///
@@ -49,6 +52,27 @@ impl RustPSQLPool {
 }
 
 impl RustPSQLPool {
+    pub async fn inner_connection<'a>(&'a self) -> RustPSQLDriverPyResult<Connection> {
+        let db_pool_arc = self.db_pool.clone();
+
+        let db_pool_guard = db_pool_arc.read().await;
+
+        let db_pool_manager = db_pool_guard
+            .as_ref()
+            .ok_or(RustPSQLDriverError::DatabasePoolError(
+                "Database pool is not initialized".into(),
+            ))?
+            .get()
+            .await?;
+
+        let inner_connection = RustConnection {
+            db_client: Arc::new(tokio::sync::RwLock::new(db_pool_manager)),
+        };
+
+        Ok(Connection(Arc::new(tokio::sync::RwLock::new(
+            inner_connection,
+        ))))
+    }
     /// Execute querystring with parameters.
     ///
     /// Prepare statement and cache it, then execute.
@@ -223,23 +247,16 @@ impl PSQLPool {
         })
     }
 
-    /// Return python transaction.
+    /// Return single connection.
     ///
     /// # Errors:
-    /// May return Err Result if `inner_transaction` returns error.
-    pub fn transaction<'a>(
-        &'a self,
-        py: Python<'a>,
-        isolation_level: Option<IsolationLevel>,
-    ) -> RustPSQLDriverPyResult<&'a PyAny> {
+    /// May return Err Result if `inner_connection` returns error.
+    pub fn connection<'a>(&'a self, py: Python<'a>) -> RustPSQLDriverPyResult<&'a PyAny> {
         let psql_pool_arc = self.rust_psql_pool.clone();
 
         rustengine_future(py, async move {
             let psql_pool_guard = psql_pool_arc.write().await;
-
-            let transaction = psql_pool_guard.inner_transaction(isolation_level).await?;
-
-            Ok(transaction)
+            Ok(psql_pool_guard.inner_connection().await?)
         })
     }
 

src/driver/mod.rs

@@ -1,3 +1,4 @@
+pub mod connection;
 pub mod connection_pool;
 pub mod transaction;
 pub mod transaction_options;

src/driver/transaction.rs

@@ -67,8 +67,7 @@ impl RustTransaction {
             vec_parameters.push(param);
         }
 
-        let statement: tokio_postgres::Statement =
-            db_client_guard.prepare_cached(&querystring).await?;
+        let statement = db_client_guard.prepare_cached(&querystring).await?;
 
         let result = db_client_guard
             .query(&statement, &vec_parameters.into_boxed_slice())