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. Optional in-memory mock only if BDD needs it (Redis ships RedisMock).

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 mocks must satisfy the same port.

Boundary Errors

Catch specific client/driver errors and map to domain errors with raise ... from e. Never bare except Exception unless the client exposes no typed root error:

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.

Sync vs Async

• Sync and async are separate classes — do not mix async def into a sync adapter class.
• Prefer the ArchiPy async extra when the app is async end-to-end.

Config

Adapters should read settings from BaseConfig.global_config() (or injected config), not from scattered env reads inside methods.