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