bvandeusenandClaude Sonnet 4.6
0277f5744f
auth.py sets g.user (the User object), not g.user_id directly. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
120 lines
3.7 KiB
Python
120 lines
3.7 KiB
Python
"""Calendar events REST API."""
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
|
|
from quart import Blueprint, g, jsonify, request
|
|
|
|
from fabledassistant.auth import login_required
|
|
import fabledassistant.services.events as events_svc
|
|
|
|
events_bp = Blueprint("events", __name__, url_prefix="/api/events")
|
|
|
|
|
|
def _get_current_user_id() -> int:
|
|
return g.user.id
|
|
|
|
|
|
@events_bp.get("")
|
|
@login_required
|
|
async def list_events():
|
|
date_from_str = request.args.get("from")
|
|
date_to_str = request.args.get("to")
|
|
if not date_from_str or not date_to_str:
|
|
return jsonify({"error": "from and to query params are required"}), 400
|
|
try:
|
|
date_from = datetime.fromisoformat(date_from_str)
|
|
date_to = datetime.fromisoformat(date_to_str)
|
|
except ValueError:
|
|
return jsonify({"error": "Invalid datetime format"}), 400
|
|
events = await events_svc.list_events(
|
|
user_id=_get_current_user_id(),
|
|
date_from=date_from,
|
|
date_to=date_to,
|
|
)
|
|
return jsonify([e.to_dict() for e in events])
|
|
|
|
|
|
@events_bp.post("")
|
|
@login_required
|
|
async def create_event():
|
|
data = await request.get_json() or {}
|
|
if not data.get("title") or not data.get("start_dt"):
|
|
return jsonify({"error": "title and start_dt are required"}), 400
|
|
try:
|
|
start_dt = datetime.fromisoformat(data["start_dt"])
|
|
end_dt = datetime.fromisoformat(data["end_dt"]) if data.get("end_dt") else None
|
|
except ValueError:
|
|
return jsonify({"error": "Invalid datetime format"}), 400
|
|
event = await events_svc.create_event(
|
|
user_id=_get_current_user_id(),
|
|
title=data["title"],
|
|
start_dt=start_dt,
|
|
end_dt=end_dt,
|
|
all_day=data.get("all_day", False),
|
|
description=data.get("description", ""),
|
|
location=data.get("location", ""),
|
|
color=data.get("color", ""),
|
|
recurrence=data.get("recurrence"),
|
|
project_id=data.get("project_id"),
|
|
)
|
|
return jsonify(event.to_dict()), 201
|
|
|
|
|
|
@events_bp.get("/<int:event_id>")
|
|
@login_required
|
|
async def get_event(event_id: int):
|
|
event = await events_svc.get_event(
|
|
user_id=_get_current_user_id(),
|
|
event_id=event_id,
|
|
)
|
|
if event is None:
|
|
return jsonify({"error": "Event not found"}), 404
|
|
return jsonify(event.to_dict())
|
|
|
|
|
|
@events_bp.patch("/<int:event_id>")
|
|
@login_required
|
|
async def update_event(event_id: int):
|
|
data = await request.get_json() or {}
|
|
fields: dict = {}
|
|
for str_field in ("title", "description", "location", "color", "recurrence"):
|
|
if str_field in data:
|
|
fields[str_field] = data[str_field]
|
|
for bool_field in ("all_day",):
|
|
if bool_field in data:
|
|
fields[bool_field] = data[bool_field]
|
|
for int_field in ("project_id",):
|
|
if int_field in data:
|
|
fields[int_field] = data[int_field]
|
|
for dt_field in ("start_dt", "end_dt"):
|
|
if dt_field in data and data[dt_field]:
|
|
try:
|
|
fields[dt_field] = datetime.fromisoformat(data[dt_field])
|
|
except ValueError:
|
|
return jsonify({"error": f"Invalid datetime for {dt_field}"}), 400
|
|
event = await events_svc.update_event(
|
|
user_id=_get_current_user_id(),
|
|
event_id=event_id,
|
|
**fields,
|
|
)
|
|
if event is None:
|
|
return jsonify({"error": "Event not found"}), 404
|
|
return jsonify(event.to_dict())
|
|
|
|
|
|
@events_bp.delete("/<int:event_id>")
|
|
@login_required
|
|
async def delete_event(event_id: int):
|
|
event = await events_svc.get_event(
|
|
user_id=_get_current_user_id(),
|
|
event_id=event_id,
|
|
)
|
|
if event is None:
|
|
return jsonify({"error": "Event not found"}), 404
|
|
await events_svc.delete_event(
|
|
user_id=_get_current_user_id(),
|
|
event_id=event_id,
|
|
)
|
|
return "", 204
|