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 transactionsSQLiteAdapter— async SQLite implementation via aiosqlite (persistent connection, atomic upsert)PGAdapter— PostgreSQL via psycopg with apsycopg_poolconnection pool,sslmode/ libpqoptionsMariaAdapter— MariaDB/MySQL via pymysql with a bounded connection pool,ssl/optionsExecuteResult— immutable snapshot of query metadata (replaces raw cursor return)- Optional
sqlalchemyintegration for query building (pip install csrd-repository[sql])
Installation
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:
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.
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:
@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:
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
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— whenNonethe keyword is omitted, so libpq still honoursPGSSLMODE(and otherPG*variables) from the environment.options: Mapping[str, str] | None— any other libpq keyword. Keys that duplicate explicit arguments (host,port,user,password,dbname,sslmode) raiseValueError.- 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:
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:
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=Trueyields that same adapter; the joined block neither commits nor rolls back — the outermosttransaction()does. An exception escaping the joined block rolls back the whole outer transaction unless the outer code catches it. - On an unbound adapter,
join=Trueopens 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.
1 maintainer
Metadata
1 maintainer
Release files for csrd-repository 0.7.0
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | |
|---|---|---|---|
| csrd_repository-0.7.0.tar.gz | 25.4 kB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| csrd_repository-0.7.0-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 49.5 kB
Release files / csrd_repository-0.7.0.tar.gz
| Download URL | csrd_repository-0.7.0.tar.gz |
|---|---|
| Size | 25.4 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
4dcccfaf7b5e5afbce60df6d26101e5796ecd41b66005ec8cb9f9a7e65a928ff
|
|
BLAKE2b-256 checksum How to use checksums |
506e079d463c61b526d03812a00b8ad7877e63e7c17c7ab56f4b233591aa1a76
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
twine/7.0.0 CPython/3.13.14
|
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
PyPI Publish Attestation
PyPI verified that this artifact, at this checksum, originated from the publisher listed below.
Signed by GitHub Actions, verified by PyPI on Oct 16, 2026.
Transparency logRelease files / csrd_repository-0.7.0-py3-none-any.whl
| Download URL | csrd_repository-0.7.0-py3-none-any.whl |
|---|---|
| Size | 24.1 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
7dc47569e804f892d5b704e6c34f18cb3d8f849c3cd8e8d08b586c180175dcbc
|
|
BLAKE2b-256 checksum How to use checksums |
b4dc6bf991f36ade7aff2cdea433e476f965450a2faca05fdc2590b8c726893a
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
twine/7.0.0 CPython/3.13.14
|
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
PyPI Publish Attestation
PyPI verified that this artifact, at this checksum, originated from the publisher listed below.
Signed by GitHub Actions, verified by PyPI on Oct 16, 2026.
Transparency log
