115 lines
3.8 KiB
Python
115 lines
3.8 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
from collections.abc import Callable
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
from fastapi import HTTPException
|
|
from sqlalchemy import or_, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from . import calendar as calendar_service
|
|
from .models import CalendarSubscription
|
|
|
|
logger = logging.getLogger(__name__)
|
|
DEFAULT_REFRESH_INTERVAL = timedelta(minutes=15)
|
|
DEFAULT_POLL_SECONDS = 60
|
|
|
|
|
|
async def refresh_subscription_cache(
|
|
db: AsyncSession,
|
|
row: CalendarSubscription,
|
|
*,
|
|
fail_without_cache: bool = True,
|
|
attempted_at: datetime | None = None,
|
|
) -> bool:
|
|
attempt_time = attempted_at or datetime.now(UTC)
|
|
requested_url = row.url
|
|
requested_version = getattr(row, "updated_at", None)
|
|
try:
|
|
result = await asyncio.to_thread(
|
|
calendar_service.fetch_calendar,
|
|
requested_url,
|
|
etag=row.etag,
|
|
last_modified=row.last_modified,
|
|
)
|
|
await db.refresh(row)
|
|
if row.url != requested_url or getattr(row, "updated_at", None) != requested_version:
|
|
return False
|
|
if result.not_modified:
|
|
if not row.ics_cache:
|
|
raise HTTPException(502, "calendar returned not modified without cache")
|
|
elif result.content is not None:
|
|
# Parse before replacing a known-good cache.
|
|
calendar_service.parse_ics_events(
|
|
result.content,
|
|
row.name,
|
|
row.color,
|
|
datetime.now(UTC) - timedelta(days=1),
|
|
datetime.now(UTC) + timedelta(days=1),
|
|
"UTC",
|
|
)
|
|
row.ics_cache = result.content.decode("utf-8-sig")
|
|
row.etag = result.etag
|
|
row.last_modified = result.last_modified
|
|
row.refreshed_at = attempt_time
|
|
row.updated_at = attempt_time
|
|
row.last_error = None
|
|
await db.commit()
|
|
return True
|
|
except Exception as exc:
|
|
error = exc.detail if isinstance(exc, HTTPException) else str(exc)
|
|
if not row.ics_cache and fail_without_cache:
|
|
await db.rollback()
|
|
raise HTTPException(502, error) from exc
|
|
row.last_error = error
|
|
row.updated_at = attempt_time
|
|
await db.commit()
|
|
return False
|
|
|
|
|
|
async def refresh_due_subscriptions(
|
|
db: AsyncSession,
|
|
*,
|
|
now: datetime | None = None,
|
|
refresh_interval: timedelta = DEFAULT_REFRESH_INTERVAL,
|
|
) -> int:
|
|
current = now or datetime.now(UTC)
|
|
cutoff = current - refresh_interval
|
|
rows = (await db.scalars(select(CalendarSubscription).where(
|
|
CalendarSubscription.enabled.is_(True),
|
|
or_(
|
|
CalendarSubscription.updated_at.is_(None),
|
|
CalendarSubscription.updated_at <= cutoff,
|
|
),
|
|
).order_by(CalendarSubscription.refreshed_at, CalendarSubscription.created_at))).all()
|
|
for row in rows:
|
|
try:
|
|
await refresh_subscription_cache(
|
|
db, row, fail_without_cache=False, attempted_at=current
|
|
)
|
|
except Exception:
|
|
await db.rollback()
|
|
logger.exception("Unexpected calendar refresh failure", extra={"subscription_id": str(row.id)})
|
|
return len(rows)
|
|
|
|
|
|
async def calendar_refresh_loop(
|
|
session_factory: Callable[[], AsyncSession],
|
|
stop_event: asyncio.Event,
|
|
*,
|
|
refresh_interval: timedelta = DEFAULT_REFRESH_INTERVAL,
|
|
poll_seconds: int = DEFAULT_POLL_SECONDS,
|
|
) -> None:
|
|
while not stop_event.is_set():
|
|
try:
|
|
async with session_factory() as db:
|
|
await refresh_due_subscriptions(db, refresh_interval=refresh_interval)
|
|
except Exception:
|
|
logger.exception("Calendar background refresh cycle failed")
|
|
try:
|
|
await asyncio.wait_for(stop_event.wait(), timeout=poll_seconds)
|
|
except TimeoutError:
|
|
pass
|