Add ability to update config via json body to config/set endpoint

Additionally, update the config in a single rather than multiple calls for each updated key

frigate/api/app.py

@@ -6,6 +6,7 @@ import json
 import logging
 import os
 import traceback
+import urllib
 from datetime import datetime, timedelta
 from functools import reduce
 from io import StringIO
@@ -36,8 +37,10 @@ from frigate.models import Event, Timeline
 from frigate.stats.prometheus import get_metrics, update_metrics
 from frigate.util.builtin import (
     clean_camera_user_pass,
+    flatten_config_data,
     get_tz_modifiers,
-    update_yaml_from_url,
+    process_config_query_string,
+    update_yaml_file_bulk,
 )
 from frigate.util.config import find_config_file
 from frigate.util.services import (
@@ -358,14 +361,37 @@ def config_set(request: Request, body: AppConfigSetBody):
 
     with open(config_file, "r") as f:
         old_raw_config = f.read()
-        f.close()
 
     try:
-        update_yaml_from_url(config_file, str(request.url))
+        updates = {}
+
+        # process query string parameters (takes precedence over body.config_data)
+        parsed_url = urllib.parse.urlparse(str(request.url))
+        query_string = urllib.parse.parse_qs(parsed_url.query, keep_blank_values=True)
+
+        # Filter out empty keys but keep blank values for non-empty keys
+        query_string = {k: v for k, v in query_string.items() if k}
+
+        if query_string:
+            updates = process_config_query_string(query_string)
+        elif body.config_data:
+            updates = flatten_config_data(body.config_data)
+
+        if not updates:
+            return JSONResponse(
+                content=(
+                    {"success": False, "message": "No configuration data provided"}
+                ),
+                status_code=400,
+            )
+
+        # apply all updates in a single operation
+        update_yaml_file_bulk(config_file, updates)
+
+        # validate the updated config
         with open(config_file, "r") as f:
             new_raw_config = f.read()
-            f.close()
-        # Validate the config schema
+
         try:
             config = FrigateConfig.parse(new_raw_config)
         except Exception:

frigate/api/defs/request/app_body.py

@@ -1,4 +1,4 @@
-from typing import Optional
+from typing import Any, Dict, Optional
 
 from pydantic import BaseModel
 
@@ -6,6 +6,7 @@ from pydantic import BaseModel
 class AppConfigSetBody(BaseModel):
     requires_restart: int = 1
     update_topic: str | None = None
+    config_data: Optional[Dict[str, Any]] = None
 
 
 class AppPutPasswordBody(BaseModel):

frigate/util/builtin.py

@@ -14,7 +14,7 @@ import urllib.parse
 from collections.abc import Mapping
 from multiprocessing.sharedctypes import Synchronized
 from pathlib import Path
-from typing import Any, Optional, Tuple, Union
+from typing import Any, Dict, Optional, Tuple, Union
 from zoneinfo import ZoneInfoNotFoundError
 
 import numpy as np
@@ -184,26 +184,12 @@ def create_mask(frame_shape, mask):
     mask_img[:] = 255
 
 
-def update_yaml_from_url(file_path: str, url: str):
-    parsed_url = urllib.parse.urlparse(url)
-    query_string = urllib.parse.parse_qs(parsed_url.query, keep_blank_values=True)
-
-    # Filter out empty keys but keep blank values for non-empty keys
-    query_string = {k: v for k, v in query_string.items() if k}
-
+def process_config_query_string(query_string: Dict[str, list]) -> Dict[str, Any]:
+    updates = {}
     for key_path_str, new_value_list in query_string.items():
-        key_path = key_path_str.split(".")
-        key_path_copy = key_path.copy()
-        for i in range(len(key_path_copy)):
-            try:
-                index = int(key_path_copy[i])
-                key_path[i] = (key_path_copy[i - 1], index)
-                key_path.pop(i - 1)
-            except ValueError:
-                pass
-
+        # use the string key as-is for updates dictionary
         if len(new_value_list) > 1:
-            update_yaml_file(file_path, key_path, new_value_list)
+            updates[key_path_str] = new_value_list
         else:
             value = new_value_list[0]
             try:
@@ -211,10 +197,24 @@ def update_yaml_from_url(file_path: str, url: str):
                 value = ast.literal_eval(value) if "," not in value else value
             except (ValueError, SyntaxError):
                 pass
-            update_yaml_file(file_path, key_path, value)
+            updates[key_path_str] = value
+    return updates
 
 
-def update_yaml_file(file_path: str, key_path: str, new_value: Any):
+def flatten_config_data(
+    config_data: Dict[str, Any], parent_key: str = ""
+) -> Dict[str, Any]:
+    items = []
+    for key, value in config_data.items():
+        new_key = f"{parent_key}.{key}" if parent_key else key
+        if isinstance(value, dict):
+            items.extend(flatten_config_data(value, new_key).items())
+        else:
+            items.append((new_key, value))
+    return dict(items)
+
+
+def update_yaml_file_bulk(file_path: str, updates: Dict[str, Any]):
     yaml = YAML()
     yaml.indent(mapping=2, sequence=4, offset=2)
 
@@ -227,7 +227,17 @@ def update_yaml_file(file_path: str, key_path: str, new_value: Any):
         )
         return
 
-    data = update_yaml(data, key_path, new_value)
+    # Apply all updates
+    for key_path_str, new_value in updates.items():
+        key_path = key_path_str.split(".")
+        for i in range(len(key_path)):
+            try:
+                index = int(key_path[i])
+                key_path[i] = (key_path[i - 1], index)
+                key_path.pop(i - 1)
+            except ValueError:
+                pass
+        data = update_yaml(data, key_path, new_value)
 
     try:
         with open(file_path, "w") as f: