diff --git a/README.md b/README.md index 0a94c43..919dd90 100644 --- a/README.md +++ b/README.md @@ -29,7 +29,7 @@ - **Observability** — Prometheus connection-pool metrics and OpenTelemetry tracing - **DI integration** — Ready-to-use providers for [`dishka`](https://github.com/reagento/dishka) and `dependency-injector` -Only `sqlalchemy[asyncio]` and `pydantic` are required by default — everything else is an opt-in extra. +Only `sqlalchemy[asyncio]`, `pydantic` and `asyncpg` are required by default — everything else is an opt-in extra. > [!TIP] > **Building this with an AI assistant?** Hand it @@ -183,12 +183,14 @@ async def main(): user = UserDB(id=uuid4(), email="user@example.com", username="user") session.add(user) # Auto-commit on exit + + await session_manager.aclose() ``` ### 4. Unit of Work Pattern ```python -from unit_of_work_kit import AsyncSQLAlchemyUnitOfWork, AsyncSQLAlchemyUowTransaction +from sqlalchemy_foundation_kit import AsyncSQLAlchemyUnitOfWork, AsyncSQLAlchemyUowTransaction # Define your transaction with repositories class MyTransaction(AsyncSQLAlchemyUowTransaction): diff --git a/docs/agents.md b/docs/agents.md index 674e631..fc3097c 100644 --- a/docs/agents.md +++ b/docs/agents.md @@ -7,11 +7,11 @@ | | | |---|---| | Package | `sqlalchemy-foundation-kit` on PyPI, import root `sqlalchemy_foundation_kit` | -| Requires | Python 3.11+, SQLAlchemy 2 (`>=2.0.35,<3`), Pydantic 2 (`>=2.5,<3`), PostgreSQL, and `asyncpg` — the import fails without it, see rule 1 | -| Install | `pip install sqlalchemy-foundation-kit asyncpg` · extras: `settings`, `metrics`, `orjson`, `dishka`, `dependency-injector`, `telemetry`, `all` | +| Requires | Python 3.11+, SQLAlchemy 2 (`>=2.0.35,<3`), Pydantic 2 (`>=2.5,<3`), asyncpg (`>=0.30,<1`), PostgreSQL | +| Install | `pip install sqlalchemy-foundation-kit` · extras: `settings`, `metrics`, `orjson`, `dishka`, `dependency-injector`, `telemetry`, `all` | | Async | the whole library. `AsyncEngine`, `AsyncSession`, `asyncpg` | | Sync | none. There is no sync mirror and no sync entry point | -| Version | 0.2.0 — everything below was read from the source at that version | +| Version | everything below was read from the source this site was built from. Three calls described here do not work on 0.2.0 — see [fixed since 0.2.0](#fixed-since-020) | | Source | | ## How to read this page @@ -26,11 +26,11 @@ API, so it carries neither the control nor a `.md` twin — read it as HTML, or docstrings in the source. Top to bottom before writing code. [Rules that hold or break the code](#rules-that-hold-or-break-the-code) -is the section correctness lives in, and it is followed by the short list of things that -are [broken at 0.2.0](#broken-at-020) — call one of those and the process raises, not the -database. Every name used below is in the public API; if you need something not listed -here, fetch the page the [documentation map](#documentation-map) points at rather than -guessing a method that sounds plausible. +is the section correctness lives in, and it is followed by the short list of calls that +are [fixed since 0.2.0](#fixed-since-020) — if the installed version is 0.2.0, those +raise before they reach the database. Every name used below is in the public API; if you +need something not listed here, fetch the page the [documentation map](#documentation-map) +points at rather than guessing a method that sounds plausible. ## Scope @@ -170,7 +170,7 @@ use_orjson=False, metrics=None, on_engine_created=None, dispose_timeout=30.0, ** | `.engine` | the `AsyncEngine`, created in `__init__` | | `.session_maker` | the `async_sessionmaker`; this is what the unit of work wants | | `.get_session()` | async context manager yielding a session with no transaction started | -| `.get_transaction(isolation_level=None)` | **raises `TypeError` at 0.2.0** — see [broken at 0.2.0](#broken-at-020) | +| `.get_transaction(isolation_level=None)` | async context manager yielding a session with a transaction open: commits on clean exit, rolls back on exception. An `isolation_level` is set on that transaction's connection and nothing else | | `await .aclose()` | disposes the engine under `asyncio.shield`, capped at `dispose_timeout`; idempotent, logs a warning on timeout | | `async with manager:` | the same `aclose()` on exit | @@ -215,8 +215,10 @@ never propagated. an RLS context or a statement timeout on every session, calling `super().open_session(...)` inside. `flush_before_commit=None` on `transaction()` falls back to the constructor value (`True`), which flushes before the commit so an integrity error surfaces inside the block -rather than at exit. Every `isolation_level` argument in this table is broken at 0.2.0 — -see below. +rather than at exit. An `isolation_level` argument takes an `IsolationLevel` or a string +in either spelling, and is set on the connection the block checks out, so it covers this +transaction and leaves the engine alone. Setting it has to happen before the transaction +starts, so it opens the transaction as the block is entered — `query()` included. `TracedAsyncUnitOfWork(session_maker, transaction_factory, service_name="sqlalchemy-foundation-kit", *, flush_before_commit=True)` from @@ -239,8 +241,10 @@ typing a transaction structurally in a use case that should not import the concr `await try_advisory_xact_lock(session, key)` is the same lock as a free function: `pg_try_advisory_xact_lock`, non-blocking, returns `True` if taken, released at the end of -the transaction. `key` may be `str | int`; the integer is wrapped into signed 64-bit. -Read rule 8 before passing a string. +the transaction. `key` may be `str | int`; an integer is wrapped into signed 64-bit, a +string is hashed into it with BLAKE2b, reproducibly — the same string is the same lock in +every process. The `SupportsAdvisoryLock` protocol still types `key` as `int`; the mixin +and the free function take `str | int`. ### Base ORM @@ -328,10 +332,11 @@ decorator. ## Rules that hold or break the code -1. **`asyncpg` must be installed even though nothing declares it.** `AsyncCConnection` - imports it at module import time and the package `__init__` imports that, so on a clean - `pip install sqlalchemy-foundation-kit` the very first `import sqlalchemy_foundation_kit` - raises `ModuleNotFoundError: No module named 'asyncpg'`. Install `asyncpg` alongside it. +1. **`asyncpg` is a hard dependency, not an extra.** `AsyncCConnection` subclasses + `asyncpg.Connection` at module import time and the package `__init__` imports it, so + the package cannot be imported without asyncpg. `pip install sqlalchemy-foundation-kit` + brings it. On 0.2.0 it did not, and the first `import sqlalchemy_foundation_kit` raised + `ModuleNotFoundError: No module named 'asyncpg'`. 2. **PostgreSQL over asyncpg only.** The DSN is `postgresql+asyncpg://…`. `create_async_session_manager` passes an asyncpg-specific `connection_class` and asyncpg-specific `connect_args`; another driver rejects them. @@ -354,13 +359,14 @@ decorator. session is closed and its connection is back in the pool. Return domain objects or detached data, never a live `tx`. Storing one on `self`, in a module global, or in a `ContextVar` that outlives the block is the same bug. -8. **A string advisory-lock key does not lock across processes.** `try_advisory_xact_lock` - turns a `str` into an integer with Python's `hash()`, which is salted per process: - three fresh interpreters produced three different keys for `"job"`. Two replicas of the - same service therefore take *different* locks and both proceed. Pass a stable integer - you computed yourself (`zlib.crc32(b"job")`, a hash digest, a constant) for anything - that has to be exclusive beyond one process. The protocol and the mixin type `key` as - `int` for this reason; only the free function accepts `str`. +8. **A string advisory-lock key is stable across processes; on 0.2.0 it was not.** + `try_advisory_xact_lock` hashes a `str` with BLAKE2b, so every replica turns the same + string into the same lock. On 0.2.0 it used Python's `hash()`, which is salted per + process: three fresh interpreters produced three different keys for `"job"`, so two + replicas took *different* locks and both proceeded. On that version pass a stable + integer you computed yourself (`zlib.crc32(b"job")`, a digest, a constant) instead. + Either way the key changes with the hash: a rolling deploy across the fix has old and + new replicas holding different locks for the same name until it finishes. 9. **An advisory lock is only held while its transaction is.** `pg_try_advisory_xact_lock` releases at transaction end, so take it inside `transaction()` or `managed_session()` and do the protected work in the same block. Taking one in `query()` is a no-op with a @@ -380,29 +386,28 @@ decorator. `contrib.metrics` needs `[metrics]`, `contrib.telemetry` needs `[telemetry]`, `contrib.di` needs `[dishka]`, `contrib.dependency_injector` needs `[dependency-injector]`, and `use_orjson=True` needs `[orjson]`. The DI packages fail - on *import* without their extra, with a bare `AttributeError` rather than the intended - message — see below. + on *import* without their extra, with an `ImportError` naming the extra to install. 14. **Metrics never raise into your code.** Every recorder call is wrapped: a broken metrics backend is logged at exception level and discarded, and the query proceeds. 15. **Close the manager.** `await manager.aclose()` disposes the engine under a shield and a `dispose_timeout` (30s). Skipping it leaks connections; calling it twice is fine. -### Broken at 0.2.0 +### Fixed since 0.2.0 -Three published entry points raise before they reach the database. All three were run -against PostgreSQL 17 to confirm it. +Three published entry points raise before they reach the database on 0.2.0. They work on +current versions; the workaround column is what to do if the installed version is 0.2.0. -| Call | What happens | Use instead | +| Call | What 0.2.0 does | Workaround on 0.2.0 | |---|---|---| -| `manager.get_transaction()`, with or without `isolation_level` | `TypeError: Session.__init__() got an unexpected keyword argument 'execution_options'` — the argument is passed to the session factory unconditionally | `async with manager.get_session() as s, s.begin():`, or the unit of work | -| `uow.transaction(isolation_level=…)`, `uow.managed_session(isolation_level=…)`, `uow.query(isolation_level=…)` | `InvalidRequestError: This connection has already initialized a SQLAlchemy Transaction()… isolation_level may not be altered` — the level is applied after the connection has autobegun | set the level on the engine: `AsyncSessionManager(..., isolation_level="SERIALIZABLE")` or `QuerySettings(isolation_level=...)` | +| `manager.get_transaction()`, with or without `isolation_level` | `TypeError: Session.__init__() got an unexpected keyword argument 'execution_options'` — the argument was passed to the session factory unconditionally | `async with manager.get_session() as s, s.begin():`, or the unit of work | +| `uow.transaction(isolation_level=…)`, `uow.managed_session(isolation_level=…)`, `uow.query(isolation_level=…)` | `InvalidRequestError: This connection has already initialized a SQLAlchemy Transaction()… isolation_level may not be altered` — the level was applied after the connection had autobegun | set the level on the engine: `AsyncSessionManager(..., isolation_level="SERIALIZABLE")` or `QuerySettings(isolation_level=...)` | | `import sqlalchemy_foundation_kit.contrib.di` (or `.contrib.dependency_injector`) without its extra | `AttributeError: 'NoneType' object has no attribute 'APP'` (resp. `'DeclarativeContainer'`) instead of the intended `ImportError` | install the extra; the message is not the one the code meant to give you | -`IsolationLevel` itself is fine — `READ_UNCOMMITTED`, `READ_COMMITTED`, `REPEATABLE_READ`, -`SERIALIZABLE`, whose values are the PostgreSQL spellings with spaces — and so is -`normalize_isolation_level` in `sqlalchemy_foundation_kit.uow.sqlalchemy`, which accepts -either spelling in any case. It is only the plumbing that carries the value to a session -that is wrong. +`IsolationLevel` itself was always fine — `READ_UNCOMMITTED`, `READ_COMMITTED`, +`REPEATABLE_READ`, `SERIALIZABLE`, whose values are the PostgreSQL spellings with spaces — +and so is `normalize_isolation_level` in `sqlalchemy_foundation_kit.uow.sqlalchemy`, which +accepts either spelling in any case. It was only the plumbing carrying the value to a +session that was wrong. ## Common mistakes @@ -460,17 +465,16 @@ async def get_user(uow, user_id) -> User | None: ``` ```python -# WRONG — a string key hashes differently in every process, so nothing is excluded -async with uow.transaction() as tx: - if await tx.try_advisory_lock("nightly-rollup"): +# WRONG — the lock is released at transaction end, so this protects nothing +async with uow.query() as qx: + if await qx.try_advisory_lock("nightly-rollup"): ... +await do_the_rollup() # outside the block: the lock is already gone -# RIGHT — a key both replicas compute the same way -LOCK_NIGHTLY_ROLLUP = 0x6E52 # any fixed int; zlib.crc32(b"nightly-rollup") works too - +# RIGHT — take the lock and do the work in the same transaction async with uow.transaction() as tx: - if await tx.try_advisory_lock(LOCK_NIGHTLY_ROLLUP): - ... + if await tx.try_advisory_lock("nightly-rollup"): + await do_the_rollup(tx) ``` ```python @@ -501,14 +505,14 @@ SQLAlchemy's through untouched. | Raised | When | |---|---| -| `ModuleNotFoundError` | `asyncpg` is not installed (rule 1) | -| `ImportError` | an extra is missing: `orjson`, `pydantic-settings`, `prometheus-client`, the OpenTelemetry instrumentations, `dishka`, `dependency-injector`. The orjson message names a `[json]` extra that does not exist — the real one is `[orjson]` | +| `ModuleNotFoundError` | `asyncpg` is not installed — only reachable on 0.2.0, which did not declare it (rule 1) | +| `ImportError` | an extra is missing: `orjson`, `pydantic-settings`, `prometheus-client`, the OpenTelemetry instrumentations, `dishka`, `dependency-injector`. The message names the extra to install | | `RuntimeError` | `"AsyncSessionManager is closed"` — a session asked for after `aclose()` | | `ValueError` | an isolation level `normalize_isolation_level` does not know; an unregistered pool name; a pool name registered twice without `override=True`; `max_retries < 1`; a metric prefix that is not a Prometheus identifier | | `pydantic.ValidationError` | a `contrib.settings` model built without `password`, `database` or `application_name`, or a `static` pool with `max_overflow > 0` | -| `TypeError` | `manager.get_transaction()` (see above); a value orjson cannot serialize | +| `TypeError` | a value orjson cannot serialize; `manager.get_transaction()` on 0.2.0 (see above) | | `sqlalchemy.exc.IllegalStateChangeError` | one session driven by two tasks at once (rule 6) | -| `sqlalchemy.exc.InvalidRequestError` | `isolation_level` on a unit-of-work method (see above) | +| `sqlalchemy.exc.InvalidRequestError` | `isolation_level` on a unit-of-work method on 0.2.0 (see above) | | `sqlalchemy.exc.IntegrityError`, `OperationalError`, … | the database refused the statement. Inside `tx.savepoint()` these are re-raised with the surrounding transaction still usable; anywhere else they roll the whole block back | ## Documentation map diff --git a/docs/guide/advanced.md b/docs/guide/advanced.md index 2c62a39..57b2d4e 100644 --- a/docs/guide/advanced.md +++ b/docs/guide/advanced.md @@ -86,6 +86,13 @@ async with uow.transaction(isolation_level="READ COMMITTED") as tx: - `REPEATABLE_READ` — Prevents non-repeatable reads, snapshot isolation - `SERIALIZABLE` — Strongest guarantees, may have serialization failures +The level is set on the connection this block checks out, so it applies to this +transaction and leaves the engine alone. It has to be set before the transaction starts, +which means the block's transaction opens as the context manager is entered — including +in `query()`, which otherwise waits for the first statement. To set a level for every +session instead, pass `isolation_level=` to `AsyncSessionManager` or set +`QuerySettings.isolation_level`. + ### Read-Only Queries For read-only operations without transaction overhead: @@ -199,8 +206,11 @@ async with session_manager.get_transaction() as session: **Lock Keys:** -- **String keys** — Automatically hashed to integers: `"process_payments"`, `"user:123"` -- **Integer keys** — Used directly: `123456`, `user.id` +- **String keys** — Hashed to integers with BLAKE2b: `"process_payments"`, `"user:123"`. + The hash is reproducible, so every replica of a service turns the same string into the + same lock. The key changes between library versions only if this page says so. +- **Integer keys** — Used directly: `123456`, `user.id`. Values outside PostgreSQL's + `bigint` range are wrapped into it. **Lock Types:** @@ -243,12 +253,9 @@ pip install sqlalchemy-foundation-kit[metrics] ```python from sqlalchemy_foundation_kit.contrib.metrics import PostgresMetrics -# Create metrics collector -metrics = PostgresMetrics( - prefix="myapp", - service_name="identity-service", - labels={"environment": "production"}, -) +# Create metrics collector. `prefix` is the only argument; it is prepended with an +# underscore and must match ^[a-zA-Z_][a-zA-Z0-9_]*$. +metrics = PostgresMetrics(prefix="myapp") # Pass to session manager session_manager = create_async_session_manager( @@ -261,12 +268,14 @@ session_manager = create_async_session_manager( | Metric | Type | Description | |--------|------|-------------| -| `myapp_postgres_pool_size` | Gauge | Current pool size | -| `myapp_postgres_pool_checked_out` | Gauge | Connections currently in use | -| `myapp_postgres_pool_overflow` | Gauge | Overflow connections created | -| `myapp_postgres_checkout_duration_seconds` | Histogram | Time to acquire connection | -| `myapp_postgres_errors_total` | Counter | Database errors by type | -| `myapp_postgres_timeouts_total` | Counter | Connection timeout errors | +| `myapp_postgres_db_pool_size` | Gauge | Current pool size | +| `myapp_postgres_db_pool_checked_out` | Gauge | Connections currently in use | +| `myapp_postgres_db_pool_overflow` | Gauge | Overflow connections created | +| `myapp_postgres_db_connection_checkout_duration_seconds` | Histogram | Time to acquire connection | +| `myapp_postgres_db_connection_errors_total` | Counter | Database errors by type (`error_type` label) | +| `myapp_postgres_db_connection_timeouts_total` | Counter | Connection timeout errors | + +Without a `prefix` the names are `postgres_db_pool_size` and so on. **Expose metrics endpoint:** @@ -505,31 +514,46 @@ await container.shutdown_resources() ### Health Checks -```python -# Check database connectivity -is_healthy = await session_manager.healthcheck() +There is no `healthcheck()` method — the library ships the query and leaves the policy to +you: -if not is_healthy: - logger.error("Database health check failed") +```python +from sqlalchemy import text +from sqlalchemy_foundation_kit import DEFAULT_HEALTHCHECK_QUERY -# With custom query -is_healthy = await session_manager.healthcheck( - query="SELECT 1 FROM users LIMIT 1" -) +async def is_healthy(session_manager: AsyncSessionManager) -> bool: + """Check database connectivity.""" + try: + async with session_manager.get_session() as session: + await session.execute(text(DEFAULT_HEALTHCHECK_QUERY)) + except Exception: + logger.exception("Database health check failed") + return False + return True ``` +The DI providers run exactly this at startup — see `AsyncDatabaseProvider` and +`DatabaseContainer`, both of which take `healthcheck_query=None` to skip it. + ### Graceful Shutdown +`aclose()` disposes the engine under `asyncio.shield`, capped by the manager's +`dispose_timeout` (30 seconds by default). It is idempotent, and logs a warning rather +than raising if the timeout expires. + ```python import signal import asyncio +# The timeout belongs to the manager, not to the call that closes it +session_manager = create_async_session_manager(settings.postgres, dispose_timeout=30.0) + async def shutdown(session_manager: AsyncSessionManager): """Graceful shutdown handler.""" logger.info("Shutting down database connections...") # Wait for in-flight requests to complete - await session_manager.close(timeout=30.0) + await session_manager.aclose() logger.info("Database connections closed") @@ -548,34 +572,37 @@ await run_app() ### Connection Retry -Automatically retry on transient connection errors: +`retry_async_connection` is a coroutine function, not a decorator, and it retries a +callable that establishes or tests a connection — the startup wait, not every query. It +re-raises the last exception when the attempts run out: ```python +from sqlalchemy import text from sqlalchemy_foundation_kit import ( - retry_async_connection, + DEFAULT_HEALTHCHECK_QUERY, RetryConfig, + retry_async_connection, ) -# Custom retry config +# Custom retry config: attempt N sleeps retry_delay * 2 ** N, capped at max_backoff_delay retry_config = RetryConfig( - max_attempts=5, - initial_delay=1.0, - max_delay=30.0, - exponential_base=2.0, - jitter=True, + max_retries=5, + retry_delay=1.0, + max_backoff_delay=30.0, ) -@retry_async_connection(config=retry_config) -async def fetch_user(session, user_id: UUID): - """Retries on connection errors.""" - result = await session.execute( - select(UserDB).where(UserDB.id == user_id) +async def wait_for_database(session_manager: AsyncSessionManager) -> None: + async def connect() -> None: + async with session_manager.get_session() as session: + await session.execute(text(DEFAULT_HEALTHCHECK_QUERY)) + + await retry_async_connection( + connect_func=connect, + service_name="PostgreSQL", + config=retry_config, ) - return result.scalar_one_or_none() -# Use in transaction -async with session_manager.get_transaction() as session: - user = await fetch_user(session, user_id) +await wait_for_database(session_manager) ``` ### pgbouncer Compatibility diff --git a/docs/guide/configuration.md b/docs/guide/configuration.md index fbdc0dd..9e018b9 100644 --- a/docs/guide/configuration.md +++ b/docs/guide/configuration.md @@ -63,36 +63,54 @@ settings = Settings() ### Environment Variables -`BasePostgresConfig` inherits from `pydantic_settings.BaseSettings`, so it automatically loads from environment variables: +`BasePostgresConfig` declares no `model_config` of its own, so it reads no prefix and no +delimiter by itself. Environment variables reach it through the `BaseSettings` that holds +it, which is where `env_nested_delimiter` is set: + +```python +from pydantic_settings import BaseSettings, SettingsConfigDict + +class Settings(BaseSettings): + model_config = SettingsConfigDict(env_nested_delimiter="__") + + postgres: BasePostgresConfig +``` + +Every level of nesting is then one `__`, the field name included — `postgres` is a field +like any other, so `postgres.connection.host` is `POSTGRES__CONNECTION__HOST`: ```bash # Connection settings -POSTGRES_CONNECTION__HOST=db.example.com -POSTGRES_CONNECTION__PORT=5432 -POSTGRES_CONNECTION__USER=postgres -POSTGRES_CONNECTION__PASSWORD=secret123 -POSTGRES_CONNECTION__DATABASE=mydb +POSTGRES__CONNECTION__HOST=db.example.com +POSTGRES__CONNECTION__PORT=5432 +POSTGRES__CONNECTION__USER=postgres +POSTGRES__CONNECTION__PASSWORD=secret123 +POSTGRES__CONNECTION__DATABASE=mydb # Pool settings -POSTGRES_POOL__SIZE=20 -POSTGRES_POOL__MAX_OVERFLOW=30 -POSTGRES_POOL__TIMEOUT=45.0 -POSTGRES_POOL__PRE_PING=true -POSTGRES_POOL__RECYCLE=1800 +POSTGRES__POOL__SIZE=20 +POSTGRES__POOL__MAX_OVERFLOW=30 +POSTGRES__POOL__TIMEOUT=45.0 +POSTGRES__POOL__PRE_PING=true +POSTGRES__POOL__RECYCLE=1800 # Query settings -POSTGRES_QUERY__ECHO=false -POSTGRES_QUERY__STATEMENT_CACHE_SIZE=0 -POSTGRES_QUERY__ISOLATION_LEVEL="READ COMMITTED" +POSTGRES__QUERY__ECHO=false +POSTGRES__QUERY__STATEMENT_CACHE_SIZE=0 +POSTGRES__QUERY__ISOLATION_LEVEL="READ COMMITTED" # Top-level settings -POSTGRES_APPLICATION_NAME=my-service -POSTGRES_DB_SCHEMA=public -POSTGRES_USE_ORJSON_SERIALIZATION=true -POSTGRES_JIT=off -POSTGRES_METRICS_ENABLED=true +POSTGRES__APPLICATION_NAME=my-service +POSTGRES__DB_SCHEMA=public +POSTGRES__USE_ORJSON_SERIALIZATION=true +POSTGRES__JIT=off +POSTGRES__METRICS_ENABLED=true ``` +`BasePostgresMigrationsConfig` is exactly this settings class, ready made: it holds a +`postgres: BasePostgresConfig` with `env_nested_delimiter="__"` and `extra="ignore"` +already set, and reads the same names. + **Custom prefix:** ```python @@ -107,7 +125,7 @@ class Settings(BaseSettings): postgres: BasePostgresConfig ``` -Now use `MY_APP_POSTGRES_CONNECTION__HOST` instead of `POSTGRES_CONNECTION__HOST`. +Now use `MY_APP_POSTGRES__CONNECTION__HOST` instead of `POSTGRES__CONNECTION__HOST`. ### DSN Generation diff --git a/docs/guide/quickstart.md b/docs/guide/quickstart.md index ab42385..d54b824 100644 --- a/docs/guide/quickstart.md +++ b/docs/guide/quickstart.md @@ -145,7 +145,7 @@ async def main(): # Auto-rollback on exception # Graceful shutdown (wait for connections to close) - await session_manager.close() + await session_manager.aclose() ``` ### 4. Use Unit of Work Pattern @@ -360,7 +360,7 @@ async def main(): print(f"Error: {e}") # Cleanup - await session_manager.close() + await session_manager.aclose() if __name__ == "__main__": import asyncio @@ -377,22 +377,36 @@ if __name__ == "__main__": ### Health Check +The library ships the query (`DEFAULT_HEALTHCHECK_QUERY`), not a health-check method — +run it on a session: + ```python +from sqlalchemy import text +from sqlalchemy_foundation_kit import DEFAULT_HEALTHCHECK_QUERY + async def health_check(): """Check database connectivity.""" - is_healthy = await session_manager.healthcheck() - return {"database": "healthy" if is_healthy else "unhealthy"} + try: + async with session_manager.get_session() as session: + await session.execute(text(DEFAULT_HEALTHCHECK_QUERY)) + except Exception: + return {"database": "unhealthy"} + return {"database": "healthy"} ``` ### Graceful Shutdown +The disposal timeout belongs to the manager, not to the call that closes it: + ```python import signal +session_manager = create_async_session_manager(settings.postgres, dispose_timeout=30.0) + async def shutdown(session_manager: AsyncSessionManager): """Graceful shutdown handler.""" print("Shutting down...") - await session_manager.close(timeout=30.0) + await session_manager.aclose() print("Database connections closed") # Register signal handler @@ -405,16 +419,28 @@ loop.add_signal_handler( ### Retry on Connection Error +`retry_async_connection` is a coroutine function, not a decorator. It retries a callable +that establishes or tests a connection — typically once, at startup: + ```python -from sqlalchemy_foundation_kit import retry_async_connection, DEFAULT_RETRY_CONFIG +from sqlalchemy import text +from sqlalchemy_foundation_kit import ( + DEFAULT_HEALTHCHECK_QUERY, + RetryConfig, + retry_async_connection, +) + +async def wait_for_database(session_manager: AsyncSessionManager) -> None: + """Wait for PostgreSQL to accept connections, with exponential backoff.""" + async def connect() -> None: + async with session_manager.get_session() as session: + await session.execute(text(DEFAULT_HEALTHCHECK_QUERY)) -@retry_async_connection(config=DEFAULT_RETRY_CONFIG) -async def fetch_user(session, user_id: UUID): - """Retries on connection errors.""" - result = await session.execute( - select(UserDB).where(UserDB.id == user_id) + await retry_async_connection( + connect_func=connect, + service_name="PostgreSQL", + config=RetryConfig(max_retries=5, retry_delay=1.0, max_backoff_delay=30.0), ) - return result.scalar_one_or_none() ``` ### Custom JSON Type diff --git a/docs/index.md b/docs/index.md index f63a1d1..773f07e 100644 --- a/docs/index.md +++ b/docs/index.md @@ -11,7 +11,7 @@ - **Observability** — Prometheus connection-pool metrics and OpenTelemetry tracing - **DI integration** — Ready-to-use providers for [`dishka`](https://github.com/reagento/dishka) and `dependency-injector` -Only `sqlalchemy[asyncio]` and `pydantic` are required by default — everything else is an opt-in extra. +Only `sqlalchemy[asyncio]`, `pydantic` and `asyncpg` are required by default — everything else is an opt-in extra. ## Key Features @@ -148,7 +148,7 @@ async def main(): # Auto-commit on exit, auto-rollback on exception # Graceful shutdown - await session_manager.close() + await session_manager.aclose() ``` ## Architecture diff --git a/docs/reference/index.md b/docs/reference/index.md index e28d986..866297a 100644 --- a/docs/reference/index.md +++ b/docs/reference/index.md @@ -147,6 +147,12 @@ Async session manager with connection pooling and health checks. show_root_heading: false members: - AsyncSessionManagerBuilder + +::: sqlalchemy_foundation_kit.session.factories + options: + heading_level: 3 + show_root_heading: false + members: - create_async_session_manager ::: sqlalchemy_foundation_kit.session.connection @@ -189,11 +195,15 @@ async with session_manager.get_transaction() as session: session.add(user) # Auto-commit on exit, auto-rollback on exception -# Health check -is_healthy = await session_manager.healthcheck() +# Health check: run the query yourself, there is no healthcheck() method +from sqlalchemy import text +from sqlalchemy_foundation_kit import DEFAULT_HEALTHCHECK_QUERY + +async with session_manager.get_session() as session: + await session.execute(text(DEFAULT_HEALTHCHECK_QUERY)) -# Graceful shutdown -await session_manager.close(timeout=30.0) +# Graceful shutdown; the timeout is the manager's dispose_timeout +await session_manager.aclose() ``` --- diff --git a/pyproject.toml b/pyproject.toml index abdf2ab..3551cb8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -31,6 +31,7 @@ classifiers = [ dependencies = [ "sqlalchemy[asyncio]>=2.0.35,<3.0.0", "pydantic>=2.5.0,<3.0.0", + "asyncpg>=0.30.0,<1.0.0", ] [project.urls] diff --git a/sqlalchemy_foundation_kit/base/_optional.py b/sqlalchemy_foundation_kit/base/_optional.py index 33be635..6a31509 100644 --- a/sqlalchemy_foundation_kit/base/_optional.py +++ b/sqlalchemy_foundation_kit/base/_optional.py @@ -11,7 +11,7 @@ def require_optional(module_name: str, extra_name: str) -> types.ModuleType: Args: module_name: Name of the module to import (e.g., "orjson", "opentelemetry"). - extra_name: Name of the pip extra that provides this dependency (e.g., "json", "telemetry"). + extra_name: Name of the pip extra that provides this dependency (e.g., "orjson", "telemetry"). Returns: The imported module. @@ -20,7 +20,7 @@ def require_optional(module_name: str, extra_name: str) -> types.ModuleType: ImportError: If the module is not installed, with installation instructions. Examples: - >>> orjson = require_optional("orjson", "json") + >>> orjson = require_optional("orjson", "orjson") >>> from opentelemetry import trace # or >>> otel = require_optional("opentelemetry", "telemetry") diff --git a/sqlalchemy_foundation_kit/base/serialization.py b/sqlalchemy_foundation_kit/base/serialization.py index 1a5a610..4e3b56f 100644 --- a/sqlalchemy_foundation_kit/base/serialization.py +++ b/sqlalchemy_foundation_kit/base/serialization.py @@ -57,7 +57,7 @@ def _json_serializer(obj: object) -> str: >>> _json_serializer({"key": "value"}) '{"key":"value"}' """ - orjson = require_optional("orjson", "json") + orjson = require_optional("orjson", "orjson") try: return orjson.dumps(obj, default=_default_json_encoder).decode("utf-8") # type: ignore[no-any-return] @@ -82,7 +82,7 @@ def configure_orjson_serialization() -> dict[str, object]: >>> "json_deserializer" in config True """ - orjson = require_optional("orjson", "json") + orjson = require_optional("orjson", "orjson") return { "json_serializer": _json_serializer, diff --git a/sqlalchemy_foundation_kit/contrib/dependency_injector/_deps.py b/sqlalchemy_foundation_kit/contrib/dependency_injector/_deps.py index 974a641..6a6b591 100644 --- a/sqlalchemy_foundation_kit/contrib/dependency_injector/_deps.py +++ b/sqlalchemy_foundation_kit/contrib/dependency_injector/_deps.py @@ -6,14 +6,7 @@ from __future__ import annotations -try: - from dependency_injector import containers, providers - - HAS_DEPENDENCY_INJECTOR = True -except ImportError: - HAS_DEPENDENCY_INJECTOR = False - containers = None # type: ignore[misc,assignment] - providers = None # type: ignore[misc,assignment] +from typing import Any, NoReturn def check_dependency_injector() -> None: @@ -29,6 +22,35 @@ def check_dependency_injector() -> None: ) +class _MissingDependencyInjector: + """Stand-in for a dependency-injector name, raising the intended ImportError on first use. + + ``BaseDIContainer`` subclasses ``containers.DeclarativeContainer`` and the container + modules build ``providers.*`` in their class bodies, which runs before any check a + base class could perform. Without this, importing them without dependency-injector + installed fails with ``AttributeError: 'NoneType' object has no attribute + 'DeclarativeContainer'`` instead of the message that says what to install. + """ + + def __getattr__(self, name: str) -> NoReturn: + check_dependency_injector() + raise AttributeError(name) # pragma: no cover - unreachable, check raises + + def __call__(self, *args: Any, **kwargs: Any) -> NoReturn: + check_dependency_injector() + raise TypeError("dependency-injector is not installed") # pragma: no cover - unreachable + + +try: + from dependency_injector import containers, providers + + HAS_DEPENDENCY_INJECTOR = True +except ImportError: + HAS_DEPENDENCY_INJECTOR = False + containers = _MissingDependencyInjector() # type: ignore[misc,assignment] + providers = _MissingDependencyInjector() # type: ignore[misc,assignment] + + __all__ = [ "HAS_DEPENDENCY_INJECTOR", "check_dependency_injector", diff --git a/sqlalchemy_foundation_kit/contrib/di/_deps.py b/sqlalchemy_foundation_kit/contrib/di/_deps.py index 30ac6af..b6ac60e 100644 --- a/sqlalchemy_foundation_kit/contrib/di/_deps.py +++ b/sqlalchemy_foundation_kit/contrib/di/_deps.py @@ -6,15 +6,7 @@ from __future__ import annotations -try: - from dishka import Provider, Scope, provide - - HAS_DISHKA = True -except ImportError: - HAS_DISHKA = False - Provider = object # type: ignore[misc,assignment] - Scope = None # type: ignore[misc,assignment] - provide = None # type: ignore[misc,assignment] +from typing import Any, NoReturn def check_dishka() -> None: @@ -29,4 +21,33 @@ def check_dishka() -> None: ) +class _MissingDishka: + """Stand-in for a dishka name, raising the intended ImportError on first use. + + Provider modules read ``Scope.APP`` and apply ``@provide`` in their class bodies, + which runs before any check a base class could perform. Without this, importing + them without dishka installed fails with ``AttributeError: 'NoneType' object has + no attribute 'APP'`` instead of the message that says what to install. + """ + + def __getattr__(self, name: str) -> NoReturn: + check_dishka() + raise AttributeError(name) # pragma: no cover - unreachable, check_dishka raises + + def __call__(self, *args: Any, **kwargs: Any) -> NoReturn: + check_dishka() + raise TypeError("dishka is not installed") # pragma: no cover - unreachable + + +try: + from dishka import Provider, Scope, provide + + HAS_DISHKA = True +except ImportError: + HAS_DISHKA = False + Provider = object # type: ignore[misc,assignment] + Scope = _MissingDishka() # type: ignore[misc,assignment] + provide = _MissingDishka() # type: ignore[misc,assignment] + + __all__ = ["HAS_DISHKA", "Provider", "Scope", "check_dishka", "provide"] diff --git a/sqlalchemy_foundation_kit/session/locks.py b/sqlalchemy_foundation_kit/session/locks.py index ee8550f..4395114 100644 --- a/sqlalchemy_foundation_kit/session/locks.py +++ b/sqlalchemy_foundation_kit/session/locks.py @@ -1,5 +1,7 @@ """PostgreSQL advisory locks (async).""" +import hashlib + from sqlalchemy import text from sqlalchemy.ext.asyncio import AsyncSession @@ -15,6 +17,10 @@ async def try_advisory_xact_lock(session: AsyncSession, key: str | int) -> bool: at transaction end. String keys are hashed to integers. The key is then truncated to signed 64-bit as Postgres expects. + A string key produces the same lock in every process, on every host and in + every release of the library, so two replicas of a service asking for + ``"nightly-rollup"`` contend for one lock. + Args: session: SQLAlchemy AsyncSession within an active transaction. key: Lock identifier (string or integer). Strings are hashed to integers. @@ -31,7 +37,7 @@ async def try_advisory_xact_lock(session: AsyncSession, key: str | int) -> bool: ... await session.commit() """ # Convert string keys to integers via hashing - int_key = hash(key) if isinstance(key, str) else key + int_key = _hash_lock_key(key) if isinstance(key, str) else key result = await session.execute( text("SELECT pg_try_advisory_xact_lock(:k)"), @@ -40,6 +46,28 @@ async def try_advisory_xact_lock(session: AsyncSession, key: str | int) -> bool: return bool(result.scalar()) +def _hash_lock_key(key: str) -> int: + """Hash a string lock key into the PostgreSQL bigint range, reproducibly. + + Python's built-in ``hash()`` is unusable here: string hashing is salted per + interpreter, so the same key becomes a different lock in every process and two + replicas of the same service lock nothing against each other. BLAKE2b is + deterministic, so the key is stable across processes, hosts and restarts. + + Args: + key: Lock identifier. + + Returns: + A signed 64-bit integer suitable for ``pg_try_advisory_xact_lock``. + + Examples: + >>> _hash_lock_key("nightly-rollup") == _hash_lock_key("nightly-rollup") + True + """ + digest = hashlib.blake2b(key.encode("utf-8"), digest_size=8).digest() + return int.from_bytes(digest, byteorder="big", signed=True) + + def _to_signed64(key: int) -> int: """Wrap integer to PostgreSQL signed 64-bit bigint range. diff --git a/sqlalchemy_foundation_kit/session/manager.py b/sqlalchemy_foundation_kit/session/manager.py index 04b4be0..9372a66 100644 --- a/sqlalchemy_foundation_kit/session/manager.py +++ b/sqlalchemy_foundation_kit/session/manager.py @@ -238,15 +238,18 @@ async def get_transaction(self, isolation_level: str | None = None) -> AsyncIter """Get a new database session with automatic transaction management. Args: - isolation_level: Optional isolation level for the transaction. + isolation_level: Optional isolation level for the transaction, in the + PostgreSQL spelling (``"SERIALIZABLE"``, ``"REPEATABLE READ"``, …). + It is applied to the connection this transaction runs on, so it + affects only this transaction and not the engine. Yields: Managed async session with active transaction. """ self._ensure_not_closed() - options = {"isolation_level": isolation_level} if isolation_level else {} - async with ( - self._session_maker(execution_options=options) as session, - session.begin(), - ): + async with self._session_maker() as session, session.begin(): + if isolation_level is not None: + # begin() has not provisioned a connection yet, so this call is the one + # that checks it out -- the only moment the level can still be set. + await session.connection(execution_options={"isolation_level": isolation_level}) yield session diff --git a/sqlalchemy_foundation_kit/uow/sqlalchemy.py b/sqlalchemy_foundation_kit/uow/sqlalchemy.py index 485b59a..0d16718 100644 --- a/sqlalchemy_foundation_kit/uow/sqlalchemy.py +++ b/sqlalchemy_foundation_kit/uow/sqlalchemy.py @@ -74,10 +74,15 @@ async def apply_isolation_level( ) -> None: """Apply isolation level to an async session's connection. - **Implementation Detail**: - We use ``run_sync()`` because SQLAlchemy's ``execution_options()`` is a - synchronous method that configures the underlying DBAPI connection object. - We must bridge from async context to sync method via ``run_sync()``. + The level has to reach the connection *before* the connection begins its + transaction — PostgreSQL cannot change the isolation level of a transaction that + is already running, and SQLAlchemy raises ``InvalidRequestError`` if you try. So + the level is handed to the ``session.connection()`` call that checks the + connection out, which applies it and only then begins. + + Call this on a session that has not yet talked to the database. It starts the + session's transaction as a side effect, so a caller that wants to own the + transaction should join the one already open rather than call ``session.begin()``. This is a DRY utility to eliminate duplication of isolation level application logic across ``transaction()``, ``managed_session()``, and ``query()`` methods. @@ -103,9 +108,32 @@ async def apply_isolation_level( """ normalized = normalize_isolation_level(isolation_level) if normalized is not None: - conn = await session.connection() - # run_sync bridges async → sync for DBAPI-level configuration - await conn.run_sync(lambda c: c.execution_options(isolation_level=normalized)) + await session.connection(execution_options={"isolation_level": normalized}) + + +@asynccontextmanager +async def _owned_transaction(session: AsyncSession) -> AsyncIterator[None]: + """Run a block inside a transaction, committing on success and rolling back on failure. + + ``session.begin()`` refuses to start a second transaction, and applying an isolation + level has to check a connection out — which begins one. So the transaction is only + started here when the session does not already have one, and the commit/rollback is + driven explicitly instead of by the ``session.begin()`` context manager. + """ + if not session.in_transaction(): + await session.begin() + + try: + yield + except BaseException: + await session.rollback() + raise + + try: + await session.commit() + except BaseException: + await session.rollback() + raise class AsyncSQLAlchemyUowTransaction(AsyncUowTransaction): @@ -181,13 +209,14 @@ def users(self) -> UserRepository: session: AsyncSession # Type annotation for protocol compliance - async def try_advisory_lock(self, key: int) -> bool: + async def try_advisory_lock(self, key: str | int) -> bool: """Acquire a Postgres transaction-scoped advisory lock. Delegates to :func:`try_advisory_xact_lock` for actual locking logic. Args: - key: Integer lock key. + key: Lock key. An integer is used as-is; a string is hashed to one + reproducibly, so the same string is the same lock in every process. Returns: True if lock was acquired, False if already held by another session. @@ -301,7 +330,7 @@ async def transaction( if flush_before_commit is None: flush_before_commit = self._flush_before_commit - async with self.open_session(isolation_level) as session, session.begin(): + async with self.open_session(isolation_level) as session, _owned_transaction(session): uow = self._transaction_factory(session) yield uow if flush_before_commit: @@ -390,8 +419,10 @@ async def managed_session( rollback when the session closes. """ async with self.open_session(isolation_level) as session: - # Start transaction WITHOUT context manager - no auto-commit - await session.begin() + # Start transaction WITHOUT context manager - no auto-commit. + # An isolation level, if one was asked for, has already begun it. + if not session.in_transaction(): + await session.begin() try: uow = self._transaction_factory(session) yield uow, session diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index 612ba98..2d85e13 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -25,9 +25,19 @@ PostgresContainer = object # type: ignore[assignment, misc] HAS_TESTCONTAINERS = False +from sqlalchemy_foundation_kit.session.manager import AsyncSessionManager from tests.integration.models import Base +def asyncpg_url(container: PostgresContainer) -> str: + """Rewrite the container's connection URL onto the asyncpg driver.""" + url = container.get_connection_url() + if "://" in url: + _scheme, rest = url.split("://", 1) + url = f"postgresql+asyncpg://{rest}" + return url + + def is_docker_available() -> bool: """Check if Docker is available.""" if docker is None: @@ -56,13 +66,8 @@ def postgres_container() -> Generator[PostgresContainer, None, None]: @pytest_asyncio.fixture(scope="session") async def db_engine(postgres_container: PostgresContainer) -> AsyncGenerator[AsyncEngine, None]: """Create async database engine once for entire test session.""" - url = postgres_container.get_connection_url() - if "://" in url: - _scheme, rest = url.split("://", 1) - url = f"postgresql+asyncpg://{rest}" - engine = create_async_engine( - url, + asyncpg_url(postgres_container), echo=False, pool_pre_ping=True, pool_size=5, @@ -97,3 +102,16 @@ async def async_session(db_engine: AsyncEngine) -> AsyncGenerator[AsyncSession, async def async_session_factory(db_engine: AsyncEngine) -> async_sessionmaker[AsyncSession]: """Create async session factory once for entire test session.""" return async_sessionmaker(db_engine, expire_on_commit=False, class_=AsyncSession) + + +@pytest_asyncio.fixture +async def session_manager( + postgres_container: PostgresContainer, +) -> AsyncGenerator[AsyncSessionManager[AsyncSession], None]: + """Create an AsyncSessionManager pointed at the test container.""" + manager: AsyncSessionManager[AsyncSession] = AsyncSessionManager( + asyncpg_url(postgres_container), + poolclass="async_adapted_queue", + ) + yield manager + await manager.aclose() diff --git a/tests/integration/test_session_integration.py b/tests/integration/test_session_integration.py index 51265e1..2c694ac 100644 --- a/tests/integration/test_session_integration.py +++ b/tests/integration/test_session_integration.py @@ -5,9 +5,10 @@ import uuid import pytest -from sqlalchemy import select +from sqlalchemy import select, text from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy_foundation_kit.session.manager import AsyncSessionManager from tests.integration.models import Status, TestUser # ============================================================================ @@ -347,3 +348,82 @@ async def test__async_session__unique_constraint__violated_raises_error(async_se with pytest.raises(Exception): # IntegrityError async with async_session.begin(): async_session.add(user2) + + +# ============================================================================ +# AsyncSessionManager.get_transaction Tests +# ============================================================================ + + +@pytest.mark.integration +@pytest.mark.asyncio +async def test__get_transaction__clean_exit__commits( + session_manager: AsyncSessionManager[AsyncSession], + async_session: AsyncSession, +) -> None: + # Arrange + email = "get-transaction-commit@example.com" + + # Act + async with session_manager.get_transaction() as session: + session.add(TestUser(name="Committed", email=email, age=41)) + + # Assert + result = await async_session.execute(select(TestUser).where(TestUser.email == email)) + assert result.scalar_one_or_none() is not None + + +@pytest.mark.integration +@pytest.mark.asyncio +async def test__get_transaction__exception__rolls_back( + session_manager: AsyncSessionManager[AsyncSession], + async_session: AsyncSession, +) -> None: + # Arrange + email = "get-transaction-rollback@example.com" + + # Act + with pytest.raises(ValueError): + async with session_manager.get_transaction() as session: + session.add(TestUser(name="Rolled back", email=email, age=41)) + await session.flush() + raise ValueError("Intentional rollback") + + # Assert + result = await async_session.execute(select(TestUser).where(TestUser.email == email)) + assert result.scalar_one_or_none() is None + + +@pytest.mark.integration +@pytest.mark.asyncio +@pytest.mark.parametrize( + "isolation_level,expected", + [ + ("SERIALIZABLE", "serializable"), + ("REPEATABLE READ", "repeatable read"), + ], +) +async def test__get_transaction__isolation_level__applied_to_the_transaction( + session_manager: AsyncSessionManager[AsyncSession], + isolation_level: str, + expected: str, +) -> None: + # Act + async with session_manager.get_transaction(isolation_level=isolation_level) as session: + actual = (await session.execute(text("SHOW transaction_isolation"))).scalar_one() + + # Assert + assert actual == expected + + +@pytest.mark.integration +@pytest.mark.asyncio +async def test__get_transaction__no_isolation_level__leaves_the_server_default( + session_manager: AsyncSessionManager[AsyncSession], +) -> None: + # Act + async with session_manager.get_transaction() as session: + actual = (await session.execute(text("SHOW transaction_isolation"))).scalar_one() + + # Assert + assert actual == "read committed" diff --git a/tests/integration/test_uow_integration.py b/tests/integration/test_uow_integration.py index bf154ff..fe80ac3 100644 --- a/tests/integration/test_uow_integration.py +++ b/tests/integration/test_uow_integration.py @@ -7,6 +7,7 @@ from sqlalchemy.exc import DBAPIError, IntegrityError from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker +from sqlalchemy_foundation_kit.uow.enums import IsolationLevel from sqlalchemy_foundation_kit.uow.sqlalchemy import AsyncSQLAlchemyUnitOfWork, AsyncSQLAlchemyUowTransaction from tests.integration.models import TestUser @@ -171,3 +172,119 @@ async def test__savepoint__inside_managed_session__transaction_stays_usable( async with async_session_factory() as session: emails = (await session.execute(select(TestUser.email))).scalars().all() assert list(emails) == ["after@example.com"] + + +# ============================================================================ +# Isolation Level Tests +# ============================================================================ + + +@pytest.fixture +def uow( + async_session_factory: async_sessionmaker[AsyncSession], +) -> AsyncSQLAlchemyUnitOfWork[AsyncSQLAlchemyUowTransaction]: + return AsyncSQLAlchemyUnitOfWork(async_session_factory, transaction_factory=AsyncSQLAlchemyUowTransaction) + + +async def _transaction_isolation(session: AsyncSession) -> str: + return (await session.execute(text("SHOW transaction_isolation"))).scalar_one() # type: ignore[no-any-return] + + +@pytest.mark.integration +@pytest.mark.asyncio +@pytest.mark.parametrize( + "isolation_level,expected", + [ + (IsolationLevel.SERIALIZABLE, "serializable"), + (IsolationLevel.REPEATABLE_READ, "repeatable read"), + ("READ_COMMITTED", "read committed"), + ], +) +async def test__transaction__isolation_level__applied_to_the_transaction( + uow: AsyncSQLAlchemyUnitOfWork[AsyncSQLAlchemyUowTransaction], + isolation_level: IsolationLevel | str, + expected: str, +) -> None: + # Act + async with uow.transaction(isolation_level=isolation_level) as tx: + actual = await _transaction_isolation(tx.session) + + # Assert + assert actual == expected + + +@pytest.mark.integration +@pytest.mark.asyncio +async def test__managed_session__isolation_level__applied_to_the_transaction( + uow: AsyncSQLAlchemyUnitOfWork[AsyncSQLAlchemyUowTransaction], +) -> None: + # Act + async with uow.managed_session(isolation_level=IsolationLevel.SERIALIZABLE) as (_tx, session): + actual = await _transaction_isolation(session) + await session.rollback() + + # Assert + assert actual == "serializable" + + +@pytest.mark.integration +@pytest.mark.asyncio +async def test__query__isolation_level__applied_to_the_transaction( + uow: AsyncSQLAlchemyUnitOfWork[AsyncSQLAlchemyUowTransaction], +) -> None: + # Act + async with uow.query(isolation_level=IsolationLevel.REPEATABLE_READ) as qx: + actual = await _transaction_isolation(qx.session) + + # Assert + assert actual == "repeatable read" + + +@pytest.mark.integration +@pytest.mark.asyncio +async def test__transaction__isolation_level__still_commits( + uow: AsyncSQLAlchemyUnitOfWork[AsyncSQLAlchemyUowTransaction], + async_session: AsyncSession, +) -> None: + # Arrange + email = "uow-isolation-commit@example.com" + + # Act + async with uow.transaction(isolation_level=IsolationLevel.SERIALIZABLE) as tx: + tx.session.add(TestUser(name="Serializable", email=email, age=33)) + + # Assert + result = await async_session.execute(select(TestUser).where(TestUser.email == email)) + assert result.scalar_one_or_none() is not None + + +@pytest.mark.integration +@pytest.mark.asyncio +async def test__transaction__isolation_level__rolls_back_on_exception( + uow: AsyncSQLAlchemyUnitOfWork[AsyncSQLAlchemyUowTransaction], + async_session: AsyncSession, +) -> None: + # Arrange + email = "uow-isolation-rollback@example.com" + + # Act + with pytest.raises(ValueError): + async with uow.transaction(isolation_level=IsolationLevel.SERIALIZABLE) as tx: + tx.session.add(TestUser(name="Serializable", email=email, age=33)) + await tx.session.flush() + raise ValueError("Intentional rollback") + + # Assert + result = await async_session.execute(select(TestUser).where(TestUser.email == email)) + assert result.scalar_one_or_none() is None + + +@pytest.mark.integration +@pytest.mark.asyncio +async def test__transaction__invalid_isolation_level__raises_value_error( + uow: AsyncSQLAlchemyUnitOfWork[AsyncSQLAlchemyUowTransaction], +) -> None: + # Act & Assert + with pytest.raises(ValueError, match="Invalid isolation level"): + async with uow.transaction(isolation_level="NOT A LEVEL"): + pass diff --git a/tests/unit/contrib/telemetry/test_uow.py b/tests/unit/contrib/telemetry/test_uow.py index e7bbcd9..81c75c2 100644 --- a/tests/unit/contrib/telemetry/test_uow.py +++ b/tests/unit/contrib/telemetry/test_uow.py @@ -269,6 +269,8 @@ async def test__traced_async_unit_of_work__transaction_with_isolation_level__set mock_transaction = MagicMock() mock_session = AsyncMock() mock_session.begin = MagicMock() + # Asking for an isolation level checks a connection out, which begins the transaction. + mock_session.in_transaction = MagicMock(return_value=True) mock_session.commit = AsyncMock() mock_session.rollback = AsyncMock() mock_session.close = AsyncMock() diff --git a/tests/unit/contrib/test_optional_imports.py b/tests/unit/contrib/test_optional_imports.py new file mode 100644 index 0000000..fdc543e --- /dev/null +++ b/tests/unit/contrib/test_optional_imports.py @@ -0,0 +1,71 @@ +"""Unit tests for importing contrib packages without their extra installed. + +The check has to happen in a subprocess: the failure is at *import* time, in a class +body, so it cannot be reproduced by patching a flag inside an already-imported module. +""" + +from __future__ import annotations + +import subprocess +import sys + +import pytest + +_IMPORT_WITH_DEPENDENCY_HIDDEN = """ +import sys +from importlib.abc import MetaPathFinder + + +class Blocker(MetaPathFinder): + def __init__(self, name): + self._name = name + + def find_spec(self, fullname, path=None, target=None): + if fullname == self._name or fullname.startswith(self._name + "."): + raise ImportError(f"No module named {{fullname!r}}") + return None + + +sys.meta_path.insert(0, Blocker({dependency!r})) + +try: + import {module} +except ImportError as exc: + print(type(exc).__name__, exc) +else: + print("imported without error") +""" + + +def _import_with_dependency_hidden(module: str, dependency: str) -> str: + script = _IMPORT_WITH_DEPENDENCY_HIDDEN.format(module=module, dependency=dependency) + completed = subprocess.run( # noqa: S603 + [sys.executable, "-c", script], + capture_output=True, + check=True, + text=True, + ) + return completed.stdout.strip() + + +@pytest.mark.unit +def test__import_contrib_di__without_dishka__raises_import_error_naming_the_extra() -> None: + # Arrange & Act + output = _import_with_dependency_hidden("sqlalchemy_foundation_kit.contrib.di", "dishka") + + # Assert + assert output.startswith("ImportError") + assert "sqlalchemy-foundation-kit[dishka]" in output + + +@pytest.mark.unit +def test__import_contrib_dependency_injector__without_it__raises_import_error_naming_the_extra() -> None: + # Arrange & Act + output = _import_with_dependency_hidden( + "sqlalchemy_foundation_kit.contrib.dependency_injector", + "dependency_injector", + ) + + # Assert + assert output.startswith("ImportError") + assert "sqlalchemy-foundation-kit[dependency-injector]" in output diff --git a/tests/unit/session/test_locks.py b/tests/unit/session/test_locks.py index 6a74cdb..b6ce678 100644 --- a/tests/unit/session/test_locks.py +++ b/tests/unit/session/test_locks.py @@ -2,11 +2,63 @@ from __future__ import annotations +import subprocess +import sys from unittest.mock import AsyncMock, Mock import pytest -from sqlalchemy_foundation_kit.session.locks import _to_signed64, try_advisory_xact_lock +from sqlalchemy_foundation_kit.session.locks import _hash_lock_key, _to_signed64, try_advisory_xact_lock + +# ============================================================================ +# _hash_lock_key Tests +# ============================================================================ + + +def test__hash_lock_key__same_key__same_value() -> None: + # Arrange & Act + first = _hash_lock_key("nightly-rollup") + second = _hash_lock_key("nightly-rollup") + + # Assert + assert first == second + + +def test__hash_lock_key__different_keys__different_values() -> None: + # Arrange & Act & Assert + assert _hash_lock_key("job-a") != _hash_lock_key("job-b") + + +def test__hash_lock_key__any_key__fits_signed_64_bit() -> None: + # Arrange + keys = ["", "a" * 1000, "test_lock_\U0001f512", "test-lock_123!@#$%^&*()"] + + # Act & Assert + for key in keys: + assert -(2**63) <= _hash_lock_key(key) <= 2**63 - 1 + + +def test__hash_lock_key__fresh_interpreters__agree_on_the_key() -> None: + # Arrange: two replicas of a service are two processes. Python salts str.__hash__ + # per interpreter, so a hash()-based key made them lock against nothing. + script = ( + "from sqlalchemy_foundation_kit.session.locks import _hash_lock_key; print(_hash_lock_key('nightly-rollup'))" + ) + + # Act + keys = { + subprocess.run( # noqa: S603 + [sys.executable, "-c", script], + capture_output=True, + check=True, + text=True, + ).stdout.strip() + for _ in range(3) + } + + # Assert + assert keys == {str(_hash_lock_key("nightly-rollup"))} + # ============================================================================ # _to_signed64 Tests @@ -172,6 +224,23 @@ def test__to_signed64__idempotent() -> None: # ============================================================================ +@pytest.mark.asyncio +async def test__try_advisory_xact_lock__string_key__uses_the_reproducible_hash() -> None: + # Arrange + mock_result = Mock() + mock_result.scalar.return_value = True + + mock_session = AsyncMock() + mock_session.execute = AsyncMock(return_value=mock_result) + + # Act + await try_advisory_xact_lock(mock_session, "nightly-rollup") + + # Assert + params = mock_session.execute.call_args[0][1] + assert params["k"] == _to_signed64(_hash_lock_key("nightly-rollup")) + + @pytest.mark.asyncio async def test__try_advisory_xact_lock__lock_acquired__returns_true() -> None: # Arrange diff --git a/tests/unit/session/test_manager.py b/tests/unit/session/test_manager.py index e0de38e..83e5ff6 100644 --- a/tests/unit/session/test_manager.py +++ b/tests/unit/session/test_manager.py @@ -689,7 +689,7 @@ async def test__async_session_manager__get_transaction__yields_session() -> None @pytest.mark.asyncio -async def test__async_session_manager__get_transaction__with_isolation_level__passes_options() -> None: +async def test__async_session_manager__get_transaction__with_isolation_level__sets_it_on_the_connection() -> None: # Arrange mock_session = AsyncMock(spec=AsyncSession) @@ -716,12 +716,13 @@ async def test__async_session_manager__get_transaction__with_isolation_level__pa async with manager.get_transaction(isolation_level="SERIALIZABLE"): pass - # Assert - mock_maker_obj.assert_called_with(execution_options={"isolation_level": "SERIALIZABLE"}) + # Assert: the level travels with the connection checkout, never to the session factory + mock_maker_obj.assert_called_once_with() + mock_session.connection.assert_awaited_once_with(execution_options={"isolation_level": "SERIALIZABLE"}) @pytest.mark.asyncio -async def test__async_session_manager__get_transaction__no_isolation_level__no_options() -> None: +async def test__async_session_manager__get_transaction__no_isolation_level__no_connection_options() -> None: # Arrange mock_session = AsyncMock(spec=AsyncSession) @@ -749,10 +750,25 @@ async def test__async_session_manager__get_transaction__no_isolation_level__no_o pass # Assert - # Check that it was called with empty execution_options - assert mock_maker_obj.call_count == 1 - call_kwargs = mock_maker_obj.call_args[1] if mock_maker_obj.call_args[1] else {} - assert call_kwargs.get("execution_options") == {} + mock_maker_obj.assert_called_once_with() + mock_session.connection.assert_not_awaited() + + +@pytest.mark.asyncio +async def test__async_session_manager__get_transaction__real_session_maker__opens_a_transaction() -> None: + # Arrange: a real engine and sessionmaker. Nothing here reaches the database -- the + # session is opened, begun and committed without ever checking out a connection -- + # but the session is constructed for real, which is where get_transaction used to + # raise TypeError before the caller saw anything. + manager: AsyncSessionManager[AsyncSession] = AsyncSessionManager("postgresql+asyncpg://u:p@127.0.0.1:1/db") + + # Act + async with manager.get_transaction() as session: + in_transaction = session.in_transaction() + + # Assert + assert in_transaction is True + await manager.aclose() @pytest.mark.asyncio diff --git a/tests/unit/test_distribution.py b/tests/unit/test_distribution.py new file mode 100644 index 0000000..5a4d49c --- /dev/null +++ b/tests/unit/test_distribution.py @@ -0,0 +1,25 @@ +"""Unit tests for what the distribution promises to install.""" + +from __future__ import annotations + +from importlib.metadata import requires + +import pytest + +DISTRIBUTION = "sqlalchemy-foundation-kit" + + +def _unconditional_requirements() -> list[str]: + """Requirements installed by ``pip install sqlalchemy-foundation-kit``, extras aside.""" + return [requirement for requirement in requires(DISTRIBUTION) or [] if "extra ==" not in requirement] + + +@pytest.mark.unit +@pytest.mark.parametrize("distribution", ["sqlalchemy", "pydantic", "asyncpg"]) +def test__unconditional_requirements__contains_every_import_the_package_needs(distribution: str) -> None: + # Arrange: importing the package pulls in asyncpg (AsyncCConnection subclasses + # asyncpg.Connection at module scope), so a plain install has to bring it. + requirements = _unconditional_requirements() + + # Act & Assert + assert any(requirement.startswith(distribution) for requirement in requirements), requirements diff --git a/tests/unit/uow/test_sqlalchemy.py b/tests/unit/uow/test_sqlalchemy.py index 8046dee..9feba1d 100644 --- a/tests/unit/uow/test_sqlalchemy.py +++ b/tests/unit/uow/test_sqlalchemy.py @@ -148,23 +148,21 @@ async def test__apply_isolation_level__none__does_nothing() -> None: async def test__apply_isolation_level__read_committed__applies() -> None: # Arrange mock_connection = AsyncMock() - mock_connection.run_sync = AsyncMock() mock_session = AsyncMock(spec=AsyncSession) mock_session.connection = AsyncMock(return_value=mock_connection) # Act await apply_isolation_level(mock_session, IsolationLevel.READ_COMMITTED) - # Assert - mock_session.connection.assert_called_once() - mock_connection.run_sync.assert_called_once() + # Assert: the level has to reach the connection as it is checked out. Setting it + # afterwards is what PostgreSQL refuses -- the transaction has already started. + mock_session.connection.assert_awaited_once_with(execution_options={"isolation_level": "READ COMMITTED"}) @pytest.mark.asyncio async def test__apply_isolation_level__serializable__applies() -> None: # Arrange mock_connection = AsyncMock() - mock_connection.run_sync = AsyncMock() mock_session = AsyncMock(spec=AsyncSession) mock_session.connection = AsyncMock(return_value=mock_connection) @@ -172,7 +170,7 @@ async def test__apply_isolation_level__serializable__applies() -> None: await apply_isolation_level(mock_session, "SERIALIZABLE") # Assert - mock_session.connection.assert_called_once() + mock_session.connection.assert_awaited_once_with(execution_options={"isolation_level": "SERIALIZABLE"}) @pytest.mark.asyncio @@ -428,14 +426,11 @@ async def test__open_session__with_isolation_level__applies() -> None: @pytest.mark.asyncio async def test__transaction__commits_on_success() -> None: # Arrange - mock_begin_context = AsyncMock() - mock_begin_context.__aenter__ = AsyncMock(return_value=None) - mock_begin_context.__aexit__ = AsyncMock(return_value=None) - mock_session = AsyncMock(spec=AsyncSession) mock_session.__aenter__ = AsyncMock(return_value=mock_session) mock_session.__aexit__ = AsyncMock(return_value=None) - mock_session.begin = Mock(return_value=mock_begin_context) + mock_session.begin = AsyncMock() + mock_session.in_transaction = Mock(return_value=False) mock_session.flush = AsyncMock() mock_transaction = Mock() @@ -455,14 +450,11 @@ async def test__transaction__commits_on_success() -> None: @pytest.mark.asyncio async def test__transaction__flushes_before_commit__when_enabled() -> None: # Arrange - mock_begin_context = AsyncMock() - mock_begin_context.__aenter__ = AsyncMock(return_value=None) - mock_begin_context.__aexit__ = AsyncMock(return_value=None) - mock_session = AsyncMock(spec=AsyncSession) mock_session.__aenter__ = AsyncMock(return_value=mock_session) mock_session.__aexit__ = AsyncMock(return_value=None) - mock_session.begin = Mock(return_value=mock_begin_context) + mock_session.begin = AsyncMock() + mock_session.in_transaction = Mock(return_value=False) mock_session.flush = AsyncMock() mock_transaction = Mock() @@ -482,14 +474,11 @@ async def test__transaction__flushes_before_commit__when_enabled() -> None: @pytest.mark.asyncio async def test__transaction__skips_flush__when_disabled() -> None: # Arrange - mock_begin_context = AsyncMock() - mock_begin_context.__aenter__ = AsyncMock(return_value=None) - mock_begin_context.__aexit__ = AsyncMock(return_value=None) - mock_session = AsyncMock(spec=AsyncSession) mock_session.__aenter__ = AsyncMock(return_value=mock_session) mock_session.__aexit__ = AsyncMock(return_value=None) - mock_session.begin = Mock(return_value=mock_begin_context) + mock_session.begin = AsyncMock() + mock_session.in_transaction = Mock(return_value=False) mock_session.flush = AsyncMock() mock_transaction = Mock() @@ -509,14 +498,11 @@ async def test__transaction__skips_flush__when_disabled() -> None: @pytest.mark.asyncio async def test__transaction__flush_override__overrides_default() -> None: # Arrange - mock_begin_context = AsyncMock() - mock_begin_context.__aenter__ = AsyncMock(return_value=None) - mock_begin_context.__aexit__ = AsyncMock(return_value=None) - mock_session = AsyncMock(spec=AsyncSession) mock_session.__aenter__ = AsyncMock(return_value=mock_session) mock_session.__aexit__ = AsyncMock(return_value=None) - mock_session.begin = Mock(return_value=mock_begin_context) + mock_session.begin = AsyncMock() + mock_session.in_transaction = Mock(return_value=False) mock_session.flush = AsyncMock() mock_transaction = Mock() @@ -536,14 +522,11 @@ async def test__transaction__flush_override__overrides_default() -> None: @pytest.mark.asyncio async def test__transaction__flush_error__logs_warning_and_raises() -> None: # Arrange - mock_begin_context = AsyncMock() - mock_begin_context.__aenter__ = AsyncMock(return_value=None) - mock_begin_context.__aexit__ = AsyncMock(return_value=None) - mock_session = AsyncMock(spec=AsyncSession) mock_session.__aenter__ = AsyncMock(return_value=mock_session) mock_session.__aexit__ = AsyncMock(return_value=None) - mock_session.begin = Mock(return_value=mock_begin_context) + mock_session.begin = AsyncMock() + mock_session.in_transaction = Mock(return_value=False) mock_session.flush = AsyncMock(side_effect=SQLAlchemyError("Flush error")) mock_transaction = Mock() @@ -561,6 +544,58 @@ async def test__transaction__flush_error__logs_warning_and_raises() -> None: mock_logger.warning.assert_called_once() +@pytest.mark.asyncio +async def test__transaction__isolation_level_already_began__joins_it_and_commits() -> None: + # Arrange: applying an isolation level checks a connection out, which begins the + # transaction. session.begin() would raise on top of that, so the UoW joins it. + mock_session = AsyncMock(spec=AsyncSession) + mock_session.__aenter__ = AsyncMock(return_value=mock_session) + mock_session.__aexit__ = AsyncMock(return_value=None) + mock_session.begin = Mock() + mock_session.in_transaction = Mock(return_value=True) + mock_session.flush = AsyncMock() + mock_session.commit = AsyncMock() + + mock_transaction = Mock() + mock_session_maker = Mock(return_value=mock_session) + mock_factory = Mock(return_value=mock_transaction) + + uow = AsyncSQLAlchemyUnitOfWork(mock_session_maker, mock_factory) + + # Act + async with uow.transaction(isolation_level=IsolationLevel.SERIALIZABLE) as tx: + assert tx is mock_transaction + + # Assert + mock_session.begin.assert_not_called() + mock_session.commit.assert_awaited_once() + + +@pytest.mark.asyncio +async def test__transaction__exception__rolls_back() -> None: + # Arrange + mock_session = AsyncMock(spec=AsyncSession) + mock_session.__aenter__ = AsyncMock(return_value=mock_session) + mock_session.__aexit__ = AsyncMock(return_value=None) + mock_session.begin = AsyncMock() + mock_session.in_transaction = Mock(return_value=False) + mock_session.rollback = AsyncMock() + mock_session.commit = AsyncMock() + + mock_session_maker = Mock(return_value=mock_session) + mock_factory = Mock(return_value=Mock()) + + uow = AsyncSQLAlchemyUnitOfWork(mock_session_maker, mock_factory) + + # Act & Assert + with pytest.raises(ValueError): + async with uow.transaction(): + raise ValueError("Test error") + + mock_session.rollback.assert_awaited_once() + mock_session.commit.assert_not_awaited() + + # ============================================================================ # managed_session Tests # ============================================================================ @@ -573,6 +608,7 @@ async def test__managed_session__yields_transaction_and_session() -> None: mock_session.__aenter__ = AsyncMock(return_value=mock_session) mock_session.__aexit__ = AsyncMock(return_value=None) mock_session.begin = AsyncMock() + mock_session.in_transaction = Mock(return_value=False) mock_session.commit = AsyncMock() mock_transaction = Mock() @@ -597,6 +633,7 @@ async def test__managed_session__requires_manual_commit() -> None: mock_session.__aenter__ = AsyncMock(return_value=mock_session) mock_session.__aexit__ = AsyncMock(return_value=None) mock_session.begin = AsyncMock() + mock_session.in_transaction = Mock(return_value=False) mock_session.commit = AsyncMock() mock_transaction = Mock() @@ -613,6 +650,28 @@ async def test__managed_session__requires_manual_commit() -> None: mock_session.commit.assert_called_once() +@pytest.mark.asyncio +async def test__managed_session__isolation_level_already_began__does_not_begin_again() -> None: + # Arrange + mock_session = AsyncMock(spec=AsyncSession) + mock_session.__aenter__ = AsyncMock(return_value=mock_session) + mock_session.__aexit__ = AsyncMock(return_value=None) + mock_session.begin = AsyncMock() + mock_session.in_transaction = Mock(return_value=True) + + mock_session_maker = Mock(return_value=mock_session) + mock_factory = Mock(return_value=Mock()) + + uow = AsyncSQLAlchemyUnitOfWork(mock_session_maker, mock_factory) + + # Act + async with uow.managed_session(isolation_level="SERIALIZABLE"): + pass + + # Assert + mock_session.begin.assert_not_called() + + @pytest.mark.asyncio async def test__managed_session__exception__rolls_back() -> None: # Arrange @@ -620,6 +679,7 @@ async def test__managed_session__exception__rolls_back() -> None: mock_session.__aenter__ = AsyncMock(return_value=mock_session) mock_session.__aexit__ = AsyncMock(return_value=None) mock_session.begin = AsyncMock() + mock_session.in_transaction = Mock(return_value=False) mock_session.rollback = AsyncMock() mock_transaction = Mock() diff --git a/uv.lock b/uv.lock index f118068..0926119 100644 --- a/uv.lock +++ b/uv.lock @@ -1576,6 +1576,7 @@ asyncio = [ name = "sqlalchemy-foundation-kit" source = { editable = "." } dependencies = [ + { name = "asyncpg" }, { name = "pydantic" }, { name = "sqlalchemy", extra = ["asyncio"] }, ] @@ -1645,6 +1646,7 @@ test = [ [package.metadata] requires-dist = [ + { name = "asyncpg", specifier = ">=0.30.0,<1.0.0" }, { name = "dependency-injector", marker = "extra == 'all'", specifier = ">=4.41.0,<5.0.0" }, { name = "dependency-injector", marker = "extra == 'dependency-injector'", specifier = ">=4.41.0,<5.0.0" }, { name = "dishka", marker = "extra == 'all'", specifier = ">=1.3.0,<2.0.0" },