Nikolay Tatarinov
aa89494bd5
All checks were successful
CI / lint-and-test (push) Successful in 22s
- Added support for filtering calendar events by type in the ICS generation API endpoint, allowing users to specify whether to include only duty shifts or all event types (duty, unavailable, vacation). - Updated the `get_duties_for_user` function to accept an optional `event_types` parameter, enabling more flexible data retrieval based on user preferences. - Enhanced unit tests to cover the new event type filtering functionality, ensuring correct behavior and reliability of the ICS generation process.
133 lines
4.4 KiB
Python
133 lines
4.4 KiB
Python
"""FastAPI app: /api/duties, /api/calendar-events, personal ICS, and static webapp at /app."""
|
|
|
|
import logging
|
|
import re
|
|
from datetime import date, timedelta
|
|
from typing import Literal
|
|
|
|
import duty_teller.config as config
|
|
from fastapi import Depends, FastAPI, Request
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from fastapi.responses import Response
|
|
from fastapi.staticfiles import StaticFiles
|
|
from sqlalchemy.orm import Session
|
|
|
|
from duty_teller.api.calendar_ics import get_calendar_events
|
|
from duty_teller.api.dependencies import (
|
|
fetch_duties_response,
|
|
get_db_session,
|
|
get_validated_dates,
|
|
require_miniapp_username,
|
|
)
|
|
from duty_teller.api.personal_calendar_ics import build_personal_ics
|
|
from duty_teller.db.repository import get_duties_for_user, get_user_by_calendar_token
|
|
from duty_teller.db.schemas import CalendarEvent, DutyWithUser
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
# Calendar tokens are secrets.token_urlsafe(32) → base64url, length 43
|
|
_CALENDAR_TOKEN_RE = re.compile(r"^[A-Za-z0-9_-]{40,50}$")
|
|
|
|
|
|
def _is_valid_calendar_token(token: str) -> bool:
|
|
"""Return True if token matches expected format (length and alphabet). Rejects invalid before DB."""
|
|
return bool(token and _CALENDAR_TOKEN_RE.match(token))
|
|
|
|
|
|
app = FastAPI(title="Duty Teller API")
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=config.CORS_ORIGINS,
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
|
|
@app.get(
|
|
"/api/duties",
|
|
response_model=list[DutyWithUser],
|
|
summary="List duties",
|
|
description=(
|
|
"Returns duties for the given date range. Requires Telegram Mini App initData "
|
|
"(or MINI_APP_SKIP_AUTH / private IP in dev)."
|
|
),
|
|
)
|
|
def list_duties(
|
|
request: Request,
|
|
dates: tuple[str, str] = Depends(get_validated_dates),
|
|
_username: str = Depends(require_miniapp_username),
|
|
session: Session = Depends(get_db_session),
|
|
):
|
|
from_date_val, to_date_val = dates
|
|
log.info(
|
|
"GET /api/duties from %s",
|
|
request.client.host if request.client else "?",
|
|
)
|
|
return fetch_duties_response(session, from_date_val, to_date_val)
|
|
|
|
|
|
@app.get(
|
|
"/api/calendar-events",
|
|
response_model=list[CalendarEvent],
|
|
summary="List calendar events",
|
|
description=(
|
|
"Returns calendar events for the date range, including external ICS when "
|
|
"EXTERNAL_CALENDAR_ICS_URL is set. Auth same as /api/duties."
|
|
),
|
|
)
|
|
def list_calendar_events(
|
|
dates: tuple[str, str] = Depends(get_validated_dates),
|
|
_username: str = Depends(require_miniapp_username),
|
|
):
|
|
from_date_val, to_date_val = dates
|
|
url = config.EXTERNAL_CALENDAR_ICS_URL
|
|
if not url:
|
|
return []
|
|
events = get_calendar_events(url, from_date=from_date_val, to_date=to_date_val)
|
|
return [CalendarEvent(date=e["date"], summary=e["summary"]) for e in events]
|
|
|
|
|
|
@app.get(
|
|
"/api/calendar/ical/{token}.ics",
|
|
summary="Personal calendar ICS",
|
|
description=(
|
|
"Returns an ICS calendar with the subscribing user's events. "
|
|
"By default only duty shifts are included; use query parameter events=all "
|
|
"for all event types (duty, unavailable, vacation). "
|
|
"No Telegram auth; access is by secret token in the URL."
|
|
),
|
|
)
|
|
def get_personal_calendar_ical(
|
|
token: str,
|
|
events: Literal["duty", "all"] = "duty",
|
|
session: Session = Depends(get_db_session),
|
|
) -> Response:
|
|
"""
|
|
Return ICS calendar with the subscribing user's events.
|
|
Default: only duty shifts. Use ?events=all for duty, unavailable, vacation.
|
|
No Telegram auth; access is by secret token in the URL.
|
|
"""
|
|
if not _is_valid_calendar_token(token):
|
|
return Response(status_code=404, content="Not found")
|
|
user = get_user_by_calendar_token(session, token)
|
|
if user is None:
|
|
return Response(status_code=404, content="Not found")
|
|
today = date.today()
|
|
from_date = (today - timedelta(days=365)).strftime("%Y-%m-%d")
|
|
to_date = (today + timedelta(days=365 * 2)).strftime("%Y-%m-%d")
|
|
event_types = ["duty"] if events == "duty" else None
|
|
duties_with_name = get_duties_for_user(
|
|
session, user.id, from_date=from_date, to_date=to_date, event_types=event_types
|
|
)
|
|
ics_bytes = build_personal_ics(duties_with_name)
|
|
return Response(
|
|
content=ics_bytes,
|
|
media_type="text/calendar; charset=utf-8",
|
|
)
|
|
|
|
|
|
webapp_path = config.PROJECT_ROOT / "webapp"
|
|
if webapp_path.is_dir():
|
|
app.mount("/app", StaticFiles(directory=str(webapp_path), html=True), name="webapp")
|