42 lines
1.3 KiB
Python
42 lines
1.3 KiB
Python
from fastapi import HTTPException
|
|
from sqlalchemy import insert, select
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.ext.asyncio.session import AsyncSession
|
|
from starlette.types import HTTPExceptionHandler
|
|
|
|
from ..models import UserModel
|
|
from ..schemas.user_schema import UserReadDTO, UserWriteDTO
|
|
|
|
|
|
class UserRepository:
|
|
def __init__(self, session: AsyncSession):
|
|
self.session = session
|
|
|
|
async def add_one(
|
|
self,
|
|
data: UserWriteDTO,
|
|
) -> UserReadDTO:
|
|
stmt = insert(UserModel).values(**data.model_dump()).returning(UserModel)
|
|
res = await self.session.execute(stmt)
|
|
return UserReadDTO.model_validate(res.scalar_one())
|
|
|
|
async def find_all(self) -> list[UserReadDTO]:
|
|
stmt = select(UserModel)
|
|
res = await self.session.execute(stmt)
|
|
res = [UserReadDTO.model_validate(row) for row in res.scalars().all()]
|
|
return res
|
|
|
|
async def find_one(self, filter: dict) -> UserReadDTO | None:
|
|
stmt = select(UserModel).filter_by(**filter)
|
|
res = await self.session.execute(stmt)
|
|
res = res.scalar_one_or_none()
|
|
if res is not None:
|
|
return UserReadDTO.model_validate(res)
|
|
return None
|
|
|
|
async def update_one(self, filter: dict, data: dict):
|
|
return
|
|
|
|
async def delete_one(self, filter: dict):
|
|
return
|