feat(calendar): events REST API blueprint and registration

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

src/fabledassistant/app.py

@@ -29,6 +29,7 @@ from fabledassistant.routes.shares import shares_bp
 from fabledassistant.routes.in_app_notifications import notifications_bp
 from fabledassistant.routes.users import users_bp
 from fabledassistant.routes.api_keys import api_keys_bp
+from fabledassistant.routes.events import events_bp
 from fabledassistant.routes.search import search_bp
 
 STATIC_DIR = Path(__file__).parent / "static"
@@ -82,6 +83,7 @@ def create_app() -> Quart:
     app.register_blueprint(notifications_bp)
     app.register_blueprint(users_bp)
     app.register_blueprint(api_keys_bp)
+    app.register_blueprint(events_bp)
     app.register_blueprint(search_bp)
 
     @app.before_request

src/fabledassistant/routes/events.py

@@ -0,0 +1,119 @@
+"""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

tests/test_events_routes.py

@@ -0,0 +1,67 @@
+"""Route-level tests for the events blueprint.
+
+Full HTTP integration tests require a live DB (not available in unit test
+environment). These tests cover structural correctness and the route
+module's public interface; ownership enforcement is covered by the service
+tests in test_events_service.py.
+"""
+
+
+def test_events_blueprint_registered():
+    """events_bp must be importable and have the correct name."""
+    from fabledassistant.routes.events import events_bp
+    assert events_bp.name == "events"
+    assert events_bp.url_prefix == "/api/events"
+
+
+def test_events_blueprint_has_five_routes():
+    """Blueprint must declare routes for GET/POST '' and GET/PATCH/DELETE '/<id>'."""
+    from fabledassistant.routes.events import events_bp
+    methods_by_rule: dict[str, set[str]] = {}
+    for rule in events_bp.deferred_functions:
+        pass  # deferred; inspect via url_map after binding
+    # Import routes module to confirm all 5 view functions exist
+    from fabledassistant.routes import events as events_module
+    assert callable(events_module.list_events)
+    assert callable(events_module.create_event)
+    assert callable(events_module.get_event)
+    assert callable(events_module.update_event)
+    assert callable(events_module.delete_event)
+
+
+def test_events_blueprint_registered_in_app():
+    """events_bp must be registered in the app factory."""
+    from fabledassistant.app import create_app
+    app = create_app()
+    # Check the blueprint is present in the app's blueprints dict
+    assert "events" in app.blueprints
+
+
+def test_events_service_ownership_enforced_on_get():
+    """get_event returns None for a different user — route will 404."""
+    # Ownership is enforced by the service filtering by user_id.
+    # The service returns None when the event belongs to a different user,
+    # and the route converts that to a 404 response.
+    import inspect
+    from fabledassistant.services import events as events_svc
+    sig = inspect.signature(events_svc.get_event)
+    assert "user_id" in sig.parameters
+    assert "event_id" in sig.parameters
+
+
+def test_events_service_ownership_enforced_on_update():
+    """update_event takes user_id — route passes current user's id."""
+    import inspect
+    from fabledassistant.services import events as events_svc
+    sig = inspect.signature(events_svc.update_event)
+    assert "user_id" in sig.parameters
+    assert "event_id" in sig.parameters
+
+
+def test_events_service_ownership_enforced_on_delete():
+    """delete_event takes user_id — route verifies ownership before deleting."""
+    import inspect
+    from fabledassistant.services import events as events_svc
+    sig = inspect.signature(events_svc.delete_event)
+    assert "user_id" in sig.parameters
+    assert "event_id" in sig.parameters