Cheng Zhou
73 lines
2.1 KiB
Python
73 lines
2.1 KiB
Python
import uuid
|
|
from typing import Sequence
|
|
|
|
from sqlalchemy import select, update
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.security import hash_password
|
|
from app.models.user import User
|
|
from app.schemas.user import UserCreate, UserUpdate
|
|
|
|
|
|
async def get_by_username(db: AsyncSession, username: str) -> User | None:
|
|
result = await db.execute(select(User).where(User.username == username))
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def get_by_id(db: AsyncSession, user_id: uuid.UUID) -> User | None:
|
|
result = await db.execute(select(User).where(User.id == user_id))
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def create_user(db: AsyncSession, user_in: UserCreate) -> User:
|
|
user = User(
|
|
username=user_in.username,
|
|
hashed_password=hash_password(user_in.password),
|
|
role=user_in.role,
|
|
is_active=True,
|
|
)
|
|
db.add(user)
|
|
await db.commit()
|
|
await db.refresh(user)
|
|
return user
|
|
|
|
|
|
async def update_user(db: AsyncSession, user: User, user_in: UserUpdate) -> User:
|
|
update_data = {}
|
|
if user_in.role is not None:
|
|
update_data["role"] = user_in.role
|
|
if user_in.is_active is not None:
|
|
update_data["is_active"] = user_in.is_active
|
|
if user_in.password:
|
|
update_data["hashed_password"] = hash_password(user_in.password)
|
|
|
|
if update_data:
|
|
await db.execute(
|
|
update(User)
|
|
.where(User.id == user.id)
|
|
.values(**update_data)
|
|
)
|
|
await db.commit()
|
|
await db.refresh(user)
|
|
return user
|
|
|
|
|
|
async def list_users(db: AsyncSession, skip: int = 0, limit: int = 100) -> Sequence[User]:
|
|
result = await db.execute(select(User).offset(skip).limit(limit))
|
|
return result.scalars().all()
|
|
|
|
|
|
async def ensure_admin_exists(db: AsyncSession, *, default_password: str = "admin123") -> None:
|
|
result = await db.execute(select(User).where(User.username == "admin"))
|
|
admin = result.scalar_one_or_none()
|
|
if admin:
|
|
return
|
|
new_admin = User(
|
|
username="admin",
|
|
hashed_password=hash_password(default_password),
|
|
role="ADMIN",
|
|
is_active=True,
|
|
)
|
|
db.add(new_admin)
|
|
await db.commit()
|