44 lines
1.4 KiB
Python
44 lines
1.4 KiB
Python
from typing import Optional, List
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.user import User
|
|
|
|
|
|
class UserRepository:
|
|
def __init__(self, db: AsyncSession):
|
|
self.db = db
|
|
|
|
async def get_by_id(self, user_id: int) -> Optional[User]:
|
|
result = await self.db.execute(select(User).where(User.id == user_id))
|
|
return result.scalar_one_or_none()
|
|
|
|
async def get_by_username(self, username: str) -> Optional[User]:
|
|
result = await self.db.execute(select(User).where(User.username == username))
|
|
return result.scalar_one_or_none()
|
|
|
|
async def get_all(self) -> List[User]:
|
|
result = await self.db.execute(select(User))
|
|
return list(result.scalars().all())
|
|
|
|
async def create(self, data: dict) -> User:
|
|
user = User(**data)
|
|
self.db.add(user)
|
|
await self.db.flush()
|
|
return user
|
|
|
|
async def update(self, user_id: int, data: dict) -> Optional[User]:
|
|
user = await self.get_by_id(user_id)
|
|
if not user:
|
|
return None
|
|
for key, value in data.items():
|
|
setattr(user, key, value)
|
|
await self.db.flush()
|
|
return user
|
|
|
|
async def delete(self, user_id: int) -> bool:
|
|
user = await self.get_by_id(user_id)
|
|
if not user:
|
|
return False
|
|
await self.db.delete(user)
|
|
return True |