Metadata-Version: 2.5
Name: csrd-repository
Version: 0.7.0
Summary: Base repository and database adapter abstractions
Project-URL: Repository, https://github.com/csrd-api/fastapi-common
Project-URL: Documentation, https://github.com/csrd-api/fastapi-common/tree/main/packages/repository
Project-URL: Changelog, https://github.com/csrd-api/fastapi-common/blob/main/CHANGELOG.md
License: MIT
Requires-Python: >=3.12
Requires-Dist: aiosqlite<1,>=0.21
Requires-Dist: csrd-models
Provides-Extra: maria
Requires-Dist: pymysql<2,>=1.1; extra == 'maria'
Provides-Extra: pg
Requires-Dist: psycopg-pool<4,>=3.2; extra == 'pg'
Requires-Dist: psycopg[binary]<4,>=3.2; extra == 'pg'
Provides-Extra: sql
Requires-Dist: sqlalchemy<3,>=2.0; extra == 'sql'
Description-Content-Type: text/markdown

 # csrd-repository

Base repository and database adapter abstractions for FastAPI microservices.

**Package**: `csrd.repository` · **Import**: `from csrd.repository import BaseRepository, SQLiteAdapter`

## What's included

- `BaseRepository` — abstract repository with model parsing (requires an adapter)
- `ABCDatabaseAdapter` / `DBProtocol` — async database adapter abstraction with lifecycle and explicit transactions
- `SQLiteAdapter` — async SQLite implementation via aiosqlite (persistent connection, atomic upsert)
- `PGAdapter` — PostgreSQL via psycopg with a `psycopg_pool` connection pool, `sslmode` / libpq `options`
- `MariaAdapter` — MariaDB/MySQL via pymysql with a bounded connection pool, `ssl` / `options`
- `ExecuteResult` — immutable snapshot of query metadata (replaces raw cursor return)
- Optional `sqlalchemy` integration for query building (`pip install csrd-repository[sql]`)

## Installation

```bash
uv pip install "csrd-repository @ git+ssh://git@github.com/csrd-api/fastapi-common.git#subdirectory=packages/repository"
```

## Dependencies

- `csrd-models` (Tier 2)

Network drivers are optional extras:

```bash
uv pip install "csrd-repository[pg]"     # psycopg[binary] + psycopg-pool
uv pip install "csrd-repository[maria]"  # pymysql
```

## Adapter lifecycle and connection pooling

Create **one adapter per process**, `connect()` it at startup and `close()` it
at shutdown. `PGAdapter` owns a `psycopg_pool.ConnectionPool`; `MariaAdapter`
owns a small built-in bounded pool; `SQLiteAdapter` holds one connection.
Queries issued before `connect()` (or after `close()`) raise `RuntimeError`
— the adapters never open a hidden, unmanaged pool.

```python
adapter = PGAdapter(
    host="db", port=5432, user="svc", password=secret, database="svc",
    pool_min_size=1,      # opened eagerly by connect()
    pool_max_size=10,     # hard cap on concurrent server connections
    pool_max_idle=600.0,  # seconds before an idle connection is closed
    pool_timeout=30.0,    # wait for a free connection / for connect()
    pool_check=True,      # validate a connection on checkout, replace dead ones
)
```

`connect()` waits until `pool_min_size` connections are established, so an
unreachable database fails at startup. Plain queries borrow a pooled
connection in autocommit mode for one statement.

### FastAPI

The scaffolded lifespan creates the adapter once and stores it on app state;
request handlers (HTTP and WebSocket) read it from there:

```python
@asynccontextmanager
async def _db_lifespan(app: FastAPI) -> AsyncIterator[dict[str, object]]:
    adapter = create_adapter(settings)
    await adapter.connect()
    try:
        yield {"db_adapter": adapter}
    finally:
        await adapter.close()


def _adapter_from_state(connection: HTTPConnection) -> PGAdapter:
    return connection.app.state.db_adapter


Adapter = Annotated[PGAdapter, Depends(_adapter_from_state)]
```

Never use `Depends(create_adapter)`: that would build (and, with pooling,
need to open) a new adapter per request.

### Workers and scripts (outside FastAPI)

Use the adapter as an async context manager, or call `connect()`/`close()`
around the worker's lifetime — not per task:

```python
async def main() -> None:
    async with create_adapter(settings) as adapter:   # connect() ... close()
        await run_worker(adapter)
```

If tasks run on short-lived event loops (for example one `asyncio.run()` per
task), wrap each run in `async with create_adapter(settings) as adapter:` so
the pool is always closed; prefer one long-lived loop and adapter per worker
process where possible, since every `connect()` opens fresh connections.

## PostgreSQL TLS and connection options

```python
PGAdapter(
    host="10.10.10.80", port=5432, user="svc", password=secret, database="svc",
    sslmode="verify-full",
    options={
        "sslrootcert": "/etc/ssl/certs/db-ca.pem",
        "application_name": "catalog-service",
        "connect_timeout": "5",
    },
)
```

- `sslmode: str | None = None` — when `None` the keyword is omitted, so libpq
  still honours `PGSSLMODE` (and other `PG*` variables) from the environment.
- `options: Mapping[str, str] | None` — any other libpq keyword. Keys that
  duplicate explicit arguments (`host`, `port`, `user`, `password`, `dbname`,
  `sslmode`) raise `ValueError`.
- The conninfo is built with `psycopg.conninfo.make_conninfo`, so passwords
  containing spaces, quotes or backslashes are quoted correctly. The pool uses
  this same conninfo.

`MariaAdapter` accepts `ssl: Mapping[str, Any] | None` (passed to
`pymysql.connect(ssl=...)`, e.g. `{"ca": "/etc/ssl/ca.pem"}`) and
`options: Mapping[str, Any] | None` for other `pymysql.connect` keywords.

## Transactions

All adapters expose an explicit unit-of-work context:

```python
async with adapter.transaction() as tx:
    users = user_repository.with_adapter(tx)
    credentials = credential_repository.with_adapter(tx)
    user = await users.insert("users", user_values)
    await credentials.insert("credentials", {"user_id": user["id"], **credential_values})
```

The yielded adapter uses one physical connection (borrowed from the pool for
the whole block with autocommit off, and returned rolled back/reset). Normal
exit commits; exceptions and cancellation roll back. A transaction-bound
adapter raises `RuntimeError` if reused after context exit.

### Joining an open transaction

`adapter.in_transaction` is `True` while an adapter is bound to an open
transaction. Nesting `transaction()` on a bound adapter raises
`RuntimeError` unless you opt in with `join=True`:

```python
async def insert_many(self, rows):
    # All-or-nothing when called alone; part of the caller's unit of work
    # when called with a transaction-bound adapter.
    async with self._adapter.transaction(join=True) as tx:
        for row in rows:
            await tx.insert("cards", row)
```

- On a bound adapter, `join=True` yields **that same adapter**; the joined
  block neither commits nor rolls back — the outermost `transaction()` does.
  An exception escaping the joined block rolls back the whole outer
  transaction unless the outer code catches it.
- On an unbound adapter, `join=True` opens a new transaction as usual.
- Joining is never implicit: code whose write must survive an outer rollback
  (for example a security revocation) should keep calling plain
  `transaction()` on an unbound adapter, which never joins.

Locking remains database-specific: PostgreSQL and MariaDB support their native
row-lock syntax (for example `SELECT ... FOR UPDATE`); SQLite serializes
operations on its single connection and does not support `FOR UPDATE`.
