Added Portal impl

Author: chandr-andr

src/driver/inner_transaction.rs

@@ -6,21 +6,21 @@ use crate::exceptions::rust_errors::PSQLPyResult;
 
 pub enum PsqlpyTransaction {
     PoolTrans(dp_Transaction<'static>),
-    SingleConnTrans(tp_Transaction<'static>)
+    SingleConnTrans(tp_Transaction<'static>),
 }
 
 impl PsqlpyTransaction {
     async fn commit(self) -> PSQLPyResult<()> {
         match self {
             PsqlpyTransaction::PoolTrans(p_txid) => Ok(p_txid.commit().await?),
-            PsqlpyTransaction::SingleConnTrans(s_txid) => Ok(s_txid.commit().await?)
+            PsqlpyTransaction::SingleConnTrans(s_txid) => Ok(s_txid.commit().await?),
         }
     }
 
     async fn rollback(self) -> PSQLPyResult<()> {
         match self {
             PsqlpyTransaction::PoolTrans(p_txid) => Ok(p_txid.rollback().await?),
-            PsqlpyTransaction::SingleConnTrans(s_txid) => Ok(s_txid.rollback().await?)
+            PsqlpyTransaction::SingleConnTrans(s_txid) => Ok(s_txid.rollback().await?),
         }
     }
 
@@ -29,7 +29,7 @@ impl PsqlpyTransaction {
             PsqlpyTransaction::PoolTrans(p_txid) => {
                 p_txid.savepoint(sp_name).await?;
                 Ok(())
-            },
+            }
             PsqlpyTransaction::SingleConnTrans(s_txid) => {
                 s_txid.savepoint(sp_name).await?;
                 Ok(())
@@ -40,11 +40,15 @@ impl PsqlpyTransaction {
     async fn release_savepoint(&self, sp_name: &str) -> PSQLPyResult<()> {
         match self {
             PsqlpyTransaction::PoolTrans(p_txid) => {
-                p_txid.batch_execute(format!("RELEASE SAVEPOINT {sp_name}").as_str()).await?;
+                p_txid
+                    .batch_execute(format!("RELEASE SAVEPOINT {sp_name}").as_str())
+                    .await?;
                 Ok(())
-            },
+            }
             PsqlpyTransaction::SingleConnTrans(s_txid) => {
-                s_txid.batch_execute(format!("RELEASE SAVEPOINT {sp_name}").as_str()).await?;
+                s_txid
+                    .batch_execute(format!("RELEASE SAVEPOINT {sp_name}").as_str())
+                    .await?;
                 Ok(())
             }
         }
@@ -53,31 +57,38 @@ impl PsqlpyTransaction {
     async fn rollback_savepoint(&self, sp_name: &str) -> PSQLPyResult<()> {
         match self {
             PsqlpyTransaction::PoolTrans(p_txid) => {
-                p_txid.batch_execute(format!("ROLLBACK TO SAVEPOINT {sp_name}").as_str()).await?;
+                p_txid
+                    .batch_execute(format!("ROLLBACK TO SAVEPOINT {sp_name}").as_str())
+                    .await?;
                 Ok(())
-            },
+            }
             PsqlpyTransaction::SingleConnTrans(s_txid) => {
-                s_txid.batch_execute(format!("ROLLBACK TO SAVEPOINT {sp_name}").as_str()).await?;
+                s_txid
+                    .batch_execute(format!("ROLLBACK TO SAVEPOINT {sp_name}").as_str())
+                    .await?;
                 Ok(())
             }
         }
     }
 
     async fn bind<T>(&self, statement: &T, params: &[&(dyn ToSql + Sync)]) -> PSQLPyResult<Portal>
     where
-        T: ?Sized + ToStatement {
+        T: ?Sized + ToStatement,
+    {
         match self {
             PsqlpyTransaction::PoolTrans(p_txid) => Ok(p_txid.bind(statement, params).await?),
-            PsqlpyTransaction::SingleConnTrans(s_txid) => Ok(s_txid.bind(statement, params).await?)
+            PsqlpyTransaction::SingleConnTrans(s_txid) => {
+                Ok(s_txid.bind(statement, params).await?)
+            }
         }
     }
 
     pub async fn query_portal(&self, portal: &Portal, size: i32) -> PSQLPyResult<Vec<Row>> {
         match self {
-            PsqlpyTransaction::PoolTrans(p_txid)
-                => Ok(p_txid.query_portal(portal, size).await?),
-            PsqlpyTransaction::SingleConnTrans(s_txid)
-                => Ok(s_txid.query_portal(portal, size).await?)
+            PsqlpyTransaction::PoolTrans(p_txid) => Ok(p_txid.query_portal(portal, size).await?),
+            PsqlpyTransaction::SingleConnTrans(s_txid) => {
+                Ok(s_txid.query_portal(portal, size).await?)
+            }
         }
     }
 }

src/driver/mod.rs

@@ -4,7 +4,9 @@ pub mod connection_pool;
 pub mod connection_pool_builder;
 pub mod cursor;
 pub mod inner_connection;
+pub mod inner_transaction;
 pub mod listener;
+pub mod portal;
 pub mod transaction;
 pub mod transaction_options;
 pub mod utils;

src/driver/portal.rs

@@ -0,0 +1,52 @@
+use std::sync::Arc;
+
+use pyo3::{pyclass, pymethods};
+use tokio_postgres::Portal as tp_Portal;
+
+use crate::{exceptions::rust_errors::PSQLPyResult, query_result::PSQLDriverPyQueryResult};
+
+use super::inner_transaction::PsqlpyTransaction;
+
+#[pyclass]
+struct Portal {
+    transaction: Arc<PsqlpyTransaction>,
+    inner: tp_Portal,
+    array_size: i32,
+}
+
+impl Portal {
+    async fn query_portal(&self, size: i32) -> PSQLPyResult<PSQLDriverPyQueryResult> {
+        let result = self.transaction.query_portal(&self.inner, size).await?;
+        Ok(PSQLDriverPyQueryResult::new(result))
+    }
+}
+
+#[pymethods]
+impl Portal {
+    #[getter]
+    fn get_array_size(&self) -> i32 {
+        self.array_size
+    }
+
+    #[setter]
+    fn set_array_size(&mut self, value: i32) {
+        self.array_size = value;
+    }
+
+    async fn fetch_one(&self) -> PSQLPyResult<PSQLDriverPyQueryResult> {
+        self.query_portal(1).await
+    }
+
+    #[pyo3(signature = (size=None))]
+    async fn fetch_many(&self, size: Option<i32>) -> PSQLPyResult<PSQLDriverPyQueryResult> {
+        self.query_portal(size.unwrap_or(self.array_size)).await
+    }
+
+    async fn fetch_all(&self) -> PSQLPyResult<PSQLDriverPyQueryResult> {
+        self.query_portal(-1).await
+    }
+
+    async fn close(&mut self) {
+        let _ = Arc::downgrade(&self.transaction);
+    }
+}