24 lines
690 B
Python
24 lines
690 B
Python
from pydantic import BaseModel
|
|
from sqlalchemy import insert, select
|
|
|
|
from src.core.database import Base
|
|
|
|
|
|
class BaseRepo:
|
|
model: type[Base] = None
|
|
|
|
def __init__(self, session):
|
|
self.session = session
|
|
|
|
async def create_one(self, data: BaseModel):
|
|
statement = insert(self.model).values(data.model_dump()).returning(self.model)
|
|
result = await self.session.execute(statement)
|
|
obj = result.scalar_one()
|
|
return obj
|
|
|
|
async def get_one_or_none(self, **filter_by):
|
|
query = select(self.model).filter_by(**filter_by)
|
|
result = await self.session.execute(query)
|
|
model = result.scalars().one_or_none()
|
|
return model
|