Skip to content

code-patterns.md

Latest commit

 

History

History
117 lines (83 loc) · 3.84 KB

code-patterns.md

File metadata and controls

117 lines (83 loc) · 3.84 KB
Raw

Code Patterns Reference

Copy-paste starting points, each pointing at the real file in this template.

1. Entity — domain/entities/article.py

Frozen dataclass on BaseEntity with create() / from_dict() / to_dict() / validate() and immutable mark_* transitions:

@dataclass(eq=False, frozen=True)
class ArticleEntity(BaseEntity):
    title: str
    status: ArticleStatus
    published_at: Optional[datetime] = None

    def mark_published(self, now=None) -> "ArticleEntity":
        ts = generate_timestamp(now)
        return replace(self, status=ArticleStatus.PUBLISHED, published_at=ts, updated_at=ts)

2. Value object & enum — domain/value_objects/article.py

class ArticleStatus(StrEnum):
    DRAFT = "draft"
    PUBLISHED = "published"

@dataclass(frozen=True)
class AuthorVO(BaseVO):
    author_id: str
    display_name: str
    # create() validates invariants; from_dict()/to_dict() serialize.

3. Domain exception — domain/exceptions/article.py

class ArticleError(DomainError): ...
class ArticleNotFoundError(ArticleError, EntityNotFoundError): ...

4. Repository port — domain/ports/article.py

class ArticleRepository(ABC):
    @abstractmethod
    async def get_by_id(self, article_id: str) -> Optional[ArticleEntity]: ...
    @abstractmethod
    async def get_by_ids(self, article_ids: list[str]) -> list[ArticleEntity]: ...  # batch, no N+1

5. Collection adapter — adapters/mongodb/collections/article_adapter.py

Low-level, session-aware driver calls only; never touches entities.

6. Repository implementation — adapters/repositories/mongodb/article.py

Entity ⇄ document via MongoIdManager; the adapter does the raw I/O.

async def get_by_id(self, article_id):
    doc = await self._adapter.find_one({"_id": ObjectId(article_id)})
    return ArticleEntity.from_dict(MongoIdManager.from_document(doc)) if doc else None

7. Application service — service_layer/application/article.py

Read (repo direct), single-CUD (repo direct), and multi-CUD (UoW callback):

async def publish_article(self, article_id):
    article = await self._article_repo.get_by_id(article_id)   # read
    published = article.mark_published()
    event = DomainEventEntity.create(EventType.ARTICLE_PUBLISHED, article_id, {...})

    async def _transaction(uow):                                # multi-CUD
        await uow.article_repo.update(published)
        await uow.domain_event_repo.append(event)

    await self._uow_factory().commit(_transaction)
    return ArticleDTO.from_entity(published)

8. API route — entrypoints/api/routes/article.py

Thin: validate → one service call → translate exceptions → response. Custom types wrapped in nested Annotated for Path/Query.

9. Schema — entrypoints/api/schemas/

Requests validate input; responses expose from_dto(). error_responses(...) documents detail codes for OpenAPI. Custom types + validators in common.py.

10. Dependency injection — entrypoints/api/dependencies/

config → clients → repositories → services. Each get_*_service receives resolved deps via Depends and only assembles.

11. Worker task — entrypoints/worker/tasks/article/notify_published.py

@task
async def notify_published(data: dict[str, Any]) -> None:
    service = WorkerDependencies.get_article_service()   # assemble
    await service.notify_published(data["article_id"])   # call

12. CLI job — entrypoints/cli/jobs/article/backfill_slugs.py

@job + Click option auto-generation from the signature; goes through a service, never a repository.

13. Registering a new entity in the UoW — adapters/uow/mongo_unit_of_work.py

  1. Declare the attribute on AbstractUnitOfWork (under TYPE_CHECKING).
  2. Instantiate its adapter + repository in MongoUnitOfWork._bind_session.