24 lines
1 KiB
Python
24 lines
1 KiB
Python
from sqlalchemy import select, insert
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
|
||
class BaseDAO:
|
||
model = None
|
||
|
||
def __init__(self, session: AsyncSession):
|
||
self.session = session
|
||
|
||
async def add(self, **data): # Метод добавляет данные в БД
|
||
stmt = insert(self.model).values(**data).returning(self.model.id)
|
||
result = await self.session.execute(stmt)
|
||
return result.scalar()
|
||
|
||
async def find_one_or_none(self, **filter_by): # Метод проверяет наличие строки с заданными параметрами
|
||
query = select(self.model).filter_by(**filter_by)
|
||
result = await self.session.execute(query)
|
||
return result.scalar_one_or_none()
|
||
|
||
async def find_all(self, **filter_by): # Метод возвращает все строки таблицы или те, которые соответствуют отбору
|
||
query = select(self.model.__table__.columns).filter_by(**filter_by)
|
||
result = await self.session.execute(query)
|
||
return result.mappings().all()
|