Default Pool

amsdal_glue.DefaultConnectionPool ¶

Bases: ConnectionPoolBase

DefaultConnectionPool manages a pool of connections, ensuring that connections are reused and not exceeded. It extends the ConnectionPoolBase class.

Attributes:

Name	Type	Description
`connections`	`dict[str \| None, tuple[ConnectionBase, datetime]]`	A dictionary of connections with their last used time.
`_max_connections`	`int`	The maximum number of connections allowed.
`_expiration_time`	`int`	The time in seconds after which a connection is considered expired.

Example

This example demonstrates how to create SQLite-base connection pool with a maximum of 10 connections, supported by SqliteConnection connection:

from amsdal_glue import DefaultConnectionPool
from amsdal_glue import SqliteConnection

sqlite_connection_pool = DefaultConnectionPool(
    SqliteConnection,
    db_path='my_db.sqlite',
    max_connections=10,
)

Here is an example of how to create a Postgres-based connection with some parameters supported by PostgresConnection connection:

from amsdal_glue import DefaultConnectionPool
from amsdal_glue import PostgresConnection

pg_connection_pool = DefaultConnectionPool(
    PostgresConnection,
    dsn='postgres://db_user:db_password@localhost:5433/db_name',
    schema='public',
    timezone='UTC',
)

get_connection ¶

get_connection(transaction_id=None)

Retrieves a connection for the given transaction ID or creates a new one if none are available.

Parameters:

Name	Type	Description	Default
`transaction_id`	`str \| None`	The ID of the transaction.	`None`

Returns:

Name	Type	Description
`ConnectionBase`	`ConnectionBase`	A connection for the given transaction ID.

disconnect_connection ¶

disconnect_connection(transaction_id=None)

Disconnects the connection for the given transaction ID.

Parameters:

Name	Type	Description	Default
`transaction_id`	`str \| None`	The ID of the transaction.	`None`

disconnect ¶

disconnect()

Disconnects all connections in the pool.