Refactoring: Event Bus & services decoupling (#4956)

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

Author: warrkan

application/backend/app/api/dependencies.py

@@ -12,18 +12,21 @@
 from app.core.jobs.control_plane import JobQueue
 from app.db import get_db_session
 from app.scheduler import Scheduler
+from app.schemas import ProjectView
 from app.services import (
-    ActivePipelineService,
     ConfigurationService,
     DatasetService,
     MetricsService,
     ModelService,
+    PipelineMetricsService,
     PipelineService,
     ProjectService,
+    ResourceNotFoundError,
     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.webrtc.manager import WebRTCManager
 
@@ -128,9 +131,9 @@ def get_data_dir(request: Request) -> Path:
     return request.app.state.settings.data_dir
 
 
-def get_active_pipeline_service(request: Request) -> ActivePipelineService:
-    """Provides an ActivePipelineService instance for managing the active pipeline."""
-    return request.app.state.active_pipeline_service
+def get_event_bus(request: Request) -> EventBus:
+    """Provides an EventBus instance."""
+    return request.app.state.event_bus
 
 
 def get_data_collector(request: Request) -> DataCollector:
@@ -144,32 +147,29 @@ def get_metrics_service(scheduler: Annotated[Scheduler, Depends(get_scheduler)])
 
 
 def get_configuration_service(
-    active_pipeline_service: Annotated[ActivePipelineService, Depends(get_active_pipeline_service)],
-    scheduler: Annotated[Scheduler, Depends(get_scheduler)],
+    event_bus: Annotated[EventBus, Depends(get_event_bus)],
     db: Annotated[Session, Depends(get_db)],
 ) -> ConfigurationService:
-    """Provides a ConfigurationService instance with the active pipeline service and config changed condition."""
-    return ConfigurationService(
-        active_pipeline_service=active_pipeline_service,
-        db_session=db,
-        config_changed_condition=scheduler.mp_config_changed_condition,
-    )
+    """Provides a ConfigurationService instance."""
+    return ConfigurationService(event_bus=event_bus, db_session=db)
 
 
 def get_pipeline_service(
-    active_pipeline_service: Annotated[ActivePipelineService, Depends(get_active_pipeline_service)],
-    data_collector: Annotated[DataCollector, Depends(get_data_collector)],
-    metrics_service: Annotated[MetricsService, Depends(get_metrics_service)],
-    scheduler: Annotated[Scheduler, Depends(get_scheduler)],
+    event_bus: Annotated[EventBus, Depends(get_event_bus)],
     db: Annotated[Session, Depends(get_db)],
 ) -> PipelineService:
-    """Provides a PipelineService instance with the active pipeline service and config changed condition."""
-    return PipelineService(
-        active_pipeline_service=active_pipeline_service,
-        data_collector=data_collector,
+    """Provides a PipelineService instance ."""
+    return PipelineService(event_bus=event_bus, db_session=db)
+
+
+def get_pipeline_metrics_service(
+    pipeline_service: Annotated[PipelineService, Depends(get_pipeline_service)],
+    metrics_service: Annotated[MetricsService, Depends(get_metrics_service)],
+) -> PipelineMetricsService:
+    """Provides a PipelineMetricsService instance."""
+    return PipelineMetricsService(
+        pipeline_service=pipeline_service,
         metrics_service=metrics_service,
-        config_changed_condition=scheduler.mp_config_changed_condition,
-        db_session=db,
     )
 
 
@@ -198,22 +198,31 @@ def get_label_service(db: Annotated[Session, Depends(get_db)]) -> LabelService:
 def get_project_service(
     data_dir: Annotated[Path, Depends(get_data_dir)],
     db: Annotated[Session, Depends(get_db)],
+    pipeline_service: Annotated[PipelineService, Depends(get_pipeline_service)],
     label_service: Annotated[LabelService, Depends(get_label_service)],
 ) -> ProjectService:
     """Provides a ProjectService instance for managing projects."""
-    return ProjectService(data_dir=data_dir, db_session=db, label_service=label_service)
+    return ProjectService(
+        data_dir=data_dir, db_session=db, label_service=label_service, pipeline_service=pipeline_service
+    )
 
 
 def get_dataset_service(
-    data_dir: Annotated[Path, Depends(get_data_dir)],
-    db: Annotated[Session, Depends(get_db)],
-    project_service: Annotated[ProjectService, Depends(get_project_service)],
-    label_service: Annotated[LabelService, Depends(get_label_service)],
+    data_dir: Annotated[Path, Depends(get_data_dir)], db: Annotated[Session, Depends(get_db)]
 ) -> DatasetService:
     """Provides a DatasetService instance."""
-    return DatasetService(
-        data_dir=data_dir, db_session=db, project_service=project_service, label_service=label_service
-    )
+    return DatasetService(data_dir=data_dir, db_session=db)
+
+
+def get_project(
+    project_id: Annotated[UUID, Depends(get_project_id)],
+    project_service: Annotated[ProjectService, Depends(get_project_service)],
+) -> ProjectView:
+    """Provides a ProjectView instance for request scoped project."""
+    try:
+        return project_service.get_project_by_id(project_id)
+    except ResourceNotFoundError as e:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e))
 
 
 def get_base_weights_service(data_dir: Annotated[Path, Depends(get_data_dir)]) -> BaseWeightsService:

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

@@ -8,9 +8,9 @@
 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_id
+from app.api.dependencies import get_dataset_item_id, get_dataset_service, get_file_name_and_extension, get_project
 from app.core.models import Pagination
-from app.schemas import DatasetItem, DatasetItemsWithPagination
+from app.schemas import DatasetItem, DatasetItemsWithPagination, ProjectView
 from app.schemas.dataset_item import (
     DatasetItemAnnotation,
     DatasetItemAnnotationsWithSource,
@@ -74,7 +74,7 @@
     },
 )
 def add_dataset_item(
-    project_id: Annotated[UUID, Depends(get_project_id)],
+    project: Annotated[ProjectView, Depends(get_project)],
     dataset_service: Annotated[DatasetService, Depends(get_dataset_service)],
     file_name_and_extension: Annotated[tuple[str, str], Depends(get_file_name_and_extension)],
     file: Annotated[UploadFile, File()],
@@ -83,7 +83,7 @@ def add_dataset_item(
     name, format = file_name_and_extension
     try:
         return dataset_service.create_dataset_item(
-            project_id=project_id,
+            project=project,
             data=file.file,
             name=name,
             format=format,
@@ -100,7 +100,7 @@ def add_dataset_item(
     },
 )
 def list_dataset_items(
-    project_id: Annotated[UUID, Depends(get_project_id)],
+    project: Annotated[ProjectView, Depends(get_project)],
     dataset_service: Annotated[DatasetService, Depends(get_dataset_service)],
     limit: Annotated[int, Query(ge=1, le=MAX_DATASET_ITEMS_NUMBER_RETURNED)] = DEFAULT_DATASET_ITEMS_NUMBER_RETURNED,
     offset: Annotated[int, Query(ge=0)] = 0,
@@ -112,9 +112,9 @@ def list_dataset_items(
         raise HTTPException(
             status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Start date must be before end date."
         )
-    total = dataset_service.count_dataset_items(project_id=project_id, start_date=start_date, end_date=end_date)
+    total = dataset_service.count_dataset_items(project=project, start_date=start_date, end_date=end_date)
     dataset_items = dataset_service.list_dataset_items(
-        project_id=project_id, limit=limit, offset=offset, start_date=start_date, end_date=end_date
+        project=project, limit=limit, offset=offset, start_date=start_date, end_date=end_date
     )
     return DatasetItemsWithPagination(
         items=dataset_items,
@@ -136,13 +136,13 @@ def list_dataset_items(
     },
 )
 def get_dataset_item(
-    project_id: Annotated[UUID, Depends(get_project_id)],
+    project: Annotated[ProjectView, Depends(get_project)],
     dataset_item_id: Annotated[UUID, Depends(get_dataset_item_id)],
     dataset_service: Annotated[DatasetService, Depends(get_dataset_service)],
 ) -> DatasetItem:
     """Get information about a specific dataset item"""
     try:
-        return dataset_service.get_dataset_item_by_id(project_id=project_id, dataset_item_id=dataset_item_id)
+        return dataset_service.get_dataset_item_by_id(project=project, dataset_item_id=dataset_item_id)
     except ResourceNotFoundError as e:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e))
 
@@ -156,14 +156,14 @@ def get_dataset_item(
     },
 )
 def get_dataset_item_binary(
-    project_id: Annotated[UUID, Depends(get_project_id)],
+    project: Annotated[ProjectView, Depends(get_project)],
     dataset_item_id: Annotated[UUID, Depends(get_dataset_item_id)],
     dataset_service: Annotated[DatasetService, Depends(get_dataset_service)],
 ) -> FileResponse:
     """Get dataset item binary content"""
     try:
         binary_path = dataset_service.get_dataset_item_binary_path_by_id(
-            project_id=project_id, dataset_item_id=dataset_item_id
+            project=project, dataset_item_id=dataset_item_id
         )
         return FileResponse(path=binary_path)
     except ResourceNotFoundError as e:
@@ -179,14 +179,14 @@ def get_dataset_item_binary(
     },
 )
 def get_dataset_item_thumbnail(
-    project_id: Annotated[UUID, Depends(get_project_id)],
+    project: Annotated[ProjectView, Depends(get_project)],
     dataset_item_id: Annotated[UUID, Depends(get_dataset_item_id)],
     dataset_service: Annotated[DatasetService, Depends(get_dataset_service)],
 ) -> FileResponse:
     """Get dataset item thumbnail binary content"""
     try:
         thumbnail_path = dataset_service.get_dataset_item_thumbnail_path_by_id(
-            project_id=project_id, dataset_item_id=dataset_item_id
+            project=project, dataset_item_id=dataset_item_id
         )
         return FileResponse(path=thumbnail_path)
     except ResourceNotFoundError as e:
@@ -203,13 +203,13 @@ def get_dataset_item_thumbnail(
     },
 )
 def delete_dataset_item(
-    project_id: Annotated[UUID, Depends(get_project_id)],
+    project: Annotated[ProjectView, Depends(get_project)],
     dataset_item_id: Annotated[UUID, Depends(get_dataset_item_id)],
     dataset_service: Annotated[DatasetService, Depends(get_dataset_service)],
 ) -> None:
     """Delete an item from the dataset"""
     try:
-        dataset_service.delete_dataset_item(project_id=project_id, dataset_item_id=dataset_item_id)
+        dataset_service.delete_dataset_item(project=project, dataset_item_id=dataset_item_id)
     except ResourceNotFoundError as e:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e))
 
@@ -224,7 +224,7 @@ def delete_dataset_item(
     },
 )
 def set_dataset_item_annotations(
-    project_id: Annotated[UUID, Depends(get_project_id)],
+    project: Annotated[ProjectView, Depends(get_project)],
     dataset_item_id: Annotated[UUID, Depends(get_dataset_item_id)],
     dataset_item_annotations: Annotated[
         SetDatasetItemAnnotations, Body(openapi_examples=SET_DATASET_ITEM_ANNOTATIONS_BODY_EXAMPLES)
@@ -234,7 +234,7 @@ def set_dataset_item_annotations(
     """Set dataset item annotations"""
     try:
         return dataset_service.set_dataset_item_annotations(
-            project_id=project_id, dataset_item_id=dataset_item_id, annotations=dataset_item_annotations.annotations
+            project=project, dataset_item_id=dataset_item_id, annotations=dataset_item_annotations.annotations
         )
     except ResourceNotFoundError as e:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e))
@@ -254,13 +254,13 @@ def set_dataset_item_annotations(
     },
 )
 def get_dataset_item_annotations(
-    project_id: Annotated[UUID, Depends(get_project_id)],
+    project: Annotated[ProjectView, Depends(get_project)],
     dataset_item_id: Annotated[UUID, Depends(get_dataset_item_id)],
     dataset_service: Annotated[DatasetService, Depends(get_dataset_service)],
 ) -> DatasetItemAnnotationsWithSource:
     """Get the dataset item annotations"""
     try:
-        return dataset_service.get_dataset_item_annotations(project_id=project_id, dataset_item_id=dataset_item_id)
+        return dataset_service.get_dataset_item_annotations(project=project, dataset_item_id=dataset_item_id)
     except (ResourceNotFoundError, NotAnnotatedError) as e:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e))
 
@@ -275,13 +275,13 @@ def get_dataset_item_annotations(
     },
 )
 def delete_dataset_item_annotation(
-    project_id: Annotated[UUID, Depends(get_project_id)],
+    project: Annotated[ProjectView, Depends(get_project)],
     dataset_item_id: Annotated[UUID, Depends(get_dataset_item_id)],
     dataset_service: Annotated[DatasetService, Depends(get_dataset_service)],
 ) -> None:
     """Delete dataset item annotations"""
     try:
-        dataset_service.delete_dataset_item_annotations(project_id=project_id, dataset_item_id=dataset_item_id)
+        dataset_service.delete_dataset_item_annotations(project=project, dataset_item_id=dataset_item_id)
     except ResourceNotFoundError as e:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e))
 
@@ -298,15 +298,15 @@ def delete_dataset_item_annotation(
     },
 )
 def assign_dataset_item_subset(
-    project_id: Annotated[UUID, Depends(get_project_id)],
+    project: Annotated[ProjectView, Depends(get_project)],
     dataset_item_id: Annotated[UUID, Depends(get_dataset_item_id)],
     dataset_service: Annotated[DatasetService, Depends(get_dataset_service)],
     subset_config: Annotated[DatasetItemAssignSubset, Body()],
 ) -> DatasetItem:
     """Assign dataset item subset"""
     try:
         return dataset_service.assign_dataset_item_subset(
-            project_id=project_id, dataset_item_id=dataset_item_id, subset=subset_config.subset
+            project=project, dataset_item_id=dataset_item_id, subset=subset_config.subset
         )
     except ResourceNotFoundError as e:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e))

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

@@ -12,10 +12,10 @@
 from fastapi.openapi.models import Example
 from pydantic import ValidationError
 
-from app.api.dependencies import get_pipeline_service, get_project_id
+from app.api.dependencies import get_pipeline_metrics_service, get_pipeline_service, get_project_id
 from app.schemas.metrics import PipelineMetrics
 from app.schemas.pipeline import DataCollectionPolicyAdapter, PipelineStatus, PipelineView
-from app.services import PipelineService, ResourceNotFoundError
+from app.services import PipelineMetricsService, PipelineService, ResourceNotFoundError
 
 logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/api/projects/{project_id}/pipeline", tags=["Pipelines"])
@@ -176,7 +176,7 @@ def disable_pipeline(
 )
 def get_project_metrics(
     project_id: Annotated[UUID, Depends(get_project_id)],
-    pipeline_service: Annotated[PipelineService, Depends(get_pipeline_service)],
+    pipeline_metrics_service: Annotated[PipelineMetricsService, Depends(get_pipeline_metrics_service)],
     time_window: int = 60,
 ) -> PipelineMetrics:
     """
@@ -191,7 +191,7 @@ def get_project_metrics(
         )
 
     try:
-        return pipeline_service.get_pipeline_metrics(project_id, time_window)
+        return pipeline_metrics_service.get_pipeline_metrics(project_id, time_window)
     except ResourceNotFoundError as e:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e))
     except ValueError as e:

application/backend/app/db/schema.py

@@ -34,7 +34,6 @@ class ProjectDB(BaseID):
     task_type: Mapped[str] = mapped_column(String(50), nullable=False)
     exclusive_labels: Mapped[bool] = mapped_column(Boolean, default=False)
 
-    pipeline = relationship("PipelineDB", back_populates="project", uselist=False)
     model_revisions = relationship("ModelRevisionDB", back_populates="project")
 
 
@@ -48,7 +47,6 @@ class PipelineDB(Base):
     is_running: Mapped[bool] = mapped_column(Boolean, default=False)
     data_collection_policies: Mapped[list] = mapped_column(JSON, nullable=False, default=list)
 
-    project = relationship("ProjectDB", back_populates="pipeline")
     sink = relationship("SinkDB", uselist=False)
     source = relationship("SourceDB", uselist=False)
     model_revision = relationship("ModelRevisionDB", uselist=False)