diff --git a/backend/app/routers/dinsar_production.py b/backend/app/routers/dinsar_production.py index 4e4ae6e..f145dcc 100644 --- a/backend/app/routers/dinsar_production.py +++ b/backend/app/routers/dinsar_production.py @@ -8,8 +8,8 @@ from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel, Field from .dependencies import _get_current_user, _require_admin +from .. import database from ..config import read_int_env, settings -from ..database import AsyncSessionLocal from ..models import AuthUserORM from ..services.dinsar_production_service import dinsar_production_service from ..services.job_queue_service import job_queue_service @@ -51,6 +51,14 @@ def _get_registry(): return registry +def _new_session(): + if database.AsyncSessionLocal is None: + database.init_db() + if database.AsyncSessionLocal is None: + raise RuntimeError("Database session factory is not initialized.") + return database.AsyncSessionLocal() + + @router.get("/engines") async def list_engines(): registry = _get_registry() @@ -184,9 +192,7 @@ async def submit_run( try: if req.engine_code == "sarscape": - if AsyncSessionLocal is None: - raise RuntimeError("Database session factory is not initialized.") - async with AsyncSessionLocal() as db: + async with _new_session() as db: result = await dinsar_production_service.create_run( engine_code=req.engine_code, profile_code=req.profile, @@ -239,8 +245,6 @@ async def submit_run( @router.get("/runs") async def list_runs(limit: int = 20): - if AsyncSessionLocal is None: - raise HTTPException(status_code=500, detail="Database session factory is not initialized.") - async with AsyncSessionLocal() as db: + async with _new_session() as db: result = await dinsar_production_service.list_runs(db, limit=limit) return result diff --git a/backend/app/routers/tasks.py b/backend/app/routers/tasks.py index 6ae1196..a2f596e 100644 --- a/backend/app/routers/tasks.py +++ b/backend/app/routers/tasks.py @@ -8,6 +8,7 @@ from fastapi import APIRouter, Depends, HTTPException, Request from fastapi.responses import StreamingResponse from pydantic import BaseModel +from .. import database from ..auth_utils import verify_password from ..models import AuthUserORM, TaskInfo from ..services.dinsar_production_service import dinsar_production_service @@ -20,7 +21,6 @@ from ..services.task_service import ( task_service, ) from ..auth_service import SESSION_COOKIE_NAME, get_user_by_session_token -from ..database import AsyncSessionLocal from .dependencies import _require_admin router = APIRouter() @@ -30,6 +30,14 @@ class ForceCancelRequest(BaseModel): password: str +def _new_session(): + if database.AsyncSessionLocal is None: + database.init_db() + if database.AsyncSessionLocal is None: + raise RuntimeError("Database session factory is not initialized.") + return database.AsyncSessionLocal() + + @router.get("/tasks/active", response_model=List[TaskInfo]) async def get_active_tasks(limit: int = TASK_ACTIVE_DEFAULT_LIMIT, offset: int = 0): """获取当前正在运行的所有后台任务""" @@ -44,12 +52,12 @@ async def stream_active_tasks(request: Request): """SSE 端点:每 3s 推送一次活跃任务列表,替代前端轮询。""" # Authenticate via Cookie at connection establishment token = request.cookies.get(SESSION_COOKIE_NAME) - if token and AsyncSessionLocal: - async with AsyncSessionLocal() as db: + if token: + async with _new_session() as db: user = await get_user_by_session_token(db, token) if not user: raise HTTPException(status_code=401, detail="Authentication required.") - elif not token: + else: raise HTTPException(status_code=401, detail="Authentication required.") async def event_generator(): @@ -127,14 +135,13 @@ async def force_cancel_task( if task.status not in ("PENDING", "RUNNING"): raise HTTPException(status_code=400, detail="任务已结束,无需取消") killed_pid = None - if AsyncSessionLocal is not None: - async with AsyncSessionLocal() as db: - run = await dinsar_production_service.request_cancel(task_id, db=db) - if run is not None: - killed_pid = await dinsar_production_service.kill_active_execution_by_task_id( - task_id, - db=db, - ) + async with _new_session() as db: + run = await dinsar_production_service.request_cancel(task_id, db=db) + if run is not None: + killed_pid = await dinsar_production_service.kill_active_execution_by_task_id( + task_id, + db=db, + ) await task_service.update_task(task_id, status="CANCELLED", message="管理员强制取消") return { "message": "任务已强制取消", diff --git a/backend/app/services/job_handlers.py b/backend/app/services/job_handlers.py index b2e024d..4832b58 100644 --- a/backend/app/services/job_handlers.py +++ b/backend/app/services/job_handlers.py @@ -17,8 +17,8 @@ logger = logging.getLogger(__name__) from sqlalchemy import select +from .. import database from ..config import settings -from ..database import AsyncSessionLocal from ..models import SystemJobORM, DinsarResultORM, HazardPointORM, DinsarTaskItemORM, PsTaskItemORM, SARSceneGeoORM, FloodDetectionORM, WaterDetectionORM, GF3ProcessingORM, AiDiagnosisORM from ..scheduler import scan_data_job from .data_service import data_service @@ -80,6 +80,14 @@ JOB_TYPE_REBUILD_PSINSAR_CATALOG = "REBUILD_PSINSAR_CATALOG" COPY_ALLOWED_STATUSES = {"PENDING", "IN_PROGRESS", "COMPLETED", "FAILED"} +def AsyncSessionLocal(): + if database.AsyncSessionLocal is None: + database.init_db() + if database.AsyncSessionLocal is None: + raise RuntimeError("Database session factory is not initialized.") + return database.AsyncSessionLocal() + + def _normalize_copy_statuses(raw_statuses: Any) -> List[str]: if not raw_statuses: return ["COMPLETED"]