39 lines
1.1 KiB
Python
39 lines
1.1 KiB
Python
from typing import Generic, TypeVar, Type
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
EntityType = TypeVar("EntityType")
|
|
CreateType = TypeVar("CreateType")
|
|
|
|
|
|
class RepBaseMixin(Generic[EntityType]):
|
|
session: AsyncSession
|
|
|
|
|
|
class RepDeleteMixin(RepBaseMixin[EntityType]):
|
|
async def delete(self, obj: EntityType, is_soft: bool) -> None:
|
|
if not is_soft:
|
|
await self.session.delete(obj)
|
|
await self.session.commit()
|
|
return
|
|
|
|
if not hasattr(obj, "is_deleted"):
|
|
raise AttributeError(
|
|
f"{obj.__class__.__name__} does not support soft delete (missing is_deleted field)"
|
|
)
|
|
obj.is_deleted = True
|
|
self.session.add(obj)
|
|
await self.session.commit()
|
|
|
|
|
|
class RepCreateMixin(Generic[EntityType, CreateType]):
|
|
session: AsyncSession
|
|
entity_class: Type[EntityType]
|
|
|
|
async def create(self, data: CreateType) -> int:
|
|
obj = self.entity_class(**data.model_dump())
|
|
self.session.add(obj)
|
|
await self.session.commit()
|
|
await self.session.refresh(obj)
|
|
return obj.id
|