feat: implement admin panel functionality in Mini App

- Added new API endpoints for admin features: `GET /api/admin/me`, `GET /api/admin/users`, and `PATCH /api/admin/duties/:id` to manage user duties.
- Introduced `UserForAdmin` and `AdminDutyReassignBody` schemas for handling admin-related data.
- Updated documentation to include Mini App design guidelines and admin panel functionalities.
- Enhanced tests for admin API to ensure proper access control and functionality.
- Improved error handling and localization for admin actions.
This commit is contained in:
2026-03-06 09:57:26 +03:00
parent 68b1884b73
commit c390a4dd6e
28 changed files with 2045 additions and 15 deletions

View File

@@ -12,6 +12,7 @@ from duty_teller.db.repository import (
get_duties,
get_user_by_telegram_id,
can_access_miniapp_for_telegram_user,
is_admin_for_telegram_user,
)
from duty_teller.db.schemas import DUTY_EVENT_TYPES, DutyWithUser
from duty_teller.db.session import session_scope
@@ -159,6 +160,103 @@ def get_authenticated_username(
return username or (user.full_name or "") or f"id:{telegram_user_id}"
def get_authenticated_telegram_id(
request: Request,
x_telegram_init_data: str | None,
session: Session,
) -> int:
"""Return Telegram user id for the authenticated miniapp user; 0 if skip-auth.
Same validation as get_authenticated_username. Used to check is_admin.
Args:
request: FastAPI request (for Accept-Language in error messages).
x_telegram_init_data: Raw X-Telegram-Init-Data header value.
session: DB session.
Returns:
telegram_user_id (int). When MINI_APP_SKIP_AUTH, returns 0 (no real user).
Raises:
HTTPException: 403 if initData missing/invalid or user not in allowlist.
"""
if config.MINI_APP_SKIP_AUTH:
return 0
init_data = (x_telegram_init_data or "").strip()
if not init_data:
log.warning("no X-Telegram-Init-Data header")
lang = _lang_from_accept_language(request.headers.get("Accept-Language"))
raise HTTPException(status_code=403, detail=t(lang, "api.open_from_telegram"))
max_age = config.INIT_DATA_MAX_AGE_SECONDS or None
telegram_user_id, username, auth_reason, lang = validate_init_data_with_reason(
init_data, config.BOT_TOKEN, max_age_seconds=max_age
)
if auth_reason != "ok":
log.warning("initData validation failed: %s", auth_reason)
raise HTTPException(
status_code=403, detail=_auth_error_detail(auth_reason, lang)
)
if telegram_user_id is None:
log.warning("initData valid but telegram_user_id missing")
raise HTTPException(status_code=403, detail=t(lang, "api.access_denied"))
user = get_user_by_telegram_id(session, telegram_user_id)
if not user:
log.warning(
"user not in DB (username=%s, telegram_id=%s)",
username,
telegram_user_id,
)
raise HTTPException(status_code=403, detail=t(lang, "api.access_denied"))
if not can_access_miniapp_for_telegram_user(session, telegram_user_id):
failed_phone = config.normalize_phone(user.phone) if user.phone else None
log.warning(
"access denied (username=%s, telegram_id=%s, phone=%s)",
username,
telegram_user_id,
failed_phone or "",
)
raise HTTPException(status_code=403, detail=t(lang, "api.access_denied"))
return telegram_user_id
def get_authenticated_telegram_id_dep(
request: Request,
x_telegram_init_data: Annotated[
str | None, Header(alias="X-Telegram-Init-Data")
] = None,
session: Session = Depends(get_db_session),
) -> int:
"""FastAPI dependency: return telegram_user_id for authenticated miniapp user (0 if skip-auth)."""
return get_authenticated_telegram_id(request, x_telegram_init_data, session)
def require_admin_telegram_id(
request: Request,
x_telegram_init_data: Annotated[
str | None, Header(alias="X-Telegram-Init-Data")
] = None,
session: Session = Depends(get_db_session),
) -> int:
"""FastAPI dependency: require valid miniapp auth and admin role; return telegram_user_id.
When MINI_APP_SKIP_AUTH is True, admin routes are disabled (403).
Raises:
HTTPException: 403 if initData missing/invalid, user not in allowlist, or not admin.
"""
if config.MINI_APP_SKIP_AUTH:
log.warning("Admin routes disabled when MINI_APP_SKIP_AUTH is set")
lang = _lang_from_accept_language(request.headers.get("Accept-Language"))
raise HTTPException(status_code=403, detail=t(lang, "import.admin_only"))
telegram_user_id = get_authenticated_telegram_id(
request, x_telegram_init_data, session
)
if not is_admin_for_telegram_user(session, telegram_user_id):
lang = _lang_from_accept_language(request.headers.get("Accept-Language"))
raise HTTPException(status_code=403, detail=t(lang, "import.admin_only"))
return telegram_user_id
def fetch_duties_response(
session: Session, from_date: str, to_date: str
) -> list[DutyWithUser]: