59 lines
1.5 KiB
Python
59 lines
1.5 KiB
Python
from typing import Sequence
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import select, delete, update
|
|
from ..schemas import UserRead, UserCreate,UserUpdate
|
|
from ..models import UserModel
|
|
|
|
|
|
|
|
async def get_all_users(
|
|
session: AsyncSession
|
|
) -> Sequence[UserModel]:
|
|
stmt = select(UserModel).order_by(UserModel.id)
|
|
result = await session.scalars(stmt)
|
|
return result.all()
|
|
|
|
async def get_user(
|
|
session: AsyncSession,
|
|
user_id: int
|
|
) -> UserModel:
|
|
db_user = await session.get(UserModel, user_id)
|
|
if db_user is None:
|
|
return None
|
|
return db_user
|
|
|
|
async def create_user(
|
|
session:AsyncSession,
|
|
user_create: UserCreate
|
|
) -> UserModel:
|
|
user = UserModel(**user_create.model_dump())
|
|
session.add(user)
|
|
await session.commit()
|
|
return user
|
|
|
|
async def delete_user(
|
|
session: AsyncSession,
|
|
user_id: int
|
|
) -> UserModel:
|
|
db_user = await session.get(UserModel, user_id)
|
|
if db_user is None:
|
|
return None
|
|
stmt = delete(UserModel).filter(UserModel.id == user_id)
|
|
await session.execute(stmt)
|
|
await session.commit()
|
|
return db_user.one_or_none()
|
|
|
|
async def update_user(
|
|
session:AsyncSession,
|
|
user_id: int,
|
|
user_update: UserUpdate
|
|
) -> UserModel:
|
|
db_user = await session.get(UserModel, user_id)
|
|
if db_user is None:
|
|
return None
|
|
for var, value in vars(user_update).items():
|
|
setattr(db_user, var, value) if value else None
|
|
await session.commit()
|
|
await session.refresh(db_user)
|
|
return db_user
|