add task post

src/repository/base.py

@@ -1,37 +1,41 @@
-from typing import Any
+from typing import Any, Generic, Mapping, Sequence, Type, TypeVar
 
-from pydantic import BaseModel
 from sqlalchemy import delete, insert, select
+from sqlalchemy.ext.asyncio import AsyncSession
 
 from src.core.database import Base
 
+ModelType = TypeVar("ModelType", bound=Base)
 
-class BaseRepo:
-    model: type[Base]
 
-    def __init__(self, session):
-        self.session = session
+class BaseRepo(Generic[ModelType]):
+    model: Type[ModelType]
 
-    async def get_filtered(self, *filter, **filter_by) -> list[Base]:
-        query = select(self.model).filter(*filter).filter_by(**filter_by)
+    def __init__(self, session: AsyncSession) -> None:
+        self.session: AsyncSession = session
+
+    async def get_filtered(
+        self, *filters: Any, **filter_by: Any
+    ) -> Sequence[ModelType]:
+        query = select(self.model).filter(*filters).filter_by(**filter_by)
         result = await self.session.execute(query)
         models = result.scalars().all()
         return models
 
-    async def create_one(self, data: BaseModel) -> Base:
-        statement = insert(self.model).values(data.model_dump()).returning(self.model)
+    async def create_one(self, data: Mapping[str, Any]) -> ModelType:
+        statement = insert(self.model).values(data).returning(self.model)
         result = await self.session.execute(statement)
-        obj = result.scalar_one()
+        obj: ModelType = result.scalar_one()
         return obj
 
-    async def get_one_or_none(self, **filter_by: Any) -> Base | None:
+    async def get_one_or_none(self, **filter_by: Any) -> ModelType | None:
         query = select(self.model).filter_by(**filter_by)
         result = await self.session.execute(query)
-        model = result.scalars().one_or_none()
-        return model
+        model_obj: ModelType | None = result.scalars().one_or_none()
+        return model_obj
 
-    async def get_all(self, *args, **kwargs) -> list[Base]:
-        result = await self.get_filtered(*args, **kwargs)
+    async def get_all(self, *args: Any, **kwargs: Any) -> Sequence[ModelType]:
+        result: Sequence[ModelType] = await self.get_filtered(*args, **kwargs)
         return result
 
     async def delete_one(self, **filter_by) -> None: