sqlalchemy_async_session_factory.engine¶

Async engine factory with connection pooling.

Extracted from GridFlow backend/src/database.py

Functions¶

create_async_engine_with_pool(...)

Create async SQLAlchemy engine with connection pooling.

Module Contents¶

sqlalchemy_async_session_factory.engine.create_async_engine_with_pool(database_url: str, echo: bool = False, pool_size: int = 5, max_overflow: int = 5, pool_timeout: int = 30, pool_pre_ping: bool = True, pool_recycle: int = 300, **kwargs) → sqlalchemy.ext.asyncio.AsyncEngine¶

Create async SQLAlchemy engine with connection pooling.

Optimized for PostgreSQL with asyncpg driver. Also supports SQLite with aiosqlite driver (uses StaticPool, ignores pool params).

Args:

database_url: Database connection URL echo: Echo SQL queries (default: False) pool_size: Connections in pool (ignored for SQLite) max_overflow: Max overflow connections (ignored for SQLite) pool_timeout: Connection timeout seconds (ignored for SQLite) pool_pre_ping: Test connections before use (default: True) pool_recycle: Recycle connections after seconds (default: 300) **kwargs: Additional engine configuration

Returns:

Configured AsyncEngine instance

Example:

>>> engine = create_async_engine_with_pool(
...     "postgresql+asyncpg://user:pass@localhost/db",
...     pool_size=10,
... )