feat(api_gateway): 集成 csst-dag 并重构任务提交界面 (4dc4d8a8) · Commits · csst-cicd / csst-pipeline-portal

.gitignore

+1 −1

Original line number	Diff line number	Diff line
		@@ -63,4 +63,4 @@ data/
		# Generated during Ansible variable import
		docker-celery-3.0.5/api_gateway/ccds/
		docker-celery-3.0.5/api_gateway/csst-dfs-client/
		external/csst-dag
		#external/csst-dag

docker-celery-3.0.5/Makefile

+5 −1

Original line number	Diff line number	Diff line
		@@ -78,7 +78,7 @@ build:
		if [ -d "../external/csst-dags" ]; then echo " [OK] csst-dags"; else echo " [MISS] csst-dags"; fi;
		@if [ ! -f $(REQ_FILE) ]; then echo "Error: $(REQ_FILE) not found"; exit 1; fi
		@set -e; \
		cleanup() { rm -rf ./api_gateway/csst-dfs-client ./api_gateway/ccds ./external/js9 >/dev/null 2>&1 \|\| true; }; \
		cleanup() { rm -rf ./api_gateway/csst-dfs-client ./api_gateway/ccds ./api_gateway/csst-dag ./external/js9 >/dev/null 2>&1 \|\| true; }; \
		trap cleanup EXIT; \
		total=$$(grep -v '^#' $(REQ_FILE) \| grep -v '^$$' \| wc -l); \
		total_imgs=$$(($$total+$(CUSTOM_IMAGE_COUNT))); \
		@@ -110,6 +110,10 @@ build:
		mkdir -p ./api_gateway/ccds; \
		rsync -a --delete --exclude ".git" --exclude "__pycache__" ../external/ccds/ ./api_gateway/ccds/; \
		fi; \
		if [ -d "../external/csst-dag" ]; then \
		mkdir -p ./api_gateway/csst-dag; \
		rsync -a --delete --exclude ".git" --exclude "__pycache__" ../external/csst-dag/ ./api_gateway/csst-dag/; \
		fi; \
		docker build $(DOCKER_BUILD_FLAGS) $(API_GATEWAY_BUILD_ARGS) -t $(TARGET_API_GATEWAY) ./api_gateway; \
		echo "[$$(($$total+3))/$${total_imgs}] Building Task Portal Image ($(TARGET_TASK_PORTAL)) with BUILD_NUMBER=$(BUILD_NUMBER)..."; \
		docker build $(DOCKER_BUILD_FLAGS) --build-arg VITE_BUILD_NUMBER=$(BUILD_NUMBER) --build-arg VITE_BUILD_TIME="$(shell date '+%Y-%m-%d %H:%M:%S')" -t $(TARGET_TASK_PORTAL) ./frontend_vue; \

docker-celery-3.0.5/ansible/update.yml

+1 −1

Original line number	Diff line number	Diff line
		@@ -199,7 +199,7 @@
		- name: Deactivate legacy __low DAG ids (keep history, hide from UI)
		shell: \|
		docker exec csst-airflow-postgres-1 psql -U airflow -d airflow -v ON_ERROR_STOP=1 \
		-c "UPDATE dag SET is_active = FALSE WHERE dag_id LIKE '%__low';"
		-c "UPDATE dag SET is_paused = TRUE WHERE dag_id LIKE '%__low';"
		args:
		chdir: "{{ _deploy_dir }}"
		ignore_errors: yes

docker-celery-3.0.5/api_gateway/Dockerfile

+8 −0

Original line number	Diff line number	Diff line
		@@ -10,6 +10,7 @@ ARG PIP_RETRIES=5

		COPY csst-dfs-client /app/csst-dfs-client/
		COPY ccds /app/ccds/
		COPY csst-dag /app/csst-dag/
		RUN set -e; \
		pip_args="--no-cache-dir --default-timeout=${PIP_DEFAULT_TIMEOUT} --retries=${PIP_RETRIES}"; \
		if [ -n "${PIP_INDEX_URL}" ]; then pip_args="$pip_args -i ${PIP_INDEX_URL}"; fi; \
		@@ -24,6 +25,13 @@ RUN set -e; \
		if [ -n "${PIP_TRUSTED_HOST}" ]; then for host in ${PIP_TRUSTED_HOST}; do pip_args="$pip_args --trusted-host ${host}"; done; fi; \
		if [ -f "/app/ccds/setup.py" ] \|\| [ -f "/app/ccds/pyproject.toml" ]; then pip install $pip_args /app/ccds/; fi

		RUN set -e; \
		pip_args="--no-cache-dir --default-timeout=${PIP_DEFAULT_TIMEOUT} --retries=${PIP_RETRIES}"; \
		if [ -n "${PIP_INDEX_URL}" ]; then pip_args="$pip_args -i ${PIP_INDEX_URL}"; fi; \
		if [ -n "${PIP_EXTRA_INDEX_URL}" ]; then for url in ${PIP_EXTRA_INDEX_URL}; do pip_args="$pip_args --extra-index-url ${url}"; done; fi; \
		if [ -n "${PIP_TRUSTED_HOST}" ]; then for host in ${PIP_TRUSTED_HOST}; do pip_args="$pip_args --trusted-host ${host}"; done; fi; \
		if [ -f "/app/csst-dag/setup.py" ] \|\| [ -f "/app/csst-dag/pyproject.toml" ]; then pip install $pip_args --no-deps /app/csst-dag/; fi

		COPY . .

		EXPOSE 38000

docker-celery-3.0.5/api_gateway/main.py

+73 −5

Original line number	Diff line number	Diff line
		@@ -279,6 +279,7 @@ class BatchSubmitRequest(BaseModel):
		dag_group_run: Optional[Dict[str, Any]] = None
		dag_run_list: Optional[List[Dict[str, Any]]] = None
		# v2 fields
		group: Optional[Dict[str, Any]] = None
		job: Optional[Dict[str, Any]] = None
		dag_runs: Optional[List[Dict[str, Any]]] = None

		@@ -299,7 +300,7 @@ def batch_submit_tasks(
		seen_task_ids = set()

		# 兼容 v1 和 v2 载荷
		group_info = dict(payload.job or payload.dag_group_run or {})
		group_info = dict(payload.group or payload.job or payload.dag_group_run or {})
		tasks = payload.dag_runs or payload.dag_run_list or []

		group_run_id = str(group_info.get("dag_run_group") or group_info.get("dag_group_run") or "").strip()
		@@ -311,9 +312,9 @@ def batch_submit_tasks(
		# Fallback to existing fields if new fields are used
		job_info = task.get("job", {}) if isinstance(task.get("job"), dict) else {}

		requested_dag_id = job_info.get("dag_id") or task.get("dag_id") or task.get("dag")
		requested_dag_id = job_info.get("dag") or task.get("dag_id") or task.get("dag")
		dag_run_id = job_info.get("dag_run") or task.get("dag_run_id") or task.get("dag_run")
		requested_priority = task.get("priority")
		requested_priority = job_info.get("priority") or task.get("priority") or group_info.get("priority")

		if not requested_dag_id or not dag_run_id:
		raise HTTPException(status_code=400, detail="Each task must contain 'dag_id' and 'dag_run' (or v1 equivalent)")
		@@ -436,7 +437,74 @@ def get_task_metadata_fields(db: Session = Depends(get_db), current_user: User =
		return {"fields": result}


		@app.get("/api/tasks/{task_id}")
		@app.get("/api/utils/uuid7")
		def generate_uuid7_endpoint(count: int = 1):
		"""
		Generate UUIDv7s via the core csst_dag package to ensure high concurrency monotonic safety.
		"""
		import sys
		from pathlib import Path

		safe_count = min(max(1, int(count)), 10000)

		try:
		from csst_dag.v2.utils import generate_uuid7
		return {"uuids": [generate_uuid7() for _ in range(safe_count)]}
		except Exception:
		try:
		project_root = Path(__file__).resolve().parents[2]
		csst_dag_root = project_root / "external" / "csst-dag"
		if csst_dag_root.exists():
		sys.path.insert(0, str(csst_dag_root))
		from csst_dag.v2.utils import generate_uuid7
		return {"uuids": [generate_uuid7() for _ in range(safe_count)]}
		except Exception as e:
		logger.error(f"uuid7 生成失败，退回 uuid4: {e}")
		return {"uuids": [str(uuid.uuid4()) for _ in range(safe_count)]}


		class DagRunGroupTemplateRequest(BaseModel):
		dag: str
		batch_id: str = "default"
		priority: str = "low"
		data: Dict[str, Any] = Field(default_factory=dict)
		proc: Dict[str, Any] = Field(default_factory=dict)


		@app.post("/api/utils/dagrun-group/template")
		def generate_dagrun_group_template(payload: DagRunGroupTemplateRequest = Body(...)):
		"""根据 CSST-DAG v2 Models 生成 DagRunGroup 预览。

		Parameters
		----------
		payload : DagRunGroupTemplateRequest
		前端表单输入。包含 `dag`, `batch_id`, `priority`, 以及 `data`/`proc` 两组参数。

		Returns
		-------
		dict
		结构为 `{group: {...}, dag_runs: [...]}`。
		"""
		from pathlib import Path
		import sys

		try:
		from csst_dag.v2.models import DagRunGroup
		except Exception:
		project_root = Path(__file__).resolve().parents[2]
		csst_dag_root = project_root / "external" / "csst-dag"
		if csst_dag_root.exists():
		sys.path.insert(0, str(csst_dag_root))
		from csst_dag.v2.models import DagRunGroup

		group = DagRunGroup.trigger(
		dag=payload.dag,
		batch_id=payload.batch_id,
		priority=payload.priority,
		data=dict(payload.data or {}),
		proc=dict(payload.proc or {}),
		)
		return group.to_payload()
		def get_task_status(task_id: str, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
		"""Get single task status and logs."""
		task = db.query(TaskRecord).filter(TaskRecord.task_id == task_id).first()
		@@ -1217,7 +1285,7 @@ def get_dfs_catalog_search(


		@app.get("/api/ccds/pmaps")
		def get_ccds_pmaps(observatory: str = "csst", current_user: User = Depends(get_current_user)):
		def get_ccds_pmaps(observatory: str = "csst"):
		try:
		import os
		import requests