integrate advanced alchemy (#7)

Author: lesnik512

app/api/decks.py

@@ -1,27 +1,36 @@
 import typing
 
 import litestar
+from advanced_alchemy.exceptions import NotFoundError
 from litestar import status_codes
 from litestar.contrib.pydantic import PydanticDTO
 from litestar.exceptions import HTTPException
+from sqlalchemy import orm
 from that_depends.providers import container_context
 
 from app import ioc, models, schemas
 
 
+if typing.TYPE_CHECKING:
+    from app.repositories import CardsService, DecksService
+
+
 @litestar.get("/decks/")
 @container_context()
 async def list_decks() -> schemas.Decks:
-    decks_repo = await ioc.IOCContainer.decks_repo()
-    objects = await decks_repo.all()
+    decks_service: DecksService = await ioc.IOCContainer.decks_service()
+    objects = await decks_service.list()
     return schemas.Decks(items=objects)  # type: ignore[arg-type]
 
 
 @litestar.get("/decks/{deck_id:int}/")
 @container_context()
 async def get_deck(deck_id: int) -> schemas.Deck:
-    decks_repo = await ioc.IOCContainer.decks_repo()
-    instance = await decks_repo.get_by_id(deck_id, prefetch=("cards",))
+    decks_service: DecksService = await ioc.IOCContainer.decks_service()
+    instance = await decks_service.get_one_or_none(
+        models.Deck.id == deck_id,
+        load=[orm.selectinload(models.Deck.cards)],
+    )
     if not instance:
         raise HTTPException(status_code=status_codes.HTTP_404_NOT_FOUND, detail="Deck is not found")
 
@@ -34,38 +43,35 @@ async def update_deck(
     deck_id: int,
     data: schemas.DeckCreate,
 ) -> schemas.Deck:
-    decks_repo = await ioc.IOCContainer.decks_repo()
-    instance = await decks_repo.get_by_id(deck_id)
-    if not instance:
-        raise HTTPException(status_code=status_codes.HTTP_404_NOT_FOUND, detail="Deck is not found")
-
-    await decks_repo.update_attrs(instance, **data.model_dump())
-    await decks_repo.save(instance)
+    decks_service: DecksService = await ioc.IOCContainer.decks_service()
+    try:
+        instance = await decks_service.update(data=data.model_dump(), item_id=deck_id)
+    except NotFoundError:
+        raise HTTPException(status_code=status_codes.HTTP_404_NOT_FOUND, detail="Deck is not found") from None
     return schemas.Deck.model_validate(instance)
 
 
 @litestar.post("/decks/")
 @container_context()
 async def create_deck(data: schemas.DeckCreate) -> schemas.Deck:
-    decks_repo = await ioc.IOCContainer.decks_repo()
-    instance = models.Deck(**data.model_dump())
-    await decks_repo.save(instance)
+    decks_service: DecksService = await ioc.IOCContainer.decks_service()
+    instance = await decks_service.create(data)
     return schemas.Deck.model_validate(instance)
 
 
 @litestar.get("/decks/{deck_id:int}/cards/")
 @container_context()
 async def list_cards(deck_id: int) -> schemas.Cards:
-    cards_repo = await ioc.IOCContainer.cards_repo()
-    objects = await cards_repo.filter({"deck_id": deck_id})
+    cards_service: CardsService = await ioc.IOCContainer.cards_service()
+    objects = await cards_service.list(models.Card.deck_id == deck_id)
     return schemas.Cards(items=objects)  # type: ignore[arg-type]
 
 
 @litestar.get("/cards/{card_id:int}/", return_dto=PydanticDTO[schemas.Card])
 @container_context()
 async def get_card(card_id: int) -> schemas.Card:
-    cards_repo = await ioc.IOCContainer.cards_repo()
-    instance = await cards_repo.get_by_id(card_id)
+    cards_service: CardsService = await ioc.IOCContainer.cards_service()
+    instance = await cards_service.get_one_or_none(models.Card.id == card_id)
     if not instance:
         raise HTTPException(status_code=status_codes.HTTP_404_NOT_FOUND, detail="Card is not found")
     return schemas.Card.model_validate(instance)
@@ -77,9 +83,9 @@ async def create_cards(
     deck_id: int,
     data: list[schemas.CardCreate],
 ) -> schemas.Cards:
-    cards_repo = await ioc.IOCContainer.cards_repo()
-    objects = await cards_repo.bulk_create(
-        [models.Card(**card.model_dump(), deck_id=deck_id) for card in data],
+    cards_service: CardsService = await ioc.IOCContainer.cards_service()
+    objects = await cards_service.create_many(
+        data=[models.Card(**card.model_dump(), deck_id=deck_id) for card in data],
     )
     return schemas.Cards(items=objects)  # type: ignore[arg-type]
 
@@ -90,9 +96,9 @@ async def update_cards(
     deck_id: int,
     data: list[schemas.Card],
 ) -> schemas.Cards:
-    cards_repo = await ioc.IOCContainer.cards_repo()
-    objects = await cards_repo.bulk_update(
-        [models.Card(**card.model_dump(exclude={"deck_id"}), deck_id=deck_id) for card in data],
+    cards_service: CardsService = await ioc.IOCContainer.cards_service()
+    objects = await cards_service.upsert_many(
+        data=[models.Card(**card.model_dump(exclude={"deck_id"}), deck_id=deck_id) for card in data],
     )
     return schemas.Cards(items=objects)  # type: ignore[arg-type]
 

app/application.py

@@ -2,10 +2,10 @@
 import typing
 
 import litestar
+from advanced_alchemy.exceptions import ForeignKeyError
 
 from app import exceptions, ioc
 from app.api.decks import ROUTER
-from app.exceptions import DatabaseValidationError
 
 
 class AppBuilder:
@@ -14,7 +14,7 @@ def __init__(self) -> None:
         self.app: litestar.Litestar = litestar.Litestar(
             debug=self.settings.debug,
             lifespan=[self.lifespan_manager],
-            exception_handlers={DatabaseValidationError: exceptions.database_validation_exception_handler},
+            exception_handlers={ForeignKeyError: exceptions.database_validation_exception_handler},
             route_handlers=[ROUTER],
         )
 

app/db/helpers.py

@@ -1,27 +1,16 @@
 import typing
 
 import litestar
+from advanced_alchemy.exceptions import ForeignKeyError
 from litestar import status_codes
 
 
-class DatabaseError(Exception):
-    pass
-
-
-class DatabaseValidationError(DatabaseError):
-    def __init__(self, message: str, field: str | None = None) -> None:
-        self.message = message
-        self.field = field
-
-
-def database_validation_exception_handler(
-    _: object, exc: DatabaseValidationError
-) -> litestar.Response[dict[str, typing.Any]]:
+def database_validation_exception_handler(_: object, exc: ForeignKeyError) -> litestar.Response[dict[str, typing.Any]]:
     return litestar.Response(
         media_type=litestar.MediaType.JSON,
         content={
             "detail": "Database validation failed",
-            "extra": [{"message": exc.message, "key": exc.field or "__root__"}],
+            "extra": [{"message": exc.detail, "key": "__root__"}],
         },
         status_code=status_codes.HTTP_400_BAD_REQUEST,
     )

app/exceptions.py

@@ -1,7 +1,7 @@
 from that_depends import BaseContainer, providers
 
-from app.db.resource import create_sa_engine, create_session
-from app.repositories.decks import CardsRepository, DecksRepository
+from app import repositories
+from app.resources.db import create_sa_engine, create_session
 from app.settings import Settings
 
 
@@ -11,5 +11,5 @@ class IOCContainer(BaseContainer):
     database_engine = providers.Resource(create_sa_engine, settings=settings.cast)
     session = providers.ContextResource(create_session, engine=database_engine.cast)
 
-    decks_repo = providers.Factory(DecksRepository, session=session)
-    cards_repo = providers.Factory(CardsRepository, session=session)
+    decks_service = providers.Factory(repositories.DecksService, session=session)
+    cards_service = providers.Factory(repositories.CardsService, session=session)

app/helpers/__init__.py

@@ -1,53 +1,23 @@
-import datetime
-import logging
 import typing
 
 import sqlalchemy as sa
+from advanced_alchemy.base import BigIntAuditBase
 from sqlalchemy import orm
 
-from app.helpers.datetime import generate_utc_dt
-
-
-logger = logging.getLogger(__name__)
-
 
 METADATA: typing.Final = sa.MetaData()
+orm.DeclarativeBase.metadata = METADATA
 
 
-class Base(orm.DeclarativeBase):
-    metadata = METADATA
-
-
-class BaseModel(Base):
-    __abstract__ = True
-
-    id: orm.Mapped[typing.Annotated[int, orm.mapped_column(primary_key=True)]]
-    created_at: orm.Mapped[
-        typing.Annotated[
-            datetime.datetime,
-            orm.mapped_column(sa.DateTime(timezone=True), default=generate_utc_dt, nullable=False),
-        ]
-    ]
-    updated_at: orm.Mapped[
-        typing.Annotated[
-            datetime.datetime,
-            orm.mapped_column(sa.DateTime(timezone=True), default=generate_utc_dt, nullable=False),
-        ]
-    ]
-
-    def __str__(self) -> str:
-        return f"<{type(self).__name__}({self.id=})>"
-
-
-class Deck(BaseModel):
+class Deck(BigIntAuditBase):
     __tablename__ = "decks"
 
     name: orm.Mapped[str] = orm.mapped_column(sa.String, nullable=False)
     description: orm.Mapped[str | None] = orm.mapped_column(sa.String, nullable=True)
     cards: orm.Mapped[list["Card"]] = orm.relationship("Card", lazy="noload", uselist=True)
 
 
-class Card(BaseModel):
+class Card(BigIntAuditBase):
     __tablename__ = "cards"
     __table_args__ = (sa.UniqueConstraint("deck_id", "front", name="card_deck_id_front_uc"),)
 

app/helpers/datetime.py

@@ -0,0 +1,20 @@
+from advanced_alchemy.repository import SQLAlchemyAsyncRepository
+from advanced_alchemy.service import SQLAlchemyAsyncRepositoryService
+
+from app import models
+
+
+class DecksRepository(SQLAlchemyAsyncRepository[models.Deck]):
+    model_type = models.Deck
+
+
+class DecksService(SQLAlchemyAsyncRepositoryService[models.Deck]):
+    repository_type = DecksRepository
+
+
+class CardsRepository(SQLAlchemyAsyncRepository[models.Card]):
+    model_type = models.Card
+
+
+class CardsService(SQLAlchemyAsyncRepositoryService[models.Card]):
+    repository_type = CardsRepository