Use body for multiple events endpoints (create and end event)

frigate/api/defs/events_body.py

@@ -1,4 +1,5 @@
-from typing import Union
+from datetime import datetime
+from typing import Optional, Union
 
 from pydantic import BaseModel, Field
 
@@ -12,3 +13,16 @@ class EventsDescriptionBody(BaseModel):
     description: Union[str, None] = Field(
         title="The description of the event", min_length=1
     )
+
+
+class EventsCreateBody(BaseModel):
+    source_type: Optional[str] = "api"
+    sub_label: Optional[str] = None
+    score: Optional[int] = 0
+    duration: Optional[int] = 30
+    include_recording: Optional[bool] = True
+    draw: Optional[dict] = {}
+
+
+class EventsEndBody(BaseModel):
+    end_time: Optional[int] = datetime.now().timestamp()

frigate/api/event.py

@@ -4,7 +4,6 @@ import base64
 import io
 import logging
 import os
-from datetime import datetime
 from functools import reduce
 from pathlib import Path
 from urllib.parse import unquote
@@ -18,7 +17,12 @@ from peewee import JOIN, DoesNotExist, fn, operator
 from PIL import Image
 from playhouse.shortcuts import model_to_dict
 
-from frigate.api.defs.events_body import EventsDescriptionBody, EventsSubLabelBody
+from frigate.api.defs.events_body import (
+    EventsCreateBody,
+    EventsDescriptionBody,
+    EventsEndBody,
+    EventsSubLabelBody,
+)
 from frigate.api.defs.events_query_parameters import (
     DEFAULT_TIME_RANGE,
     EventsQueryParams,
@@ -1008,7 +1012,7 @@ def create_event(
     request: Request,
     camera_name: str,
     label: str,
-    body: dict = None,
+    body: EventsCreateBody = None,
 ):
     if not camera_name or not request.app.frigate_config.cameras.get(camera_name):
         return JSONResponse(
@@ -1024,21 +1028,18 @@ def create_event(
             status_code=404,
         )
 
-    json: dict[str, any] = body or {}
-
     try:
         frame = request.app.detected_frames_processor.get_current_frame(camera_name)
 
         event_id = request.app.external_processor.create_manual_event(
             camera_name,
             label,
-            # TODO: Create body model
+            body.source_type,
-            json.get("source_type", "api"),
+            body.sub_label,
-            json.get("sub_label", None),
+            body.score,
-            json.get("score", 0),
+            body.duration,
-            json.get("duration", 30),
+            body.include_recording,
-            json.get("include_recording", True),
+            body.draw,
-            json.get("draw", {}),
             frame,
         )
     except Exception as e:
@@ -1061,11 +1062,9 @@ def create_event(
 
 
 @router.put("/events/{event_id}/end")
-def end_event(request: Request, event_id: str, body: dict):
+def end_event(request: Request, event_id: str, body: EventsEndBody):
-    json: dict[str, any] = body or {}
-
     try:
-        end_time = json.get("end_time", datetime.now().timestamp())
+        end_time = body.end_time
         request.app.external_processor.finish_manual_event(event_id, end_time)
     except Exception:
         return JSONResponse(

frigate/api/notification.py

@@ -36,6 +36,7 @@ def get_vapid_pub_key(request: Request):
 @router.post("/notifications/register")
 def register_notifications(request: Request, body: dict = None):
     if request.app.frigate_config.auth.enabled:
+        # FIXME: For FastAPI the remote-user is not being populated
         username = request.headers.get("remote-user") or "admin"
     else:
         username = "admin"