openframe-adapters-db-postgres

PostgreSQL database adapter for the OpenFrame Microservice Suite.

Part of the openframe-adapters monorepo. Implements BaseRepository[T] and HealthCheck from openframe-core using asyncpg.

Installation

pip install openframe-adapters-db-postgres

Required env var:

DATABASE_URL=postgresql://user:password@host:5432/dbname

Quick start

Raw dict mode

from openframe.adapters.db.postgres import PostgresSettings, PostgresRepository

settings = PostgresSettings()  # reads DATABASE_URL from env
repo = PostgresRepository(settings, table="items", id_column="id")

item = await repo.get("abc-123")          # dict | None
items, total = await repo.list(10, 0)     # ([dict, ...], int)
created = await repo.create({"name": "x"})
updated = await repo.update({"id": "abc-123", "name": "y"})
deleted = await repo.delete("abc-123")    # bool

Typed domain mode

from dataclasses import dataclass
from openframe.adapters.db.postgres import PostgresSettings, PostgresRepository

@dataclass
class Item:
    id: str
    name: str

class ItemRepository(PostgresRepository[Item]):
    _table = "items"
    _id_column = "id"

    def _row_to_entity(self, row) -> Item:
        return Item(**dict(row))

    def _entity_to_row(self, entity: Item) -> dict:
        return {"id": entity.id, "name": entity.name}

settings = PostgresSettings()
repo = ItemRepository(settings)
item: Item | None = await repo.get("abc-123")

Configuration

All settings are read from environment variables.

Health checks

PostgresRepository implements the HealthCheck protocol from openframe-core.

alive = await repo.ping()       # SELECT 1 — fast liveness check
ready = await repo.is_ready()   # pg_tables query — full readiness check

Both methods return False on any failure and never raise.

Exception hierarchy

All exceptions are AdapterError subclasses from openframe.core.exceptions. Raw asyncpg exceptions never escape the adapter.

Development

# from the package directory
pip install -e ".[dev]"
pytest tests/ -v

Protocol conformance

from openframe.core.ports import BaseRepository
from openframe.core.health import HealthCheck

repo = PostgresRepository(settings, table="items", id_column="id")
assert isinstance(repo, BaseRepository)   # True — structural check
assert isinstance(repo, HealthCheck)      # True — structural check

No inheritance from either Protocol is required or used.

License

MIT