Class BaseRepository

src/core/db/repository.py:114–392

class BaseRepository(ABC, Generic[T]):
    def __init__(self, db_config: MyDatabaseConfig, model: type[T], id_field: str = "id"):
        self._db_config = db_config
        self._model = model
        self._id_field = id_field

    async def get_by_id(
        self,
        uid: Any,
    ) -> T:
        obj = await self.find_by_id(uid)
        if obj is None:
            raise DomainException(
                error_no=ErrorNo.REPOSITORY_DATA_BY_ID_NOT_FOUND,
                message=f"{self._model.__name__} with {self._id_field}={uid} not found",
            )
        return obj

    async def find_by_id(
        self,
        uid: Any,
    ) -> T | None:
        query = select(self._model).where(getattr(self._model, self._id_field) == uid)
        async with self.get_session() as session:
            result = await session.execute(query)
            return result.scalar_one_or_none()

    async def create(self, data: dict[str, Any] | T) -> T:
        if isinstance(data, dict):
            entity = self._model(**data)
        else:
            entity = data
        async with self.get_session() as session:
            session.add(entity)
            await session.flush()
            await session.refresh(entity)
            return entity

    async def update(
        self,
        uid: Any,
        data: dict[str, Any] | T,
    ) -> T:
        if isinstance(data, dict):
            d = data
        else:
            d = data.__dict__

        entity = await self.get_by_id(uid)

        for field, value in d.items():
            if hasattr(entity, field):
                setattr(entity, field, value)
        async with self.get_session() as session:
            session.add(entity)
            await session.flush()
            await session.refresh(entity)
            return entity