Merge pull request #33 from qaspen-python/feature/added_method_to_connect

Added connect method to create ConnectionPool

Author: chandr-andr

python/psqlpy/__init__.py

@@ -8,6 +8,7 @@
     ReadVariant,
     SingleQueryResult,
     Transaction,
+    connect,
 )
 
 __all__ = [
@@ -20,4 +21,5 @@
     "ConnRecyclingMethod",
     "IsolationLevel",
     "ReadVariant",
+    "connect",
 ]

python/psqlpy/_internal/__init__.pyi

@@ -936,3 +936,27 @@ class ConnectionPool:
         """
     def close(self: Self) -> None:
         """Close the connection pool."""
+
+def connect(
+    dsn: Optional[str] = None,
+    username: Optional[str] = None,
+    password: Optional[str] = None,
+    host: Optional[str] = None,
+    port: Optional[int] = None,
+    db_name: Optional[str] = None,
+    max_db_pool_size: int = 2,
+    conn_recycling_method: Optional[ConnRecyclingMethod] = None,
+) -> ConnectionPool:
+    """Create new connection pool.
+
+    ### Parameters:
+    - `dsn`: full dsn connection string.
+        `postgres://postgres:postgres@localhost:5432/postgres?target_session_attrs=read-write`
+    - `username`: username of the user in postgres
+    - `password`: password of the user in postgres
+    - `host`: host of postgres
+    - `port`: port of postgres
+    - `db_name`: name of the database in postgres
+    - `max_db_pool_size`: maximum size of the connection pool
+    - `conn_recycling_method`: how a connection is recycled.
+    """

python/tests/test_connection_pool.py

@@ -1,11 +1,20 @@
 import pytest
 
-from psqlpy import Connection, ConnectionPool, ConnRecyclingMethod, QueryResult
+from psqlpy import Connection, ConnectionPool, ConnRecyclingMethod, QueryResult, connect
 from psqlpy.exceptions import RustPSQLDriverPyBaseError
 
 pytestmark = pytest.mark.anyio
 
 
+async def test_connect_func() -> None:
+    """Test that connect function makes new connection pool."""
+    pg_pool = connect(
+        dsn="postgres://postgres:postgres@localhost:5432/psqlpy_test",
+    )
+
+    await pg_pool.execute("SELECT 1")
+
+
 async def test_pool_dsn_startup() -> None:
     """Test that connection pool can startup with dsn."""
     pg_pool = ConnectionPool(