Files
ForbiddenStarsApp/backend/app/routers/groups.py
T

210 lines
7.7 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,
membership_service,
stats_service,
)
router = APIRouter(prefix="/groups", tags=["groups"])
def _detail(session: Session, group_id: int, user_id: int) -> s.GroupDetail:
group = group_service.get_group(session, group_id)
member = group_service.assert_member(session, group_id, user_id)
return s.GroupDetail(
id=group.id, # type: ignore[arg-type]
name=group.name,
owner_id=group.owner_id,
my_role=member.role,
expansion_ids=group_service.group_expansion_ids(session, group_id),
)
@router.get("", response_model=list[s.GroupBrief])
def my_groups(
session: Session = Depends(get_session),
user: User = Depends(get_current_user),
) -> list[s.GroupBrief]:
return [
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]
]
@router.post("", response_model=s.GroupDetail)
def create_group(
body: s.GroupCreate,
request: Request,
session: Session = Depends(get_session),
user: User = Depends(get_current_user),
) -> s.GroupDetail:
group = group_service.create_group(session, user, body.name, body.expansion_ids)
audit_service.record(
session,
actor_id=user.id,
action="create",
entity_type="group",
entity_id=group.id,
payload={"name": group.name},
ip=request.client.host if request.client else None,
user_agent=request.headers.get("user-agent"),
)
session.commit()
return _detail(session, group.id, user.id) # type: ignore[arg-type]
@router.get("/{group_id}", response_model=s.GroupDetail)
def get_group(
group_id: int,
session: Session = Depends(get_session),
user: User = Depends(get_current_user),
) -> s.GroupDetail:
return _detail(session, group_id, user.id) # type: ignore[arg-type]
@router.patch("/{group_id}", response_model=s.GroupDetail)
def rename_group(
group_id: int,
body: s.GroupRename,
session: Session = Depends(get_session),
user: User = Depends(get_current_user),
) -> s.GroupDetail:
group_service.assert_member(session, group_id, user.id) # type: ignore[arg-type]
group = group_service.get_group(session, group_id)
group_service.rename_group(session, group, body.name)
return _detail(session, group_id, user.id) # type: ignore[arg-type]
@router.put("/{group_id}/expansions", response_model=s.GroupDetail)
def set_expansions(
group_id: int,
body: s.GroupExpansionsUpdate,
session: Session = Depends(get_session),
user: User = Depends(get_current_user),
) -> s.GroupDetail:
group_service.assert_member(session, group_id, user.id) # type: ignore[arg-type]
group = group_service.get_group(session, group_id)
group_service.set_expansions(session, group, body.expansion_ids)
return _detail(session, group_id, user.id) # type: ignore[arg-type]
@router.get("/{group_id}/factions", response_model=list[s.FactionRead])
def group_factions(
group_id: int,
session: Session = Depends(get_session),
user: User = Depends(get_current_user),
) -> list[s.FactionRead]:
group_service.assert_member(session, group_id, user.id) # type: ignore[arg-type]
return [
s.FactionRead(
id=f.id, code=f.code, name_ru=f.name_ru, expansion_id=f.expansion_id # type: ignore[arg-type]
)
for f in group_service.available_factions(session, group_id)
]
# ─── Членство ────────────────────────────────────────────────────────────────
@router.get("/{group_id}/members", response_model=list[s.MemberRead])
def list_members(
group_id: int,
session: Session = Depends(get_session),
user: User = Depends(get_current_user),
) -> list[s.MemberRead]:
group_service.assert_member(session, group_id, user.id) # type: ignore[arg-type]
return [
s.MemberRead(user_id=u.id, nickname=u.nickname, role=m.role)
for m, u in membership_service.list_members(session, group_id)
]
@router.post("/{group_id}/members", response_model=s.MemberRead)
def add_member(
group_id: int,
body: s.MemberAdd,
request: Request,
session: Session = Depends(get_session),
user: User = Depends(get_current_user),
) -> s.MemberRead:
group_service.assert_member(session, group_id, user.id) # type: ignore[arg-type]
group = group_service.get_group(session, group_id)
member, added = membership_service.add_member_by_nickname(session, group, body.nickname)
audit_service.record(
session,
actor_id=user.id,
action="create",
entity_type="group_member",
entity_id=group_id,
payload={"added_user_id": added.id, "nickname": added.nickname},
ip=request.client.host if request.client else None,
user_agent=request.headers.get("user-agent"),
)
session.commit()
return s.MemberRead(user_id=added.id, nickname=added.nickname, role=member.role)
@router.delete("/{group_id}/members/{user_id}", response_model=s.OkResponse)
def remove_member(
group_id: int,
user_id: int,
session: Session = Depends(get_session),
user: User = Depends(get_current_user),
) -> s.OkResponse:
# Любой участник управляет составом; защита создателя — в membership_service
# (нельзя удалить последнего владельца).
group_service.assert_member(session, group_id, user.id) # type: ignore[arg-type]
group = group_service.get_group(session, group_id)
membership_service.remove_member(session, group, user_id)
return s.OkResponse()
@router.patch("/{group_id}/members/{user_id}", response_model=s.MemberRead)
def change_member_role(
group_id: int,
user_id: int,
body: s.MemberRoleUpdate,
session: Session = Depends(get_session),
user: User = Depends(get_current_user),
) -> s.MemberRead:
group_service.assert_owner(session, group_id, user.id) # type: ignore[arg-type]
group = group_service.get_group(session, group_id)
member = membership_service.change_role(session, group, user_id, body.role)
from app.services.user_service import get_user
u = get_user(session, user_id)
return s.MemberRead(user_id=u.id, nickname=u.nickname, role=member.role)
# ─── Партии и статистика группы ──────────────────────────────────────────────
@router.get("/{group_id}/matches", response_model=s.MatchList)
def group_matches(
group_id: int,
limit: int = Query(20, ge=1, le=100),
offset: int = Query(0, ge=0),
session: Session = Depends(get_session),
user: User = Depends(get_current_user),
) -> dict:
group_service.assert_member(session, group_id, user.id) # type: ignore[arg-type]
return stats_service.group_match_list(session, group_id, limit=limit, offset=offset)
@router.get("/{group_id}/stats", response_model=s.GroupStats)
def group_stats(
group_id: int,
session: Session = Depends(get_session),
user: User = Depends(get_current_user),
) -> dict:
group_service.assert_member(session, group_id, user.id) # type: ignore[arg-type]
return stats_service.group_stats(session, group_id)