134 lines
5.8 KiB
Python
134 lines
5.8 KiB
Python
"""Репозитории списков конференций/пользователей/команд для админ-API (пагинация, поиск)."""
|
||
|
||
import uuid
|
||
|
||
from sqlalchemy import func, or_, select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from models.conference import Conference
|
||
from models.team import Team
|
||
from models.user import User
|
||
|
||
|
||
class AdminConferenceRepository:
|
||
"""Постраничный список конференций с фильтром по статусу и текстовым поиском."""
|
||
|
||
def __init__(self, session: AsyncSession) -> None:
|
||
self._session = session
|
||
|
||
async def list_paginated(
|
||
self, *, status: str | None, q: str | None, limit: int, offset: int
|
||
) -> tuple[list[tuple[Conference, str | None, str | None]], int]:
|
||
"""Вернуть страницу конференций (+ имя/email владельца) и общее число совпадений.
|
||
|
||
`q` ищет по названию/номеру/ссылке (регистронезависимо, `ILIKE`).
|
||
Владелец — `LEFT JOIN` (может отсутствовать, ADR-001) — имя/email
|
||
нужны колонке «Владелец» в таблице админки.
|
||
"""
|
||
filters = []
|
||
if status is not None:
|
||
filters.append(Conference.status == status)
|
||
if q:
|
||
like = f"%{q}%"
|
||
filters.append(
|
||
or_(
|
||
Conference.title.ilike(like),
|
||
Conference.number.ilike(like),
|
||
Conference.slug.ilike(like),
|
||
)
|
||
)
|
||
|
||
count_stmt = select(func.count()).select_from(Conference)
|
||
items_stmt = (
|
||
select(Conference, User.name_user, User.email)
|
||
.outerjoin(User, Conference.owner_id == User.id)
|
||
.order_by(Conference.created_at.desc())
|
||
)
|
||
for condition in filters:
|
||
count_stmt = count_stmt.where(condition)
|
||
items_stmt = items_stmt.where(condition)
|
||
items_stmt = items_stmt.limit(limit).offset(offset)
|
||
|
||
total = (await self._session.execute(count_stmt)).scalar_one()
|
||
rows = (await self._session.execute(items_stmt)).all()
|
||
items = [(conference, name, email) for conference, name, email in rows]
|
||
return items, total
|
||
|
||
|
||
class AdminUserRepository:
|
||
"""Постраничный список пользователей с текстовым поиском по email/имени."""
|
||
|
||
def __init__(self, session: AsyncSession) -> None:
|
||
self._session = session
|
||
|
||
async def list_paginated(
|
||
self, *, q: str | None, limit: int, offset: int
|
||
) -> tuple[list[tuple[User, str | None]], int]:
|
||
"""Вернуть страницу пользователей (+ имя команды) и общее число совпадений.
|
||
|
||
`LEFT JOIN` на `teams` — имя команды нужно карточке профиля/таблице
|
||
админки, у пользователя без команды — `None`.
|
||
"""
|
||
filters = []
|
||
if q:
|
||
like = f"%{q}%"
|
||
filters.append(or_(User.email.ilike(like), User.name_user.ilike(like)))
|
||
|
||
count_stmt = select(func.count()).select_from(User)
|
||
items_stmt = (
|
||
select(User, Team.name)
|
||
.outerjoin(Team, User.team_id == Team.id)
|
||
.order_by(User.created_at.desc())
|
||
)
|
||
for condition in filters:
|
||
count_stmt = count_stmt.where(condition)
|
||
items_stmt = items_stmt.where(condition)
|
||
items_stmt = items_stmt.limit(limit).offset(offset)
|
||
|
||
total = (await self._session.execute(count_stmt)).scalar_one()
|
||
rows = (await self._session.execute(items_stmt)).all()
|
||
return [(user, team_name) for user, team_name in rows], total
|
||
|
||
async def get_with_team(self, user_id: uuid.UUID) -> tuple[User, str | None] | None:
|
||
"""Пользователь + имя команды по id (карточка профиля); `None` — не найден."""
|
||
result = await self._session.execute(
|
||
select(User, Team.name)
|
||
.outerjoin(Team, User.team_id == Team.id)
|
||
.where(User.id == user_id)
|
||
)
|
||
row = result.first()
|
||
return (row[0], row[1]) if row is not None else None
|
||
|
||
|
||
class TeamRepository:
|
||
"""Справочник команд: список (сортировка по названию), поиск по имени, CRUD."""
|
||
|
||
def __init__(self, session: AsyncSession) -> None:
|
||
self._session = session
|
||
|
||
async def list_all(self) -> tuple[list[Team], int]:
|
||
"""Все команды, отсортированные по названию, и их общее число."""
|
||
items_stmt = select(Team).order_by(Team.name)
|
||
items = (await self._session.execute(items_stmt)).scalars().all()
|
||
return list(items), len(items)
|
||
|
||
async def get(self, team_id: uuid.UUID) -> Team | None:
|
||
"""Найти команду по id."""
|
||
return await self._session.get(Team, team_id)
|
||
|
||
async def get_by_name(self, name: str) -> Team | None:
|
||
"""Найти команду по точному названию (проверка дубля перед созданием/переименованием)."""
|
||
result = await self._session.execute(select(Team).where(Team.name == name))
|
||
return result.scalar_one_or_none()
|
||
|
||
async def create(self, name: str) -> Team:
|
||
"""Создать команду."""
|
||
team = Team(name=name)
|
||
self._session.add(team)
|
||
await self._session.flush()
|
||
return team
|
||
|
||
async def delete(self, team: Team) -> None:
|
||
"""Удалить команду (у пользователей `team_id` обнулится через ON DELETE SET NULL)."""
|
||
await self._session.delete(team)
|