CRM Bot
120 lines
3.4 KiB
Python
120 lines
3.4 KiB
Python
"""User service: read, update, soft-delete, list."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.security import hash_password
|
|
from app.models.user import User, UserRole
|
|
from app.schemas.user import UserCreateRequest, UserUpdate
|
|
|
|
|
|
class UserNotFound(Exception):
|
|
"""Raised when a user lookup fails."""
|
|
|
|
|
|
class EmailAlreadyTaken(Exception):
|
|
"""Raised when attempting to create/update a user with an existing email."""
|
|
|
|
|
|
async def get_user_by_id(db: AsyncSession, user_id: int) -> User | None:
|
|
"""Fetch a user by ID (active only, soft-deleted excluded)."""
|
|
result = await db.execute(
|
|
select(User).where(User.id == user_id, User.deleted_at.is_(None))
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def get_user_by_email(
|
|
db: AsyncSession, email: str, org_id: int | None = None
|
|
) -> User | None:
|
|
"""Fetch a user by email, optionally scoped to an org."""
|
|
stmt = select(User).where(
|
|
User.email == email.lower(),
|
|
User.deleted_at.is_(None),
|
|
)
|
|
if org_id is not None:
|
|
stmt = stmt.where(User.org_id == org_id)
|
|
result = await db.execute(stmt)
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def update_user_profile(
|
|
db: AsyncSession, user: User, payload: UserUpdate, *, is_admin: bool = False
|
|
) -> User:
|
|
"""Apply partial updates to a user.
|
|
|
|
Non-admin callers cannot change the role. All other fields are optional.
|
|
"""
|
|
data = payload.model_dump(exclude_unset=True)
|
|
|
|
if "role" in data and not is_admin:
|
|
# Silently drop role change for non-admin callers
|
|
data.pop("role")
|
|
|
|
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 soft_delete_user(db: AsyncSession, user: User) -> User:
|
|
"""Soft-delete a user by setting deleted_at to now."""
|
|
user.deleted_at = datetime.now(UTC)
|
|
await db.commit()
|
|
await db.refresh(user)
|
|
return user
|
|
|
|
|
|
async def create_user_as_admin(
|
|
db: AsyncSession, payload: UserCreateRequest, org_id: int
|
|
) -> User:
|
|
"""Create a new user in the given org (admin-only flow)."""
|
|
existing = await get_user_by_email(db, payload.email, org_id=org_id)
|
|
if existing is not None:
|
|
raise EmailAlreadyTaken(
|
|
f"A user with email {payload.email!r} already exists in this org."
|
|
)
|
|
|
|
user = User(
|
|
org_id=org_id,
|
|
email=payload.email.lower(),
|
|
password_hash=hash_password(payload.password),
|
|
name=payload.name,
|
|
role=payload.role if payload.role else UserRole.sales_rep,
|
|
)
|
|
db.add(user)
|
|
await db.commit()
|
|
await db.refresh(user)
|
|
return user
|
|
|
|
|
|
async def list_users(
|
|
db: AsyncSession, org_id: int, skip: int = 0, limit: int = 50
|
|
) -> list[User]:
|
|
"""List active users in an org, paginated."""
|
|
result = await db.execute(
|
|
select(User)
|
|
.where(User.org_id == org_id, User.deleted_at.is_(None))
|
|
.order_by(User.id)
|
|
.offset(skip)
|
|
.limit(limit)
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def count_users_in_org(db: AsyncSession, org_id: int) -> int:
|
|
"""Count active users in an org."""
|
|
result = await db.execute(
|
|
select(func.count())
|
|
.select_from(User)
|
|
.where(User.org_id == org_id, User.deleted_at.is_(None))
|
|
)
|
|
return int(result.scalar_one())
|