src.lib.postgresql_base

Base class for PostgreSQL repository adapters.

 1"""Base class for PostgreSQL repository adapters."""
 2
 3from typing import LiteralString
 4
 5from psycopg.abc import Params
 6from psycopg_pool import ConnectionPool
 7
 8
 9class PostgreSQLRepositoryBase:
10    """Base class providing PostgreSQL connection pooling and query helpers."""
11
12    _pool: ConnectionPool
13    _connection_string: str
14
15    def __init__(self, connection_string: str) -> None:
16        """Initialise with a PostgreSQL connection string.
17
18        Args:
19            connection_string: PostgreSQL connection string.
20        """
21        self._connection_string = connection_string
22        self._pool = ConnectionPool(
23            conninfo=connection_string,
24            open=True,
25            min_size=1,
26            max_size=10,
27        )
28
29    def execute(self, sql: LiteralString, params: Params = ()) -> None:
30        """Execute a SQL statement with commit.
31
32        Args:
33            sql: The SQL statement to execute.
34            params: Parameters for the SQL statement.
35        """
36        with self._pool.connection() as conn:
37            with conn.cursor() as cursor:
38                _ = cursor.execute(sql, params)
39                conn.commit()
40
41    def conn(self):
42        """Execute a SQL statement and return a single row.
43
44        Args:
45            sql: The SQL statement to execute.
46            params: Parameters for the SQL statement.
47
48        Returns:
49            A single row as a tuple, or None if no results.
50        """
51        return self._pool.connection()
52
53    def close_db(self) -> None:
54        """Close the database connection pool."""
55        self._pool.close()
class PostgreSQLRepositoryBase: 
10class PostgreSQLRepositoryBase:
11    """Base class providing PostgreSQL connection pooling and query helpers."""
12
13    _pool: ConnectionPool
14    _connection_string: str
15
16    def __init__(self, connection_string: str) -> None:
17        """Initialise with a PostgreSQL connection string.
18
19        Args:
20            connection_string: PostgreSQL connection string.
21        """
22        self._connection_string = connection_string
23        self._pool = ConnectionPool(
24            conninfo=connection_string,
25            open=True,
26            min_size=1,
27            max_size=10,
28        )
29
30    def execute(self, sql: LiteralString, params: Params = ()) -> None:
31        """Execute a SQL statement with commit.
32
33        Args:
34            sql: The SQL statement to execute.
35            params: Parameters for the SQL statement.
36        """
37        with self._pool.connection() as conn:
38            with conn.cursor() as cursor:
39                _ = cursor.execute(sql, params)
40                conn.commit()
41
42    def conn(self):
43        """Execute a SQL statement and return a single row.
44
45        Args:
46            sql: The SQL statement to execute.
47            params: Parameters for the SQL statement.
48
49        Returns:
50            A single row as a tuple, or None if no results.
51        """
52        return self._pool.connection()
53
54    def close_db(self) -> None:
55        """Close the database connection pool."""
56        self._pool.close()

Base class providing PostgreSQL connection pooling and query helpers.

PostgreSQLRepositoryBase(connection_string: str) 
16    def __init__(self, connection_string: str) -> None:
17        """Initialise with a PostgreSQL connection string.
18
19        Args:
20            connection_string: PostgreSQL connection string.
21        """
22        self._connection_string = connection_string
23        self._pool = ConnectionPool(
24            conninfo=connection_string,
25            open=True,
26            min_size=1,
27            max_size=10,
28        )

Initialise with a PostgreSQL connection string.

Args: connection_string: PostgreSQL connection string.

def execute( self, sql: LiteralString, params: Sequence[Any] | Mapping[str, Any] = ()) -> None: 
30    def execute(self, sql: LiteralString, params: Params = ()) -> None:
31        """Execute a SQL statement with commit.
32
33        Args:
34            sql: The SQL statement to execute.
35            params: Parameters for the SQL statement.
36        """
37        with self._pool.connection() as conn:
38            with conn.cursor() as cursor:
39                _ = cursor.execute(sql, params)
40                conn.commit()

Execute a SQL statement with commit.

Args: sql: The SQL statement to execute. params: Parameters for the SQL statement.

def conn(self): 
42    def conn(self):
43        """Execute a SQL statement and return a single row.
44
45        Args:
46            sql: The SQL statement to execute.
47            params: Parameters for the SQL statement.
48
49        Returns:
50            A single row as a tuple, or None if no results.
51        """
52        return self._pool.connection()

Execute a SQL statement and return a single row.

Args: sql: The SQL statement to execute. params: Parameters for the SQL statement.

Returns: A single row as a tuple, or None if no results.

def close_db(self) -> None: 
54    def close_db(self) -> None:
55        """Close the database connection pool."""
56        self._pool.close()

Close the database connection pool.