using-archipy-adapters

Using ArchiPy Adapters

Live: https://syntaxarc.github.io/ArchiPy/tutorials/adapters/

Prefer ArchiPy First

Install the matching extra and use the shipped adapter:

uv add "archipy[redis]"
# examples: postgres, sqlalchemy, sqlalchemy-async, kafka, scylladb, minio, keycloak, ...

from archipy.adapters.redis.adapters import RedisAdapter

redis = RedisAdapter()  # reads BaseConfig.global_config().REDIS

Inject *Port abstractions into logics/repositories — not concrete adapters when DI is used.

Domain Adapter Shape

Domain adapters live under the domain repository — not a top-level adapters/ package:

repositories/<domain>/
├── adapters/
│   └── <domain>_<purpose>_adapter.py   # e.g. user_db_adapter.py
└── <domain>_repository.py

Thin wrappers around ArchiPy adapters (or external clients). Own entity construction and query building. Depend on ArchiPy ports when wrapping library adapters; add a local ABC only when the domain needs a custom contract. BDD runs adapters against real testcontainers; write a port-conforming fake only for a third-party API with no container.

Ports

from abc import ABC, abstractmethod

class CachePort(ABC):
    @abstractmethod
    def get(self, key: str) -> str | None: ...

    @abstractmethod
    def set(self, key: str, value: str, ttl: int | None = None) -> None: ...

Implementations and fakes must satisfy the same port.

Boundary Errors

Catch specific client/driver errors and map to domain errors with raise ... from e. A broad except Exception is allowed only at the outermost infrastructure boundary when the client exposes no typed root error; add a narrow # noqa: BLE001 reason. Never use bare except::

try:
    await self._client.produce(topic, message)
except KafkaError as e:
    raise KafkaProduceError("Failed to produce message") from e

Never leak raw driver exceptions into logics/services.

Partial Indexes and Bound Parameters

asyncpg runs every query as a server-side prepared statement (psycopg 3 does once a query has run more than its prepare_threshold times, 5 by default). Postgres plans the first five executions of a prepared statement with the actual values, then may switch to a generic plan, which sees $1 instead of the value (plan_cache_mode = auto; see the PostgreSQL docs). With asyncpg this happens from about the sixth execution on a connection; with psycopg 3 it starts later, after the statement is prepared. A partial index (CREATE INDEX ... WHERE status = 'PENDING') is usable only when the planner can prove its predicate, so a filter that binds the predicate's column (field == value, field.in_(values), or SQLAlchemyFilterMixin._apply_filter) stops matching the index once the plan goes generic. The query falls back to a broader index or a full scan. Nothing errors, and a small development database rarely shows it.