95 lines
2.9 KiB
Python
95 lines
2.9 KiB
Python
"""Роутер текущего пользователя."""
|
|
from __future__ import annotations
|
|
|
|
from fastapi import APIRouter, Depends, Query, Request
|
|
from sqlmodel import Session
|
|
|
|
from app.auth.deps import get_current_user
|
|
from app.db.session import get_session
|
|
from app.models import User
|
|
from app.schemas import api as s
|
|
from app.services import audit_service, group_service, stats_service, user_service
|
|
|
|
router = APIRouter(prefix="/users", tags=["users"])
|
|
|
|
|
|
def build_me(session: Session, user: User) -> s.MeRead:
|
|
groups = [
|
|
s.GroupBrief(id=g.id, name=g.name, role=role)
|
|
for g, role in group_service.list_user_groups(session, user.id) # type: ignore[arg-type]
|
|
]
|
|
return s.MeRead(
|
|
id=user.id, # type: ignore[arg-type]
|
|
nickname=user.nickname,
|
|
role=user.role,
|
|
auth_provider=user.auth_provider,
|
|
telegram_id=user.telegram_id,
|
|
active_group_id=user.active_group_id,
|
|
groups=groups,
|
|
)
|
|
|
|
|
|
@router.get("/me", response_model=s.MeRead)
|
|
def get_me(
|
|
session: Session = Depends(get_session),
|
|
user: User = Depends(get_current_user),
|
|
) -> s.MeRead:
|
|
return build_me(session, user)
|
|
|
|
|
|
@router.patch("/me", response_model=s.UserRead)
|
|
def update_me(
|
|
body: s.NicknameUpdate,
|
|
request: Request,
|
|
session: Session = Depends(get_session),
|
|
user: User = Depends(get_current_user),
|
|
) -> s.UserRead:
|
|
user_service.update_nickname(session, user, body.nickname)
|
|
audit_service.record(
|
|
session,
|
|
actor_id=user.id,
|
|
action="update",
|
|
entity_type="user",
|
|
entity_id=user.id,
|
|
payload={"nickname": user.nickname},
|
|
ip=request.client.host if request.client else None,
|
|
user_agent=request.headers.get("user-agent"),
|
|
)
|
|
session.commit()
|
|
return s.UserRead(
|
|
id=user.id, # type: ignore[arg-type]
|
|
nickname=user.nickname,
|
|
role=user.role,
|
|
auth_provider=user.auth_provider,
|
|
telegram_id=user.telegram_id,
|
|
active_group_id=user.active_group_id,
|
|
)
|
|
|
|
|
|
@router.get("/me/stats", response_model=s.ProfileStats)
|
|
def my_stats(
|
|
session: Session = Depends(get_session),
|
|
user: User = Depends(get_current_user),
|
|
) -> dict:
|
|
return stats_service.profile_stats(session, user.id) # type: ignore[arg-type]
|
|
|
|
|
|
@router.put("/me/active-group", response_model=s.MeRead)
|
|
def set_active_group(
|
|
body: s.ActiveGroupUpdate,
|
|
session: Session = Depends(get_session),
|
|
user: User = Depends(get_current_user),
|
|
) -> s.MeRead:
|
|
user_service.set_active_group(session, user, body.group_id)
|
|
return build_me(session, user)
|
|
|
|
|
|
@router.get("/nickname-available", response_model=s.NicknameAvailable)
|
|
def nickname_available(
|
|
value: str = Query(..., min_length=1),
|
|
session: Session = Depends(get_session),
|
|
user: User = Depends(get_current_user),
|
|
) -> s.NicknameAvailable:
|
|
available = user_service.nickname_available(session, value.strip(), exclude_user_id=user.id)
|
|
return s.NicknameAvailable(available=available)
|