Simplify UUID validation with type annotations (#5027)

Co-authored-by: Joao Vilaca <[email protected]>

Author: itallix

application/backend/app/api/dependencies.py

@@ -4,18 +4,20 @@
 from collections.abc import Generator
 from pathlib import Path
 from typing import Annotated
-from uuid import UUID
 
 from fastapi import Depends, HTTPException, Request, UploadFile, status
 from sqlalchemy.orm import Session
 
+from app.api.validators import ProjectID, SinkID, SourceID
 from app.core.jobs.control_plane import JobQueue
 from app.db import get_db_session
 from app.models import Sink, Source
 from app.scheduler import Scheduler
 from app.schemas import ProjectView
 from app.services import (
+    BaseWeightsService,
     DatasetService,
+    LabelService,
     MetricsService,
     ModelService,
     PipelineMetricsService,
@@ -26,41 +28,12 @@
     SourceUpdateService,
     SystemService,
 )
-from app.services.base_weights_service import BaseWeightsService
 from app.services.data_collect import DataCollector
 from app.services.event.event_bus import EventBus
-from app.services.label_service import LabelService
 from app.services.training_configuration_service import TrainingConfigurationService
 from app.webrtc.manager import WebRTCManager
 
 
-def is_valid_uuid(identifier: str) -> bool:
-    """
-    Check if a given string identifier is formatted as a valid UUID.
-
-    Args:
-        identifier: String to check for UUID validity.
-
-    Returns:
-        bool: True if the string is a valid UUID, False otherwise.
-
-    Example:
-        >>> is_valid_uuid("550e8400-e29b-41d4-a716-446655440000")
-        True
-        >>> is_valid_uuid("invalid-uuid")
-        False
-
-    Note:
-        This function only validates the UUID format, not whether the UUID
-        actually exists in any system or database.
-    """
-    try:
-        UUID(identifier)
-    except ValueError:
-        return False
-    return True
-
-
 def get_file_name_and_extension(file: UploadFile) -> tuple[str, str]:
     """Return the file name and extension"""
     if not file.filename:
@@ -83,41 +56,6 @@ def get_file_size(file: UploadFile) -> int:
     return file.size
 
 
-def get_source_id(source_id: str) -> UUID:
-    """Initializes and validates a source ID"""
-    if not is_valid_uuid(source_id):
-        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid source ID")
-    return UUID(source_id)
-
-
-def get_project_id(project_id: str) -> UUID:
-    """Initializes and validates a project ID"""
-    if not is_valid_uuid(project_id):
-        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid project ID")
-    return UUID(project_id)
-
-
-def get_sink_id(sink_id: str) -> UUID:
-    """Initializes and validates a sink ID"""
-    if not is_valid_uuid(sink_id):
-        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid sink ID")
-    return UUID(sink_id)
-
-
-def get_model_id(model_id: str) -> UUID:
-    """Initializes and validates a model ID"""
-    if not is_valid_uuid(model_id):
-        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid model ID")
-    return UUID(model_id)
-
-
-def get_dataset_item_id(dataset_item_id: str) -> UUID:
-    """Initializes and validates a dataset item ID"""
-    if not is_valid_uuid(dataset_item_id):
-        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid dataset item ID")
-    return UUID(dataset_item_id)
-
-
 def get_db() -> Generator[Session]:
     """Provides a database session."""
     with get_db_session() as session:
@@ -231,7 +169,7 @@ def get_dataset_service(
 
 
 def get_project(
-    project_id: Annotated[UUID, Depends(get_project_id)],
+    project_id: ProjectID,
     project_service: Annotated[ProjectService, Depends(get_project_service)],
 ) -> ProjectView:
     """Provides a ProjectView instance for request scoped project."""
@@ -242,7 +180,7 @@ def get_project(
 
 
 def get_sink(
-    sink_id: Annotated[UUID, Depends(get_sink_id)],
+    sink_id: SinkID,
     sink_service: Annotated[SinkService, Depends(get_sink_service)],
 ) -> Sink:
     """Provides a Sink instance for request scoped sink."""
@@ -253,7 +191,7 @@ def get_sink(
 
 
 def get_source(
-    source_id: Annotated[UUID, Depends(get_source_id)],
+    source_id: SourceID,
     source_update_service: Annotated[SourceUpdateService, Depends(get_source_update_service)],
 ) -> Source:
     """Provides a Source instance for request scoped source."""

application/backend/app/api/routers/datasets.py

@@ -8,14 +8,15 @@
 from fastapi.openapi.models import Example
 from starlette.responses import FileResponse
 
-from app.api.dependencies import get_dataset_item_id, get_dataset_service, get_file_name_and_extension, get_project
+from app.api.dependencies import get_dataset_service, get_file_name_and_extension, get_project
 from app.api.schemas.dataset_item import (
     DatasetItemAnnotations,
     DatasetItemAssignSubset,
     DatasetItemsWithPagination,
     DatasetItemView,
     SetDatasetItemAnnotations,
 )
+from app.api.validators import DatasetItemID
 from app.core.models import Pagination
 from app.models import DatasetItemAnnotationStatus, DatasetItemSubset
 from app.schemas import ProjectView
@@ -159,7 +160,7 @@ def list_dataset_items(  # noqa: PLR0913
 )
 def get_dataset_item(
     project: Annotated[ProjectView, Depends(get_project)],
-    dataset_item_id: Annotated[UUID, Depends(get_dataset_item_id)],
+    dataset_item_id: DatasetItemID,
     dataset_service: Annotated[DatasetService, Depends(get_dataset_service)],
 ) -> DatasetItemView:
     """Get information about a specific dataset item"""
@@ -180,7 +181,7 @@ def get_dataset_item(
 )
 def get_dataset_item_binary(
     project: Annotated[ProjectView, Depends(get_project)],
-    dataset_item_id: Annotated[UUID, Depends(get_dataset_item_id)],
+    dataset_item_id: DatasetItemID,
     dataset_service: Annotated[DatasetService, Depends(get_dataset_service)],
 ) -> FileResponse:
     """Get dataset item binary content"""
@@ -203,7 +204,7 @@ def get_dataset_item_binary(
 )
 def get_dataset_item_thumbnail(
     project: Annotated[ProjectView, Depends(get_project)],
-    dataset_item_id: Annotated[UUID, Depends(get_dataset_item_id)],
+    dataset_item_id: DatasetItemID,
     dataset_service: Annotated[DatasetService, Depends(get_dataset_service)],
 ) -> FileResponse:
     """Get dataset item thumbnail binary content"""
@@ -227,7 +228,7 @@ def get_dataset_item_thumbnail(
 )
 def delete_dataset_item(
     project: Annotated[ProjectView, Depends(get_project)],
-    dataset_item_id: Annotated[UUID, Depends(get_dataset_item_id)],
+    dataset_item_id: DatasetItemID,
     dataset_service: Annotated[DatasetService, Depends(get_dataset_service)],
 ) -> None:
     """Delete an item from the dataset"""
@@ -248,7 +249,7 @@ def delete_dataset_item(
 )
 def set_dataset_item_annotations(
     project: Annotated[ProjectView, Depends(get_project)],
-    dataset_item_id: Annotated[UUID, Depends(get_dataset_item_id)],
+    dataset_item_id: DatasetItemID,
     dataset_item_annotations: Annotated[
         SetDatasetItemAnnotations, Body(openapi_examples=SET_DATASET_ITEM_ANNOTATIONS_BODY_EXAMPLES)
     ],
@@ -283,7 +284,7 @@ def set_dataset_item_annotations(
 )
 def get_dataset_item_annotations(
     project: Annotated[ProjectView, Depends(get_project)],
-    dataset_item_id: Annotated[UUID, Depends(get_dataset_item_id)],
+    dataset_item_id: DatasetItemID,
     dataset_service: Annotated[DatasetService, Depends(get_dataset_service)],
 ) -> DatasetItemAnnotations:
     """Get the dataset item annotations"""
@@ -313,7 +314,7 @@ def get_dataset_item_annotations(
 )
 def delete_dataset_item_annotation(
     project: Annotated[ProjectView, Depends(get_project)],
-    dataset_item_id: Annotated[UUID, Depends(get_dataset_item_id)],
+    dataset_item_id: DatasetItemID,
     dataset_service: Annotated[DatasetService, Depends(get_dataset_service)],
 ) -> None:
     """Delete dataset item annotations"""
@@ -336,7 +337,7 @@ def delete_dataset_item_annotation(
 )
 def assign_dataset_item_subset(
     project: Annotated[ProjectView, Depends(get_project)],
-    dataset_item_id: Annotated[UUID, Depends(get_dataset_item_id)],
+    dataset_item_id: DatasetItemID,
     dataset_service: Annotated[DatasetService, Depends(get_dataset_service)],
     subset_config: Annotated[DatasetItemAssignSubset, Body()],
 ) -> DatasetItemView:

application/backend/app/api/routers/models.py

@@ -2,12 +2,12 @@
 # SPDX-License-Identifier: Apache-2.0
 
 from typing import Annotated
-from uuid import UUID
 
 from fastapi import APIRouter, Depends, HTTPException, status
 
-from app.api.dependencies import get_model_id, get_model_service, get_project_id
-from app.schemas import Model
+from app.api.dependencies import get_model_service, get_project
+from app.api.validators import ModelID
+from app.schemas import Model, ProjectView
 from app.services import ModelService, ResourceInUseError, ResourceNotFoundError
 
 router = APIRouter(prefix="/api/projects/{project_id}/models", tags=["Models"])
@@ -23,12 +23,12 @@
     },
 )
 def list_models(
-    project_id: Annotated[UUID, Depends(get_project_id)],
+    project: Annotated[ProjectView, Depends(get_project)],
     model_service: Annotated[ModelService, Depends(get_model_service)],
 ) -> list[Model]:
     """Get all models in a project."""
     try:
-        return model_service.list_models(project_id)
+        return model_service.list_models(project.id)
     except ResourceNotFoundError:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Project not found")
 
@@ -43,13 +43,13 @@ def list_models(
     },
 )
 def get_model(
-    project_id: Annotated[UUID, Depends(get_project_id)],
-    model_id: Annotated[UUID, Depends(get_model_id)],
+    project: Annotated[ProjectView, Depends(get_project)],
+    model_id: ModelID,
     model_service: Annotated[ModelService, Depends(get_model_service)],
 ) -> Model:
     """Get a specific model by ID."""
     try:
-        return model_service.get_model_by_id(project_id=project_id, model_id=model_id)
+        return model_service.get_model(project_id=project.id, model_id=model_id)
     except ResourceNotFoundError as e:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e))
 
@@ -67,13 +67,13 @@ def get_model(
     },
 )
 def delete_model(
-    project_id: Annotated[UUID, Depends(get_project_id)],
-    model_id: Annotated[UUID, Depends(get_model_id)],
+    project: Annotated[ProjectView, Depends(get_project)],
+    model_id: ModelID,
     model_service: Annotated[ModelService, Depends(get_model_service)],
 ) -> None:
     """Delete a model from a project."""
     try:
-        model_service.delete_model_by_id(project_id=project_id, model_id=model_id)
+        model_service.delete_model(project_id=project.id, model_id=model_id)
     except ResourceNotFoundError as e:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e))
     except ResourceInUseError as e:

application/backend/app/api/routers/pipelines.py

@@ -4,14 +4,14 @@
 """Endpoints for managing pipelines"""
 
 from typing import Annotated
-from uuid import UUID
 
 from fastapi import APIRouter, Body, Depends, status
 from fastapi.exceptions import HTTPException
 from fastapi.openapi.models import Example
 from pydantic import ValidationError
 
-from app.api.dependencies import get_pipeline_metrics_service, get_pipeline_service, get_project_id
+from app.api.dependencies import get_pipeline_metrics_service, get_pipeline_service
+from app.api.validators import ProjectID
 from app.schemas.metrics import PipelineMetrics
 from app.schemas.pipeline import DataCollectionPolicyAdapter, PipelineStatus, PipelineView
 from app.services import PipelineMetricsService, PipelineService, ResourceNotFoundError
@@ -70,7 +70,7 @@
     },
 )
 def get_pipeline(
-    project_id: Annotated[UUID, Depends(get_project_id)],
+    project_id: ProjectID,
     pipeline_service: Annotated[PipelineService, Depends(get_pipeline_service)],
 ) -> PipelineView:
     """Get info about a given pipeline"""
@@ -91,7 +91,7 @@ def get_pipeline(
     },
 )
 def update_pipeline(
-    project_id: Annotated[UUID, Depends(get_project_id)],
+    project_id: ProjectID,
     pipeline_config: Annotated[
         dict,
         Body(
@@ -128,7 +128,7 @@ def update_pipeline(
     },
 )
 def enable_pipeline(
-    project_id: Annotated[UUID, Depends(get_project_id)],
+    project_id: ProjectID,
     pipeline_service: Annotated[PipelineService, Depends(get_pipeline_service)],
 ) -> None:
     """
@@ -153,7 +153,7 @@ def enable_pipeline(
     },
 )
 def disable_pipeline(
-    project_id: Annotated[UUID, Depends(get_project_id)],
+    project_id: ProjectID,
     pipeline_service: Annotated[PipelineService, Depends(get_pipeline_service)],
 ) -> None:
     """Stop a pipeline. The pipeline will become idle, and it won't process any data until re-enabled."""
@@ -173,7 +173,7 @@ def disable_pipeline(
     },
 )
 def get_project_metrics(
-    project_id: Annotated[UUID, Depends(get_project_id)],
+    project_id: ProjectID,
     pipeline_metrics_service: Annotated[PipelineMetricsService, Depends(get_pipeline_metrics_service)],
     time_window: int = 60,
 ) -> PipelineMetrics: