"""Calendar API views. Endpoints --------- GET /calendar/events/?from=&to=&owner= list events (date-range required) POST /calendar/events/ create event GET /calendar/events/{public_id}/ retrieve event PATCH /calendar/events/{public_id}/ update event DELETE /calendar/events/{public_id}/ delete event POST /calendar/events/{public_id}/attendees/ add attendee DELETE /calendar/events/{public_id}/attendees/{uid}/ remove attendee """ from __future__ import annotations from django.shortcuts import get_object_or_404 from rest_framework import status from rest_framework.decorators import api_view, permission_classes from rest_framework.permissions import IsAuthenticated from rest_framework.request import Request from rest_framework.response import Response from simorgh.apps.calendar import selectors as sel from simorgh.apps.calendar import services from simorgh.apps.calendar.models import CalendarEvent, EventAttendee # --------------------------------------------------------------------------- # Serialization helpers # --------------------------------------------------------------------------- def _serialize_attendee(att: EventAttendee) -> dict: return { "id": att.pk, "public_id": str(att.public_id), "user_id": att.user_id, "external_email": att.external_email, "status": att.status, } def _serialize_event(event: CalendarEvent, *, detail: bool = False) -> dict: data: dict = { "id": event.pk, "public_id": str(event.public_id), "title": event.title, "description": event.description, "location": event.location, "start_dt": event.start_dt.isoformat(), "end_dt": event.end_dt.isoformat(), "all_day": event.all_day, "color": event.color, "is_recurring": event.is_recurring, "owner_id": event.owner_id, "tenant_id": event.tenant_id, "organization_node_id": event.organization_node_id, "created_at": event.created_at.isoformat(), "updated_at": event.updated_at.isoformat(), } if detail: data["attendees"] = [_serialize_attendee(a) for a in event.attendees.select_related("user")] return data def _get_tenant_id(request: Request) -> int | None: tenant = getattr(request, "tenant", None) return tenant.pk if tenant else None # --------------------------------------------------------------------------- # Events list + create # --------------------------------------------------------------------------- @api_view(["GET", "POST"]) @permission_classes([IsAuthenticated]) def event_list_create(request: Request) -> Response: tenant_id = _get_tenant_id(request) if not tenant_id: return Response({"detail": "Tenant not resolved."}, status=status.HTTP_400_BAD_REQUEST) if request.method == "GET": from_dt_str = request.query_params.get("from") to_dt_str = request.query_params.get("to") if not from_dt_str or not to_dt_str: return Response( {"detail": "Query params 'from' and 'to' are required."}, status=status.HTTP_400_BAD_REQUEST, ) from datetime import datetime, timezone try: from_dt = datetime.fromisoformat(from_dt_str) to_dt = datetime.fromisoformat(to_dt_str) except ValueError: return Response({"detail": "Invalid date format."}, status=status.HTTP_400_BAD_REQUEST) # linked-object filter linked_type = request.query_params.get("linked_type") linked_id = request.query_params.get("linked_id") if linked_type and linked_id: events = sel.get_linked_events(tenant_id, request.user.pk, linked_type, linked_id) else: owner_id_raw = request.query_params.get("owner") owner_id = int(owner_id_raw) if owner_id_raw else None events = sel.get_events_for_range( tenant_id, request.user.pk, from_dt, to_dt, owner_id=owner_id, ) return Response([_serialize_event(e) for e in events]) # POST d = request.data org_node_id = d.get("organization_node_id") title = d.get("title", "") start_dt_str = d.get("start_dt") end_dt_str = d.get("end_dt") if not title or not start_dt_str or not end_dt_str: return Response({"detail": "title, start_dt, and end_dt are required."}, status=status.HTTP_400_BAD_REQUEST) from datetime import datetime try: start_dt = datetime.fromisoformat(start_dt_str) end_dt = datetime.fromisoformat(end_dt_str) except ValueError: return Response({"detail": "Invalid datetime format."}, status=status.HTTP_400_BAD_REQUEST) from django.core.exceptions import ValidationError try: event = services.create_event( tenant_id=tenant_id, organization_node_id=org_node_id or tenant_id, owner_id=request.user.pk, title=title, start_dt=start_dt, end_dt=end_dt, all_day=d.get("all_day", False), description=d.get("description", ""), location=d.get("location", ""), color=d.get("color", ""), is_recurring=d.get("is_recurring", False), ) except ValidationError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) return Response(_serialize_event(event), status=status.HTTP_201_CREATED) # --------------------------------------------------------------------------- # Event detail # --------------------------------------------------------------------------- @api_view(["GET", "PATCH", "DELETE"]) @permission_classes([IsAuthenticated]) def event_detail(request: Request, public_id: str) -> Response: tenant_id = _get_tenant_id(request) if not tenant_id: return Response({"detail": "Tenant not resolved."}, status=status.HTTP_400_BAD_REQUEST) event = sel.get_event_by_public_id(tenant_id, str(public_id)) if not event: return Response({"detail": "Not found."}, status=status.HTTP_404_NOT_FOUND) if request.method == "GET": return Response(_serialize_event(event, detail=True)) if request.method == "PATCH": from datetime import datetime from django.core.exceptions import ValidationError update_kwargs: dict = {} allowed = ("title", "description", "location", "start_dt", "end_dt", "all_day", "color", "is_recurring") for field in allowed: if field in request.data: val = request.data[field] if field in ("start_dt", "end_dt"): try: val = datetime.fromisoformat(val) except ValueError: return Response({"detail": f"Invalid datetime for {field}."}, status=status.HTTP_400_BAD_REQUEST) update_kwargs[field] = val try: event = services.update_event(event, **update_kwargs) except ValidationError as exc: return Response({"detail": str(exc)}, status=status.HTTP_400_BAD_REQUEST) return Response(_serialize_event(event)) # DELETE services.delete_event(event) return Response(status=status.HTTP_204_NO_CONTENT) # --------------------------------------------------------------------------- # Attendees # --------------------------------------------------------------------------- @api_view(["POST"]) @permission_classes([IsAuthenticated]) def event_attendees_add(request: Request, public_id: str) -> Response: tenant_id = _get_tenant_id(request) if not tenant_id: return Response({"detail": "Tenant not resolved."}, status=status.HTTP_400_BAD_REQUEST) event = sel.get_event_by_public_id(tenant_id, str(public_id)) if not event: return Response({"detail": "Not found."}, status=status.HTTP_404_NOT_FOUND) user_id = request.data.get("user_id") external_email = request.data.get("external_email", "") if not user_id and not external_email: return Response({"detail": "user_id or external_email required."}, status=status.HTTP_400_BAD_REQUEST) attendee = services.add_attendee(event, user_id=user_id, external_email=external_email) return Response(_serialize_attendee(attendee), status=status.HTTP_201_CREATED) @api_view(["DELETE"]) @permission_classes([IsAuthenticated]) def event_attendees_remove(request: Request, public_id: str, user_id: int) -> Response: tenant_id = _get_tenant_id(request) if not tenant_id: return Response({"detail": "Tenant not resolved."}, status=status.HTTP_400_BAD_REQUEST) event = sel.get_event_by_public_id(tenant_id, str(public_id)) if not event: return Response({"detail": "Not found."}, status=status.HTTP_404_NOT_FOUND) services.remove_attendee(event, user_id=user_id) return Response(status=status.HTTP_204_NO_CONTENT)