Create scaffolding for case management (#21293)

frigate/api/defs/request/export_case_body.py

@@ -0,0 +1,25 @@
+from typing import Optional
+
+from pydantic import BaseModel, Field
+
+
+class ExportCaseCreateBody(BaseModel):
+    """Request body for creating a new export case."""
+
+    name: str = Field(max_length=100, description="Friendly name of the export case")
+    description: Optional[str] = Field(
+        default=None, description="Optional description of the export case"
+    )
+
+
+class ExportCaseUpdateBody(BaseModel):
+    """Request body for updating an existing export case."""
+
+    name: Optional[str] = Field(
+        default=None,
+        max_length=100,
+        description="Updated friendly name of the export case",
+    )
+    description: Optional[str] = Field(
+        default=None, description="Updated description of the export case"
+    )

frigate/api/defs/response/export_case_response.py

@@ -0,0 +1,22 @@
+from typing import List, Optional
+
+from pydantic import BaseModel, Field
+
+
+class ExportCaseModel(BaseModel):
+    """Model representing a single export case."""
+
+    id: str = Field(description="Unique identifier for the export case")
+    name: str = Field(description="Friendly name of the export case")
+    description: Optional[str] = Field(
+        default=None, description="Optional description of the export case"
+    )
+    created_at: float = Field(
+        description="Unix timestamp when the export case was created"
+    )
+    updated_at: float = Field(
+        description="Unix timestamp when the export case was last updated"
+    )
+
+
+ExportCasesResponse = List[ExportCaseModel]

frigate/api/defs/response/export_response.py

@@ -15,6 +15,9 @@ class ExportModel(BaseModel):
     in_progress: bool = Field(
         description="Whether the export is currently being processed"
     )
+    export_case_id: Optional[str] = Field(
+        default=None, description="ID of the export case this export belongs to"
+    )
 
 
 class StartExportResponse(BaseModel):

frigate/api/export.py

@@ -19,8 +19,16 @@ from frigate.api.auth import (
     require_camera_access,
     require_role,
 )
+from frigate.api.defs.request.export_case_body import (
+    ExportCaseCreateBody,
+    ExportCaseUpdateBody,
+)
 from frigate.api.defs.request.export_recordings_body import ExportRecordingsBody
 from frigate.api.defs.request.export_rename_body import ExportRenameBody
+from frigate.api.defs.response.export_case_response import (
+    ExportCaseModel,
+    ExportCasesResponse,
+)
 from frigate.api.defs.response.export_response import (
     ExportModel,
     ExportsResponse,
@@ -29,7 +37,7 @@ from frigate.api.defs.response.export_response import (
 from frigate.api.defs.response.generic_response import GenericResponse
 from frigate.api.defs.tags import Tags
 from frigate.const import CLIPS_DIR, EXPORT_DIR
-from frigate.models import Export, Previews, Recordings
+from frigate.models import Export, ExportCase, Previews, Recordings
 from frigate.record.export import (
     PlaybackFactorEnum,
     PlaybackSourceEnum,
@@ -63,6 +71,110 @@ def get_exports(
     return JSONResponse(content=[e for e in exports])
 
 
+@router.get(
+    "/cases",
+    response_model=ExportCasesResponse,
+    dependencies=[Depends(allow_any_authenticated())],
+    summary="Get export cases",
+    description="Gets all export cases from the database.",
+)
+def get_export_cases():
+    cases = (
+        ExportCase.select().order_by(ExportCase.created_at.desc()).dicts().iterator()
+    )
+    return JSONResponse(content=[c for c in cases])
+
+
+@router.post(
+    "/cases",
+    response_model=ExportCaseModel,
+    dependencies=[Depends(require_role(["admin"]))],
+    summary="Create export case",
+    description="Creates a new export case.",
+)
+def create_export_case(body: ExportCaseCreateBody):
+    case = ExportCase.create(
+        id="".join(random.choices(string.ascii_lowercase + string.digits, k=12)),
+        name=body.name,
+        description=body.description,
+        created_at=Path().stat().st_mtime,
+        updated_at=Path().stat().st_mtime,
+    )
+    return JSONResponse(content=model_to_dict(case))
+
+
+@router.get(
+    "/cases/{case_id}",
+    response_model=ExportCaseModel,
+    dependencies=[Depends(allow_any_authenticated())],
+    summary="Get a single export case",
+    description="Gets a specific export case by ID.",
+)
+def get_export_case(case_id: str):
+    try:
+        case = ExportCase.get(ExportCase.id == case_id)
+        return JSONResponse(content=model_to_dict(case))
+    except DoesNotExist:
+        return JSONResponse(
+            content={"success": False, "message": "Export case not found"},
+            status_code=404,
+        )
+
+
+@router.patch(
+    "/cases/{case_id}",
+    response_model=GenericResponse,
+    dependencies=[Depends(require_role(["admin"]))],
+    summary="Update export case",
+    description="Updates an existing export case.",
+)
+def update_export_case(case_id: str, body: ExportCaseUpdateBody):
+    try:
+        case = ExportCase.get(ExportCase.id == case_id)
+    except DoesNotExist:
+        return JSONResponse(
+            content={"success": False, "message": "Export case not found"},
+            status_code=404,
+        )
+
+    if body.name is not None:
+        case.name = body.name
+    if body.description is not None:
+        case.description = body.description
+
+    case.save()
+
+    return JSONResponse(
+        content={"success": True, "message": "Successfully updated export case."}
+    )
+
+
+@router.delete(
+    "/cases/{case_id}",
+    response_model=GenericResponse,
+    dependencies=[Depends(require_role(["admin"]))],
+    summary="Delete export case",
+    description="""Deletes an export case.\n    Exports that reference this case will have their export_case set to null.\n    """,
+)
+def delete_export_case(case_id: str):
+    try:
+        case = ExportCase.get(ExportCase.id == case_id)
+    except DoesNotExist:
+        return JSONResponse(
+            content={"success": False, "message": "Export case not found"},
+            status_code=404,
+        )
+
+    # Unassign exports from this case but keep the exports themselves
+    Export.update(export_case=None).where(Export.export_case == case).execute()
+
+    case.delete_instance()
+
+    return JSONResponse(
+        content={"success": True, "message": "Successfully deleted export case."}
+    )
+
+
 @router.post(
     "/export/{camera_name}/start/{start_time}/end/{end_time}",
     response_model=StartExportResponse,

frigate/models.py

@@ -80,6 +80,14 @@ class Recordings(Model):
     regions = IntegerField(null=True)
 
 
+class ExportCase(Model):
+    id = CharField(null=False, primary_key=True, max_length=30)
+    name = CharField(index=True, max_length=100)
+    description = TextField(null=True)
+    created_at = DateTimeField()
+    updated_at = DateTimeField()
+
+
 class Export(Model):
     id = CharField(null=False, primary_key=True, max_length=30)
     camera = CharField(index=True, max_length=20)
@@ -88,6 +96,12 @@ class Export(Model):
     video_path = CharField(unique=True)
     thumb_path = CharField(unique=True)
     in_progress = BooleanField()
+    export_case = ForeignKeyField(
+        ExportCase,
+        null=True,
+        backref="exports",
+        column_name="export_case_id",
+    )
 
 
 class ReviewSegment(Model):

migrations/033_create_export_case_table.py

@@ -0,0 +1,50 @@
+"""Peewee migrations -- 033_create_export_case_table.py.
+
+Some examples (model - class or model name)::
+
+    > Model = migrator.orm['model_name']            # Return model in current state by name
+
+    > migrator.sql(sql)                             # Run custom SQL
+    > migrator.python(func, *args, **kwargs)        # Run python code
+    > migrator.create_model(Model)                  # Create a model (could be used as decorator)
+    > migrator.remove_model(model, cascade=True)    # Remove a model
+    > migrator.add_fields(model, **fields)          # Add fields to a model
+    > migrator.change_fields(model, **fields)       # Change fields
+    > migrator.remove_fields(model, *field_names, cascade=True)
+    > migrator.rename_field(model, old_field_name, new_field_name)
+    > migrator.rename_table(model, new_table_name)
+    > migrator.add_index(model, *col_names, unique=False)
+    > migrator.drop_index(model, *col_names)
+    > migrator.add_not_null(model, *field_names)
+    > migrator.drop_not_null(model, *field_names)
+    > migrator.add_default(model, field_name, default)
+
+"""
+
+import peewee as pw
+
+SQL = pw.SQL
+
+
+def migrate(migrator, database, fake=False, **kwargs):
+    migrator.sql(
+        """
+        CREATE TABLE IF NOT EXISTS "exportcase" (
+            "id" VARCHAR(30) NOT NULL PRIMARY KEY,
+            "name" VARCHAR(100) NOT NULL,
+            "description" TEXT NULL,
+            "created_at" DATETIME NOT NULL,
+            "updated_at" DATETIME NOT NULL
+        )
+        """
+    )
+    migrator.sql(
+        'CREATE INDEX IF NOT EXISTS "exportcase_name" ON "exportcase" ("name")'
+    )
+    migrator.sql(
+        'CREATE INDEX IF NOT EXISTS "exportcase_created_at" ON "exportcase" ("created_at")'
+    )
+
+
+def rollback(migrator, database, fake=False, **kwargs):
+    pass

migrations/034_add_export_case_to_exports.py

@@ -0,0 +1,40 @@
+"""Peewee migrations -- 034_add_export_case_to_exports.py.
+
+Some examples (model - class or model name)::
+
+    > Model = migrator.orm['model_name']            # Return model in current state by name
+
+    > migrator.sql(sql)                             # Run custom SQL
+    > migrator.python(func, *args, **kwargs)        # Run python code
+    > migrator.create_model(Model)                  # Create a model (could be used as decorator)
+    > migrator.remove_model(model, cascade=True)    # Remove a model
+    > migrator.add_fields(model, **fields)          # Add fields to a model
+    > migrator.change_fields(model, **fields)       # Change fields
+    > migrator.remove_fields(model, *field_names, cascade=True)
+    > migrator.rename_field(model, old_field_name, new_field_name)
+    > migrator.rename_table(model, new_table_name)
+    > migrator.add_index(model, *col_names, unique=False)
+    > migrator.drop_index(model, *col_names)
+    > migrator.add_not_null(model, *field_names)
+    > migrator.drop_not_null(model, *field_names)
+    > migrator.add_default(model, field_name, default)
+
+"""
+
+import peewee as pw
+
+SQL = pw.SQL
+
+
+def migrate(migrator, database, fake=False, **kwargs):
+    # Add nullable export_case_id column to export table
+    migrator.sql('ALTER TABLE "export" ADD COLUMN "export_case_id" VARCHAR(30) NULL')
+
+    # Index for faster case-based queries
+    migrator.sql(
+        'CREATE INDEX IF NOT EXISTS "export_export_case_id" ON "export" ("export_case_id")'
+    )
+
+
+def rollback(migrator, database, fake=False, **kwargs):
+    pass