154f968296
Phase 1-3 of decoupling: - path_setup.py adds project root to sys.path - Events-owned models in events/models/ (calendars with all related models) - All imports updated: shared.infrastructure, shared.db, shared.browser, etc. - Calendar uses container_type/container_id instead of post_id FK - CalendarEntryPost uses content_type/content_id (generic content refs) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
110 lines
3.2 KiB
Python
110 lines
3.2 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
from typing import Optional
|
|
import calendar as pycalendar
|
|
|
|
from quart import request
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import selectinload, with_loader_criteria
|
|
|
|
from models.calendars import Calendar, CalendarSlot
|
|
|
|
def parse_int_arg(name: str, default: Optional[int] = None) -> Optional[int]:
|
|
"""Parse an integer query parameter from the request."""
|
|
val = request.args.get(name, "").strip()
|
|
if not val:
|
|
return default
|
|
try:
|
|
return int(val)
|
|
except ValueError:
|
|
return default
|
|
|
|
|
|
def add_months(year: int, month: int, delta: int) -> tuple[int, int]:
|
|
"""Add (or subtract) months to a given year/month, handling year overflow."""
|
|
new_month = month + delta
|
|
new_year = year + (new_month - 1) // 12
|
|
new_month = ((new_month - 1) % 12) + 1
|
|
return new_year, new_month
|
|
|
|
|
|
def build_calendar_weeks(year: int, month: int) -> list[list[dict]]:
|
|
"""
|
|
Build a calendar grid for the given year and month.
|
|
Returns a list of weeks, where each week is a list of 7 day dictionaries.
|
|
"""
|
|
today = datetime.now(timezone.utc).date()
|
|
cal = pycalendar.Calendar(firstweekday=0) # 0 = Monday
|
|
weeks: list[list[dict]] = []
|
|
|
|
for week in cal.monthdatescalendar(year, month):
|
|
week_days = []
|
|
for d in week:
|
|
week_days.append(
|
|
{
|
|
"date": d,
|
|
"in_month": (d.month == month),
|
|
"is_today": (d == today),
|
|
}
|
|
)
|
|
weeks.append(week_days)
|
|
|
|
return weeks
|
|
|
|
|
|
async def get_calendar_by_post_and_slug(
|
|
session: AsyncSession,
|
|
post_id: int,
|
|
calendar_slug: str,
|
|
) -> Optional[Calendar]:
|
|
"""
|
|
Fetch a calendar by post_id and slug, with slots eagerly loaded.
|
|
Returns None if not found.
|
|
"""
|
|
result = await session.execute(
|
|
select(Calendar)
|
|
.options(
|
|
selectinload(Calendar.slots),
|
|
with_loader_criteria(CalendarSlot, CalendarSlot.deleted_at.is_(None)),
|
|
)
|
|
.where(
|
|
Calendar.container_type == "page",
|
|
Calendar.container_id == post_id,
|
|
Calendar.slug == calendar_slug,
|
|
Calendar.deleted_at.is_(None),
|
|
)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def get_calendar_by_slug(
|
|
session: AsyncSession,
|
|
calendar_slug: str,
|
|
) -> Optional[Calendar]:
|
|
"""
|
|
Fetch a calendar by slug only (for standalone events service).
|
|
With slots eagerly loaded. Returns None if not found.
|
|
"""
|
|
result = await session.execute(
|
|
select(Calendar)
|
|
.options(
|
|
selectinload(Calendar.slots),
|
|
with_loader_criteria(CalendarSlot, CalendarSlot.deleted_at.is_(None)),
|
|
)
|
|
.where(
|
|
Calendar.slug == calendar_slug,
|
|
Calendar.deleted_at.is_(None),
|
|
)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def update_calendar_description(
|
|
calendar: Calendar,
|
|
description: Optional[str],
|
|
) -> None:
|
|
"""Update calendar description (in-place on the calendar object)."""
|
|
calendar.description = description or None
|