64 lines
2.2 KiB
Python
64 lines
2.2 KiB
Python
"""Abstract repository contracts.
|
|
|
|
`Repository[T]` is the plain CRUD base. Use it for simple entities like
|
|
VacationDays that have no GDPR deletion requirements.
|
|
|
|
`SoftDeleteRepository[T]` extends it for entities that need an audit trail
|
|
(soft-delete) plus a GDPR hard delete. Use it for PersonalData and any other
|
|
entity that directly holds PII. The `list()` contract on soft-deletable
|
|
entities should always filter `WHERE deleted_at IS NULL` by default.
|
|
|
|
Dependency direction: routers/services depend on these interfaces only —
|
|
never on a concrete DB class.
|
|
"""
|
|
|
|
from abc import ABC, abstractmethod
|
|
from typing import Generic, TypeVar
|
|
from uuid import UUID
|
|
|
|
T = TypeVar("T")
|
|
|
|
|
|
class Repository(ABC, Generic[T]):
|
|
@abstractmethod
|
|
async def create(self, item: T) -> T:
|
|
"""Persist a new item and return it."""
|
|
|
|
@abstractmethod
|
|
async def get_by_id(self, id: UUID) -> T | None:
|
|
"""Return the item or None if it does not exist."""
|
|
|
|
@abstractmethod
|
|
async def update(self, id: UUID, item: T) -> T:
|
|
"""Update an existing item. Raise NotFoundError if missing."""
|
|
|
|
@abstractmethod
|
|
async def delete(self, id: UUID) -> None:
|
|
"""Remove an item. Raise NotFoundError if missing."""
|
|
|
|
@abstractmethod
|
|
async def list(self, **filters: object) -> list[T]:
|
|
"""Return items matching the given filters."""
|
|
|
|
|
|
class SoftDeleteRepository(Repository[T], Generic[T]):
|
|
"""Extended contract for GDPR-sensitive entities.
|
|
|
|
`delete()` on these entities sets `deleted_at` (soft delete for audit trail);
|
|
`hard_delete()` removes the row entirely (GDPR erasure). Implementations of
|
|
`list()` must filter `WHERE deleted_at IS NULL` unless `include_deleted=True`
|
|
is passed.
|
|
"""
|
|
|
|
@abstractmethod
|
|
async def soft_delete(self, id: UUID) -> None:
|
|
"""Set deleted_at to now. Raise NotFoundError if missing."""
|
|
|
|
@abstractmethod
|
|
async def hard_delete(self, id: UUID) -> None:
|
|
"""Permanently remove the row (GDPR erasure). Raise NotFoundError if missing."""
|
|
|
|
async def delete(self, id: UUID) -> None:
|
|
"""Default: calls soft_delete. Override only if hard delete is the right default."""
|
|
await self.soft_delete(id)
|