137 lines
4.0 KiB
Python
137 lines
4.0 KiB
Python
from sqlalchemy import func, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.security import hash_password, make_unusable_password, verify_password
|
|
from app.models.role import Role
|
|
from app.models.user import User
|
|
|
|
|
|
async def get(db: AsyncSession, user_id: int) -> User | None:
|
|
return await db.get(User, user_id)
|
|
|
|
|
|
async def get_by_email(db: AsyncSession, email: str) -> User | None:
|
|
result = await db.execute(select(User).where(User.email == email))
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def list_users(db: AsyncSession, skip: int = 0, limit: int = 100) -> list[User]:
|
|
result = await db.execute(select(User).order_by(User.id).offset(skip).limit(limit))
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def count(db: AsyncSession) -> int:
|
|
result = await db.execute(select(func.count()).select_from(User))
|
|
return result.scalar_one()
|
|
|
|
|
|
async def _resolve_roles(db: AsyncSession, role_ids: list[int]) -> list[Role]:
|
|
if not role_ids:
|
|
return []
|
|
result = await db.execute(select(Role).where(Role.id.in_(role_ids)))
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def _resolve_roles_by_name(db: AsyncSession, names: list[str]) -> list[Role]:
|
|
if not names:
|
|
return []
|
|
result = await db.execute(select(Role).where(Role.name.in_(names)))
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def create(
|
|
db: AsyncSession,
|
|
*,
|
|
email: str,
|
|
password: str,
|
|
full_name: str | None = None,
|
|
is_superuser: bool = False,
|
|
role_ids: list[int] | None = None,
|
|
auth_provider: str = "local",
|
|
) -> User:
|
|
user = User(
|
|
email=email,
|
|
hashed_password=hash_password(password),
|
|
full_name=full_name,
|
|
is_superuser=is_superuser,
|
|
auth_provider=auth_provider,
|
|
roles=await _resolve_roles(db, role_ids or []),
|
|
)
|
|
db.add(user)
|
|
await db.commit()
|
|
await db.refresh(user)
|
|
return user
|
|
|
|
|
|
async def upsert_ldap_user(
|
|
db: AsyncSession,
|
|
*,
|
|
email: str,
|
|
full_name: str | None = None,
|
|
default_role_names: list[str] | None = None,
|
|
) -> User:
|
|
"""Provision (or refresh) the local row for an LDAP-authenticated user.
|
|
|
|
On first login the user is created with an unusable local password (so
|
|
password auth can never succeed) and the configured default roles. On later
|
|
logins the existing row is reused; only ``full_name`` is synced from the
|
|
directory. Roles assigned afterwards in the admin UI are preserved.
|
|
"""
|
|
user = await get_by_email(db, email)
|
|
if user is None:
|
|
user = User(
|
|
email=email,
|
|
hashed_password=make_unusable_password(),
|
|
full_name=full_name,
|
|
auth_provider="ldap",
|
|
roles=await _resolve_roles_by_name(db, default_role_names or []),
|
|
)
|
|
db.add(user)
|
|
await db.commit()
|
|
await db.refresh(user)
|
|
return user
|
|
|
|
changed = False
|
|
if user.auth_provider != "ldap":
|
|
user.auth_provider = "ldap"
|
|
changed = True
|
|
if full_name and user.full_name != full_name:
|
|
user.full_name = full_name
|
|
changed = True
|
|
if changed:
|
|
await db.commit()
|
|
await db.refresh(user)
|
|
return user
|
|
|
|
|
|
async def update(db: AsyncSession, user: User, data: dict) -> User:
|
|
if "password" in data and data["password"]:
|
|
user.hashed_password = hash_password(data.pop("password"))
|
|
else:
|
|
data.pop("password", None)
|
|
|
|
if "role_ids" in data and data["role_ids"] is not None:
|
|
user.roles = await _resolve_roles(db, data.pop("role_ids"))
|
|
else:
|
|
data.pop("role_ids", None)
|
|
|
|
for field, value in data.items():
|
|
if value is not None:
|
|
setattr(user, field, value)
|
|
|
|
await db.commit()
|
|
await db.refresh(user)
|
|
return user
|
|
|
|
|
|
async def delete(db: AsyncSession, user: User) -> None:
|
|
await db.delete(user)
|
|
await db.commit()
|
|
|
|
|
|
async def authenticate(db: AsyncSession, email: str, password: str) -> User | None:
|
|
user = await get_by_email(db, email)
|
|
if not user or not verify_password(password, user.hashed_password):
|
|
return None
|
|
return user
|