from typing import Sequence from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select, delete, update from ..schemas import LoginRead, LoginCreate, LoginUpdate from ..models import LoginModel from utils.hashing import get_password_hash async def get_all_logins( session: AsyncSession ) -> Sequence[LoginModel]: stmt = select(LoginModel).order_by(LoginModel.id) result = await session.scalars(stmt) return result.all() async def get_login( session: AsyncSession, login_id: int ) -> LoginModel: db_login = await session.get(LoginModel,login_id) if db_login is None: return None return db_login async def create_login( session:AsyncSession, login_create: LoginCreate ) -> LoginModel: login = LoginModel(**login_create.model_dump()) login.password = get_password_hash(login.password) session.add(login) await session.commit() return login async def delete_login( session: AsyncSession, login_id: int ): db_login = await session.get(LoginModel,login_id) if db_login is None: return None stmt = delete(LoginModel).filter(LoginModel.id == login_id) await session.execute(stmt) await session.commit() async def update_login( session:AsyncSession, login_id: int, login_update: LoginUpdate ) -> LoginModel: db_login = await session.get(LoginModel,login_id) if db_login is None: return None for var, value in vars(login_update).items(): setattr(db_login, var, value if var != "password" else get_password_hash(value)) if value else None await session.commit() return db_login