Convert app endpoints to FastAPI

frigate/api/app.py

@@ -12,14 +12,17 @@ from typing import Optional
 import requests
 from fastapi import APIRouter, Path, Request, Response
 from fastapi.encoders import jsonable_encoder
+from fastapi.params import Depends
 from fastapi.responses import JSONResponse
-from flask import Blueprint, Flask, current_app, jsonify, make_response, request
+from flask import Blueprint, Flask, jsonify, request
 from markupsafe import escape
 from peewee import operator
 from playhouse.sqliteq import SqliteQueueDatabase
 from werkzeug.middleware.proxy_fix import ProxyFix
 
 from frigate.api.auth import AuthBp, get_jwt_secret, limiter
+from frigate.api.defs.app_body import AppConfigSetBody
+from frigate.api.defs.app_query_parameters import AppTimelineHourlyQueryParameters
 from frigate.api.defs.tags import Tags
 from frigate.config import FrigateConfig
 from frigate.const import CONFIG_DIR
@@ -44,7 +47,7 @@ logger = logging.getLogger(__name__)
 bp = Blueprint("frigate", __name__)
 bp.register_blueprint(AuthBp)
 
-router = APIRouter()
+router = APIRouter(tags=[Tags.app])
 
 
 def create_app(
@@ -99,74 +102,72 @@ def create_app(
     return app
 
 
-@router.get("/", tags=[Tags.app])
+@router.get("/")
 def is_healthy():
     return "Frigate is running. Alive and healthy!"
 
 
-@router.get("/config/schema.json", tags=[Tags.app])
+@router.get("/config/schema.json")
 def config_schema(request: Request):
     return Response(
         content=request.app.frigate_config.schema_json(), media_type="application/json"
     )
 
 
-@bp.route("/go2rtc/streams")
+@router.get("/go2rtc/streams")
 def go2rtc_streams():
     r = requests.get("http://127.0.0.1:1984/api/streams")
     if not r.ok:
         logger.error("Failed to fetch streams from go2rtc")
-        return make_response(
+        return JSONResponse(
-            jsonify({"success": False, "message": "Error fetching stream data"}),
+            content=({"success": False, "message": "Error fetching stream data"}),
-            500,
+            status_code=500,
         )
     stream_data = r.json()
     for data in stream_data.values():
         for producer in data.get("producers", []):
             producer["url"] = clean_camera_user_pass(producer.get("url", ""))
-    return jsonify(stream_data)
+    return JSONResponse(content=stream_data)
 
 
-@bp.route("/go2rtc/streams/<camera_name>")
+@router.get("/go2rtc/streams/<camera_name>")
 def go2rtc_camera_stream(camera_name: str):
     r = requests.get(
         f"http://127.0.0.1:1984/api/streams?src={camera_name}&video=all&audio=all&microphone"
     )
     if not r.ok:
         logger.error("Failed to fetch streams from go2rtc")
-        return make_response(
+        return JSONResponse(
-            jsonify({"success": False, "message": "Error fetching stream data"}),
+            content=({"success": False, "message": "Error fetching stream data"}),
-            500,
+            status_code=500,
         )
     stream_data = r.json()
     for producer in stream_data.get("producers", []):
         producer["url"] = clean_camera_user_pass(producer.get("url", ""))
-    return jsonify(stream_data)
+    return JSONResponse(content=stream_data)
 
 
-@bp.route("/version")
+@router.get("/version")
 def version():
     return VERSION
 
 
-@bp.route("/stats")
+@router.get("/stats")
-def stats():
+def stats(request: Request):
-    return jsonify(current_app.stats_emitter.get_latest_stats())
+    return JSONResponse(content=request.app.stats_emitter.get_latest_stats())
 
 
-@bp.route("/stats/history")
+@router.get("/stats/history")
-def stats_history():
+def stats_history(request: Request, keys: str = None):
-    keys = request.args.get("keys", default=None)
-
     if keys:
         keys = keys.split(",")
 
-    return jsonify(current_app.stats_emitter.get_stats_history(keys))
+    return JSONResponse(content=request.app.stats_emitter.get_stats_history(keys))
 
 
-@bp.route("/config")
+@router.get("/config")
-def config():
+def config(request: Request):
-    config_obj: FrigateConfig = current_app.frigate_config
+    config_obj: FrigateConfig = request.app.frigate_config
     config: dict[str, dict[str, any]] = config_obj.model_dump(
         mode="json", warnings="none", exclude_none=True
     )
@@ -177,7 +178,7 @@ def config():
     # remove the proxy secret
     config["proxy"].pop("auth_secret", None)
 
-    for camera_name, camera in current_app.frigate_config.cameras.items():
+    for camera_name, camera in request.app.frigate_config.cameras.items():
         camera_dict = config["cameras"][camera_name]
 
         # clean paths
@@ -193,18 +194,18 @@ def config():
         for zone_name, zone in config_obj.cameras[camera_name].zones.items():
             camera_dict["zones"][zone_name]["color"] = zone.color
 
-    config["plus"] = {"enabled": current_app.plus_api.is_active()}
+    config["plus"] = {"enabled": request.app.plus_api.is_active()}
     config["model"]["colormap"] = config_obj.model.colormap
 
     for detector_config in config["detectors"].values():
         detector_config["model"]["labelmap"] = (
-            current_app.frigate_config.model.merged_labelmap
+            request.app.frigate_config.model.merged_labelmap
         )
 
-    return jsonify(config)
+    return JSONResponse(content=config)
 
 
-@bp.route("/config/raw")
+@router.get("/config/raw")
 def config_raw():
     config_file = os.environ.get("CONFIG_FILE", "/config/config.yml")
 
@@ -215,43 +216,43 @@ def config_raw():
         config_file = config_file_yaml
 
     if not os.path.isfile(config_file):
-        return make_response(
+        return JSONResponse(
-            jsonify({"success": False, "message": "Could not find file"}), 404
+            content=({"success": False, "message": "Could not find file"}),
+            status_code=404,
         )
 
     with open(config_file, "r") as f:
         raw_config = f.read()
         f.close()
 
+        # TODO: How to return
         return raw_config, 200
 
 
-@bp.route("/config/save", methods=["POST"])
+@router.post("/config/save")
-def config_save():
+def config_save(save_option: str, body: dict):
-    save_option = request.args.get("save_option")
+    new_config = body
-
-    new_config = request.get_data().decode()
 
     if not new_config:
-        return make_response(
+        return JSONResponse(
-            jsonify(
+            content=(
                 {"success": False, "message": "Config with body param is required"}
             ),
-            400,
+            status_code=400,
         )
 
     # Validate the config schema
     try:
         FrigateConfig.parse_raw(new_config)
     except Exception:
-        return make_response(
+        return JSONResponse(
-            jsonify(
+            content=(
                 {
                     "success": False,
                     "message": f"\nConfig Error:\n\n{escape(str(traceback.format_exc()))}",
                 }
             ),
-            400,
+            status_code=400,
         )
 
     # Save the config to file
@@ -268,14 +269,14 @@ def config_save():
             f.write(new_config)
             f.close()
     except Exception:
-        return make_response(
+        return JSONResponse(
-            jsonify(
+            content=(
                 {
                     "success": False,
                     "message": "Could not write config file, be sure that Frigate has write permission on the config file.",
                 }
             ),
-            400,
+            status_code=400,
         )
 
     if save_option == "restart":
@@ -283,34 +284,34 @@ def config_save():
             restart_frigate()
         except Exception as e:
             logging.error(f"Error restarting Frigate: {e}")
-            return make_response(
+            return JSONResponse(
-                jsonify(
+                content=(
                     {
                         "success": True,
                         "message": "Config successfully saved, unable to restart Frigate",
                     }
                 ),
-                200,
+                status_code=200,
             )
 
-        return make_response(
+        return JSONResponse(
-            jsonify(
+            content=(
                 {
                     "success": True,
                     "message": "Config successfully saved, restarting (this can take up to one minute)...",
                 }
             ),
-            200,
+            status_code=200,
         )
     else:
-        return make_response(
+        return JSONResponse(
-            jsonify({"success": True, "message": "Config successfully saved."}),
+            content=({"success": True, "message": "Config successfully saved."}),
-            200,
+            status_code=200,
         )
 
 
-@bp.route("/config/set", methods=["PUT"])
+@router.put("/config/set")
-def config_set():
+def config_set(body: AppConfigSetBody):
     config_file = os.environ.get("CONFIG_FILE", f"{CONFIG_DIR}/config.yml")
 
     # Check if we can use .yaml instead of .yml
@@ -336,68 +337,68 @@ def config_set():
                 f.write(old_raw_config)
                 f.close()
             logger.error(f"\nConfig Error:\n\n{str(traceback.format_exc())}")
-            return make_response(
+            return JSONResponse(
-                jsonify(
+                content=(
                     {
                         "success": False,
                         "message": "Error parsing config. Check logs for error message.",
                     }
                 ),
-                400,
+                status_code=400,
             )
     except Exception as e:
         logging.error(f"Error updating config: {e}")
-        return make_response(
+        return JSONResponse(
-            jsonify({"success": False, "message": "Error updating config"}),
+            content=({"success": False, "message": "Error updating config"}),
-            500,
+            status_code=500,
         )
 
-    json = request.get_json(silent=True) or {}
+    if body.requires_restart == 0:
-
+        request.app.frigate_config = FrigateConfig.runtime_config(
-    if json.get("requires_restart", 1) == 0:
+            config_obj, request.app.plus_api
-        current_app.frigate_config = FrigateConfig.runtime_config(
-            config_obj, current_app.plus_api
         )
 
-    return make_response(
+    return JSONResponse(
-        jsonify(
+        content=(
             {
                 "success": True,
                 "message": "Config successfully updated, restart to apply",
             }
         ),
-        200,
+        status_code=200,
     )
 
 
-@bp.route("/ffprobe", methods=["GET"])
+@router.get("/ffprobe")
-def ffprobe():
+def ffprobe(request: Request, paths: str = ""):
-    path_param = request.args.get("paths", "")
+    path_param = paths
 
     if not path_param:
-        return make_response(
+        return JSONResponse(
-            jsonify({"success": False, "message": "Path needs to be provided."}), 404
+            content=({"success": False, "message": "Path needs to be provided."}),
+            status_code=404,
         )
 
     if path_param.startswith("camera"):
         camera = path_param[7:]
 
-        if camera not in current_app.frigate_config.cameras.keys():
+        if camera not in request.app.frigate_config.cameras.keys():
-            return make_response(
+            return JSONResponse(
-                jsonify(
+                content=(
                     {"success": False, "message": f"{camera} is not a valid camera."}
                 ),
-                404,
+                status_code=404,
             )
 
-        if not current_app.frigate_config.cameras[camera].enabled:
+        if not request.app.frigate_config.cameras[camera].enabled:
-            return make_response(
+            return JSONResponse(
-                jsonify({"success": False, "message": f"{camera} is not enabled."}), 404
+                content=({"success": False, "message": f"{camera} is not enabled."}),
+                status_code=404,
             )
 
         paths = map(
             lambda input: input.path,
-            current_app.frigate_config.cameras[camera].ffmpeg.inputs,
+            request.app.frigate_config.cameras[camera].ffmpeg.inputs,
         )
     elif "," in clean_camera_user_pass(path_param):
         paths = path_param.split(",")
@@ -408,7 +409,7 @@ def ffprobe():
     output = []
 
     for path in paths:
-        ffprobe = ffprobe_stream(current_app.frigate_config.ffmpeg, path.strip())
+        ffprobe = ffprobe_stream(request.app.frigate_config.ffmpeg, path.strip())
         output.append(
             {
                 "return_code": ffprobe.returncode,
@@ -425,14 +426,14 @@ def ffprobe():
             }
         )
 
-    return jsonify(output)
+    return JSONResponse(content=output)
 
 
-@bp.route("/vainfo", methods=["GET"])
+@router.get("/vainfo")
 def vainfo():
     vainfo = vainfo_hwaccel()
-    return jsonify(
+    return JSONResponse(
-        {
+        content={
             "return_code": vainfo.returncode,
             "stderr": (
                 vainfo.stderr.decode("unicode_escape").strip()
@@ -539,37 +540,35 @@ def logs(
         )
 
 
-@bp.route("/restart", methods=["POST"])
+@router.post("/restart")
 def restart():
     try:
         restart_frigate()
     except Exception as e:
         logging.error(f"Error restarting Frigate: {e}")
-        return make_response(
+        return JSONResponse(
-            jsonify(
+            content=(
                 {
                     "success": False,
                     "message": "Unable to restart Frigate.",
                 }
             ),
-            500,
+            status_code=500,
         )
 
-    return make_response(
+    return JSONResponse(
-        jsonify(
+        content=(
             {
                 "success": True,
                 "message": "Restarting (this can take up to one minute)...",
             }
         ),
-        200,
+        status_code=200,
     )
 
 
-@bp.route("/labels")
+@router.get("/labels")
-def get_labels():
+def get_labels(camera: str = ""):
-    camera = request.args.get("camera", type=str, default="")
-
     try:
         if camera:
             events = Event.select(Event.label).where(Event.camera == camera).distinct()
@@ -577,24 +576,23 @@ def get_labels():
             events = Event.select(Event.label).distinct()
     except Exception as e:
         logger.error(e)
-        return make_response(
+        return JSONResponse(
-            jsonify({"success": False, "message": "Failed to get labels"}), 404
+            content=({"success": False, "message": "Failed to get labels"}),
+            status_code=404,
         )
 
     labels = sorted([e.label for e in events])
-    return jsonify(labels)
+    return JSONResponse(content=labels)
 
 
-@bp.route("/sub_labels")
+@router.get("/sub_labels")
-def get_sub_labels():
+def get_sub_labels(split_joined: int):
-    split_joined = request.args.get("split_joined", type=int)
-
     try:
         events = Event.select(Event.sub_label).distinct()
     except Exception:
-        return make_response(
+        return JSONResponse(
-            jsonify({"success": False, "message": "Failed to get sub_labels"}),
+            content=({"success": False, "message": "Failed to get sub_labels"}),
-            404,
+            status_code=404,
         )
 
     sub_labels = [e.sub_label for e in events]
@@ -615,15 +613,11 @@ def get_sub_labels():
                         sub_labels.append(part.strip())
 
     sub_labels.sort()
-    return jsonify(sub_labels)
+    return JSONResponse(content=sub_labels)
 
 
-@bp.route("/timeline")
+@router.get("/timeline")
-def timeline():
+def timeline(camera: str = "all", limit: int = 100, source_id: Optional[str] = None):
-    camera = request.args.get("camera", "all")
-    source_id = request.args.get("source_id", type=str)
-    limit = request.args.get("limit", 100)
-
     clauses = []
 
     selected_columns = [
@@ -652,18 +646,18 @@ def timeline():
         .dicts()
     )
 
-    return jsonify([t for t in timeline])
+    return JSONResponse(content=[t for t in timeline])
 
 
-@bp.route("/timeline/hourly")
+@router.get("/timeline/hourly")
-def hourly_timeline():
+def hourly_timeline(params: AppTimelineHourlyQueryParameters = Depends()):
     """Get hourly summary for timeline."""
-    cameras = request.args.get("cameras", "all")
+    cameras = params.cameras
-    labels = request.args.get("labels", "all")
+    labels = params.labels
-    before = request.args.get("before", type=float)
+    before = params.before
-    after = request.args.get("after", type=float)
+    after = params.after
-    limit = request.args.get("limit", 200)
+    limit = params.limit
-    tz_name = request.args.get("timezone", default="utc", type=str)
+    tz_name = params.timezone
 
     _, minute_modifier, _ = get_tz_modifiers(tz_name)
     minute_offset = int(minute_modifier.split(" ")[0])
@@ -729,8 +723,8 @@ def hourly_timeline():
         else:
             hours[hour].insert(0, t)
 
-    return jsonify(
+    return JSONResponse(
-        {
+        content={
             "start": start,
             "end": end,
             "count": count,

frigate/api/defs/events_query_parameters.py

@@ -42,7 +42,6 @@ class EventsSearchQueryParams(BaseModel):
     zones: Optional[str] = "all"
     after: Optional[float] = None
     before: Optional[float] = None
-
     timezone: Optional[str] = "utc"