[refactor] move all registry to a container class to avoid import before dataflow init issue

Author: SunnyHaze

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

@@ -1,54 +1,51 @@
 import os
 from fastapi import APIRouter, HTTPException
 from app.schemas.dataset import DatasetIn, DatasetOut
-from app.services.dataset_registry import DatasetRegistry, _DATASET_REGISTRY
+from app.core.container import container
 from app.api.v1.resp import ok, created
 from app.api.v1.envelope import ApiResponse
 from app.api.v1.errors import *
 
 
 router = APIRouter(tags=["datasets"])
-_registry = _DATASET_REGISTRY
 
 @router.get("/", response_model=ApiResponse[list[DatasetOut]], operation_id="list_datasets", summary="返回目前所有注册的数据集列表，包含每个数据集的条目数和文件大小")
 def list_datasets():
     """返回所有数据集列表，每个数据集包含条目数(num_samples)和文件大小(file_size)信息"""
-    return ok(_registry.list())
+    return ok(container.dataset_registry.list())
 
 @router.post("/", response_model=ApiResponse[DatasetOut], operation_id="register_dataset", summary="注册一个新的数据集或更新已有数据集的信息，根据路径作为唯一主键")
 def register_dataset(payload: DatasetIn):
     try:
-        ds = _registry.add_or_update(payload.model_dump(mode="json")) # to dict
+        ds = container.dataset_registry.add_or_update(payload.model_dump(mode="json")) # to dict
     except Exception as e:
         raise HTTPException(400, f"Failed to register dataset: {e}")
     return created(ds)
 
 @router.get("/{ds_id}", response_model=ApiResponse[DatasetOut], operation_id="get_dataset", summary="根据数据集 ID 获取数据集信息")
 def get_dataset(ds_id: str):
-    ds = _registry.get(ds_id)
+    ds = container.dataset_registry.get(ds_id)
     if not ds:
         raise HTTPException(404, "Dataset not found")
     return ok(ds)
 
 @router.delete("/{ds_id}", response_model=ApiResponse[dict], operation_id="delete_dataset", summary="根据数据集 ID 删除数据集")
 def delete_dataset(ds_id: str):
-    ds = _registry.get(ds_id)
+    ds = container.dataset_registry.get(ds_id)
     if not ds:
         raise HTTPException(404, "Dataset not found")
-    _registry.remove(ds_id)
+    container.dataset_registry.remove(ds_id)
     return ok(message="Dataset deleted")
 
 
 # getting sample data for visualization
-from app.services.visualize_dataset import VisualizeDatasetService
-_visualize_service = VisualizeDatasetService()
 @router.get("/pandas_type_sample/{ds_id}", response_model=ApiResponse[str], operation_id="get_pandas_data", summary="获取指定数据集的 Pandas 类型样本数据,用于前端展示预览，可以通过start和end参数控制获取多少数据")
 def get_pandas_data(ds_id: str, start: int = 0, end: int = 5):
     try:
-        ds = _registry.get(ds_id)
+        ds = container.dataset_registry.get(ds_id)
         if not ds:
             raise HTTPException(404, "Dataset not found")
-        return ok(_visualize_service.get_pandas_read_function(ds, start, end))
+        return ok(container.dataset_visualize_service.get_pandas_read_function(ds, start, end))
     except Exception as e:
         raise HTTPException(500, f"Failed to get pandas data: {e}")
 
@@ -57,10 +54,10 @@ def get_pandas_data(ds_id: str, start: int = 0, end: int = 5):
 @router.get("/file_type_sample/{ds_id}", operation_id="get_file_type_data", summary="获取指定数据集的文件类型样本数据，用于前端展示下载，可以是图片、文本等")
 def get_file_type_data(ds_id: str):
     try:
-        ds = _registry.get(ds_id)
+        ds = container.dataset_registry.get(ds_id)
         if not ds:
             raise HTTPException(404, "Dataset not found")
-        file_path, media_type = _visualize_service.get_other_visualization_data(ds)
+        file_path, media_type = container.dataset_visualize_service.get_other_visualization_data(ds)
     except Exception as e:
         raise HTTPException(500, f"Failed to get file type data: {e}")
 
@@ -82,7 +79,7 @@ def get_dataset_preview(ds_id: str, num_lines: int = 5):
         预览内容的列表，每个元素是一个字典
     """
     try:
-        preview_data = _registry.preview(ds_id, num_lines)
+        preview_data = container.dataset_registry.preview(ds_id, num_lines)
         return ok(preview_data)
     except FileNotFoundError:
         raise HTTPException(404, "Dataset not found")
@@ -100,7 +97,7 @@ def get_dataset_columns(ds_id: str):
         列名列表，如果不支持则返回空列表
     """
     try:
-        columns_data = _registry.get_columns(ds_id)
+        columns_data = container.dataset_registry.get_columns(ds_id)
         return ok(columns_data)
     except FileNotFoundError:
         raise HTTPException(404, "Dataset not found")

backend/app/api/v1/endpoints/operators.py

@@ -15,7 +15,8 @@
 from app.api.v1.envelope import ApiResponse
 
 # --- 2. 导入服务层 ---
-from app.services.operator_registry import _op_registry, OPS_JSON_PATH
+from app.services.operator_registry import OPS_JSON_PATH
+from app.core.container import container
 
 router = APIRouter(tags=["operators"])
 
@@ -28,7 +29,7 @@
 def list_operators():
     """返回所有注册的算子列表（简化版）。"""
     try:
-        op_list = _op_registry.get_op_list()
+        op_list = container.operator_registry.get_op_list()
         return ok(op_list)
     except Exception as e:
         log.error(f"获取算子列表失败: {e}")
@@ -49,7 +50,7 @@ def list_operators_details():
     try:
         if not OPS_JSON_PATH.exists():
             log.info("ops.json 缓存文件未找到，自动触发一次算子扫描并生成缓存...")
-            ops_data = _op_registry.dump_ops_to_json()
+            ops_data = container.operator_registry.dump_ops_to_json()
         else:
             with open(OPS_JSON_PATH, "r", encoding="utf-8") as f:
                 ops_data = json.load(f)
@@ -80,7 +81,7 @@ def get_operator_detail_by_name(op_name: str):
         # 确保缓存存在
         if not OPS_JSON_PATH.exists():
             log.info("ops.json 缓存文件未找到，自动触发一次算子扫描并生成缓存...")
-            ops_data = _op_registry.dump_ops_to_json()
+            ops_data = container.operator_registry.dump_ops_to_json()
         else:
             with open(OPS_JSON_PATH, "r", encoding="utf-8") as f:
                 ops_data = json.load(f)

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

@@ -6,7 +6,7 @@
     PipelineExecutionRequest, 
     PipelineExecutionResult
 )
-from app.services.pipeline_registry import PipelineRegistry, _PIPELINE_REGISTRY
+from app.core.container import container
 from app.services.dataflow_engine import dataflow_engine
 from app.api.v1.resp import ok, created
 from app.api.v1.envelope import ApiResponse
@@ -23,7 +23,7 @@
 def list_pipelines(request: Request):
     try:
         logger.info(f"Request: {request.method} {request.url.path}")
-        pipelines = _PIPELINE_REGISTRY.list_pipelines()
+        pipelines = container.pipeline_registry.list_pipelines()
         logger.info(f"Successfully listed {len(pipelines)} pipelines")
         return ok(pipelines)
     except Exception as e:
@@ -38,9 +38,9 @@ def create_pipeline(request: Request, payload: PipelineIn):
 
         operators = pipeline_in_data.get("config", {}).get("operators", [])
         for op in operators:
-            op["params"] = _PIPELINE_REGISTRY.parse_frontend_params(op.get("params", []))
+            op["params"] = container.pipeline_registry.parse_frontend_params(op.get("params", []))
 
-        pipeline = _PIPELINE_REGISTRY.create_pipeline(pipeline_in_data)
+        pipeline = container.pipeline_registry.create_pipeline(pipeline_in_data)
         return created(pipeline)
     except ValueError as e:
         logger.error(f"Invalid pipeline configuration: {str(e)}", exc_info=True)
@@ -51,7 +51,7 @@ def create_pipeline(request: Request, payload: PipelineIn):
 
 @router.get("/{pipeline_id}", response_model=ApiResponse[PipelineOut], operation_id="get_pipeline", summary="根据ID获取Pipeline详情")
 def get_pipeline(pipeline_id: str):
-    pipeline = _PIPELINE_REGISTRY.get_pipeline(pipeline_id)
+    pipeline = container.pipeline_registry.get_pipeline(pipeline_id)
     if not pipeline:
         raise HTTPException(404, f"Pipeline with id {pipeline_id} not found")
     return ok(pipeline)
@@ -63,9 +63,9 @@ def update_pipeline(pipeline_id: str, payload: PipelineIn):
 
         # operators = pipeline_in_data.get("config", {}).get("operators", [])
         # for op in operators:
-        #     op["params"] = _PIPELINE_REGISTRY.parse_frontend_params(op.get("params", []))
+        #     op["params"] = container.pipeline_registry.parse_frontend_params(op.get("params", []))
 
-        updated_pipeline = _PIPELINE_REGISTRY.update_pipeline(pipeline_id, pipeline_in_data)
+        updated_pipeline = container.pipeline_registry.update_pipeline(pipeline_id, pipeline_in_data)
         return ok(updated_pipeline)
     except ValueError as e:
         logger.error(f"Failed to update pipeline: {str(e)}")
@@ -77,7 +77,7 @@ def update_pipeline(pipeline_id: str, payload: PipelineIn):
 @router.delete("/{pipeline_id}", response_model=ApiResponse[Dict], operation_id="delete_pipeline", summary="删除指定的Pipeline") 
 def delete_pipeline(pipeline_id: str):
     try:
-        success = _PIPELINE_REGISTRY.delete_pipeline(pipeline_id)
+        success = container.pipeline_registry.delete_pipeline(pipeline_id)
         if not success:
             raise HTTPException(404, f"Pipeline with id {pipeline_id} not found")
         return ok(message=f"Pipeline {pipeline_id} deleted successfully")
@@ -93,10 +93,10 @@ async def execute_pipeline(request: Request, pipeline_id):
     try:
         logger.info(f"Request: {request.method} {request.url.path}")
 
-        pipeline_config = _PIPELINE_REGISTRY.get_pipeline(pipeline_id)
+        pipeline_config = container.pipeline_registry.get_pipeline(pipeline_id)
 
         # 调用服务层开始执行
-        execution_id, pipeline_config, initial_result = _PIPELINE_REGISTRY.start_execution(
+        execution_id, pipeline_config, initial_result = container.pipeline_registry.start_execution(
             pipeline_id=pipeline_id, 
             config=pipeline_config
         )
@@ -111,15 +111,15 @@ async def execute_pipeline(request: Request, pipeline_id):
 
 @router.get("/execution/{execution_id}", response_model=ApiResponse[PipelineExecutionResult], operation_id="get_execution_result", summary="获取Pipeline执行结果")
 def get_execution_result(execution_id: str):
-    result = _PIPELINE_REGISTRY.get_execution_result(execution_id)
+    result = container.pipeline_registry.get_execution_result(execution_id)
     if not result:
         raise HTTPException(404, f"Execution with id {execution_id} not found")
     return ok(result)
 
 @router.get("/executions", response_model=ApiResponse[List[PipelineExecutionResult]], operation_id="list_executions", summary="列出所有Pipeline执行记录")
 def list_executions():
     try:
-        executions = _PIPELINE_REGISTRY.list_executions()
+        executions = container.pipeline_registry.list_executions()
         return ok(executions)
     except Exception as e:
         logger.error(f"Failed to list executions: {e}")

backend/app/api/v1/endpoints/prompts.py

@@ -1,7 +1,8 @@
 
 from fastapi import APIRouter, HTTPException
 from app.schemas.prompt import GetPromptSchema, PromptSourceOut, OperatorPromptMapOut, PromptInfoMapOut
-from app.services.prompt_registry import _PROMPT_REGISTRY
+# from app.services.prompt_registry import _PROMPT_REGISTRY
+from app.core.container import container
 from app.api.v1.resp import ok
 from app.api.v1.envelope import ApiResponse
 
@@ -13,7 +14,7 @@
     summary="查看所有算子及其对应的 Prompt 列表"
 )
 def get_operator_prompt_mapping():
-    result = _PROMPT_REGISTRY.list_operator_prompts()
+    result = container.prompt_registry.list_operator_prompts()
     return ok(result)
 
 @router.get(
@@ -22,15 +23,15 @@ def get_operator_prompt_mapping():
     summary="查看所有 prompt 的信息（operator, class string, category）"
 )
 def get_prompt_info():
-    return ok(_PROMPT_REGISTRY.list_prompt_info())
+    return ok(container.prompt_registry.list_prompt_info())
 
 @router.get(
     "/{operator_name}",
     response_model=ApiResponse[GetPromptSchema],
     summary="根据算子名称获取对应的 Prompt 列表"
 )
 def get_prompts(operator_name: str):
-    result = _PROMPT_REGISTRY.get_prompts(operator_name)
+    result = container.prompt_registry.get_prompts(operator_name)
     if not result:
         raise HTTPException(404, "Operator not found")
     return ok(result)
@@ -41,7 +42,7 @@ def get_prompts(operator_name: str):
     summary="根据 Prompt 名称返回 Prompt 类的源码"
 )
 def get_prompt_source(prompt_name: str):
-    result = _PROMPT_REGISTRY.get_prompt_source(prompt_name)
+    result = container.prompt_registry.get_prompt_source(prompt_name)
     if not result:
         raise HTTPException(status_code=404, detail="Prompt not found")
 

backend/app/api/v1/endpoints/serving.py

@@ -2,7 +2,7 @@
 import copy
 from typing import List, Dict, Any 
 from fastapi import APIRouter, HTTPException
-from app.services.prompt_registry import _PROMPT_REGISTRY
+from app.core.container import container
 from app.api.v1.resp import ok
 from app.api.v1.envelope import ApiResponse
 
@@ -15,7 +15,7 @@
     ServingTestSchema,
     ServingUpdateSchema
 )
-from app.services.serving_registry import _SERVING_REGISTRY, SERVING_CLS_REGISTRY
+from app.services.serving_registry import SERVING_CLS_REGISTRY
 
 router = APIRouter(tags=["serving"])
 
@@ -25,7 +25,7 @@
 )
 def list_serving_instances():
     try:
-        serving_list = _SERVING_REGISTRY._get_all()
+        serving_list = container.serving_registry._get_all()
         if not serving_list:
             result = []
         else:
@@ -52,7 +52,7 @@ def list_serving_classes():
     返回所有注册的 Serving 类及其初始化参数信息 (名称、类型、默认值)。
     """
     try:
-        classes_info = copy.deepcopy(_SERVING_REGISTRY.get_serving_classes())
+        classes_info = copy.deepcopy(container.serving_registry.get_serving_classes())
         api_llm_info = [x for x in classes_info if x['cls_name'] == 'APILLMServing_request']
         for item in api_llm_info:
             item['params'] = [p for p in item['params'] if p['name'] != 'key_name_of_api_key']
@@ -78,7 +78,7 @@ def get_serving_detail(id: str):
     根据 Serving 实例的 ID，获取其详细信息。
     """
     try:
-        serving_data = _SERVING_REGISTRY._get(id)
+        serving_data = container.serving_registry._get(id)
         print(type(serving_data))
         if not serving_data:
             raise HTTPException(status_code=404, detail=f"Serving instance with id {id} not found")
@@ -118,7 +118,7 @@ def update_serving_instance(id: str, body: ServingUpdateSchema):
 
                 params_list.append(p_dict)
 
-        success = _SERVING_REGISTRY._update(
+        success = container.serving_registry._update(
             id, 
             name=body.name, 
             params=params_list
@@ -142,7 +142,7 @@ def delete_serving_instance(id: str):
     删除指定的 Serving 实例。
     """
     try:
-        success = _SERVING_REGISTRY._delete(id)
+        success = container.serving_registry._delete(id)
         if not success:
             raise HTTPException(status_code=404, detail=f"Serving instance with id {id} not found")
         return ok({'id': id})
@@ -166,7 +166,7 @@ def create_serving_instance(
     try:
         # Get class default params info
         cls_info = None
-        all_classes = _SERVING_REGISTRY.get_serving_classes()
+        all_classes = container.serving_registry.get_serving_classes()
         for c in all_classes:
             if c['cls_name'] == cls_name:
                 cls_info = c
@@ -197,7 +197,7 @@ def create_serving_instance(
 
         new_params = list(final_params_map.values())
 
-        new_id = _SERVING_REGISTRY._set(name, cls_name, new_params)
+        new_id = container.serving_registry._set(name, cls_name, new_params)
         return ok({
             'id': new_id
         })
@@ -217,7 +217,7 @@ def test_serving_instance(id: str, body: ServingTestSchema):
     """
     try:
         prompt: str = body.prompt or "Hello, which model are you?"
-        serving_info = _SERVING_REGISTRY._get(id)
+        serving_info = container.serving_registry._get(id)
         params_dict = {}
 
         ## This part of code is only for APILLMServing_request