fix(core): centralize async/sync bridge for APScheduler scheduled jobs
Create AsyncJobRunner — centralized bridge between APScheduler (BackgroundScheduler, sync thread pool) and async coroutines. Fixes: - P0: execute_run() called without await from APScheduler thread, causing coroutine to be silently discarded (root cause: no translation history) - P0: get_async_job_runner() deadlock when called from APScheduler thread pool without running event loop - P1: ID mismatch in disable_schedule/delete_schedule routes (job_id passed instead of schedule_id) - P1: asyncio.run() in APScheduler callbacks incompatible with running event loop - Delete unused llm_analysis/scheduler.py (not used in production) Changes: core/async_job_runner.py — new: AsyncJobRunner class core/scheduler.py — use runner.run()/run_later() translate/scheduler.py — use runner.run() for execute_run mapping_service.py — remove unused BackgroundScheduler dependencies.py — add get_async_job_runner() DI app.py — init runner in lifespan api/routes/migration.py — use runner.run() _schedule_routes.py — fix ID mismatch plugins/migration.py — use runner.run() llm_analysis/scheduler.py — delete (unused) tests: 151 new/updated tests, all passing
This commit is contained in:
@@ -32,7 +32,7 @@ from ...core.logger import belief_scope, logger
|
||||
from ...core.mapping_service import IdMappingService
|
||||
from ...core.migration.dry_run_orchestrator import MigrationDryRunService
|
||||
from ...core.async_superset_client import AsyncSupersetClient
|
||||
from ...dependencies import get_config_manager, get_task_manager, has_permission
|
||||
from ...dependencies import get_async_job_runner, get_config_manager, get_task_manager, has_permission
|
||||
from ...models.dashboard import DashboardMetadata, DashboardSelection
|
||||
from ...models.mapping import ResourceMapping
|
||||
|
||||
@@ -377,12 +377,13 @@ async def trigger_sync_now(
|
||||
db.commit()
|
||||
|
||||
service = IdMappingService(db)
|
||||
runner = get_async_job_runner()
|
||||
results = {"synced": [], "failed": []}
|
||||
|
||||
for env in environments:
|
||||
try:
|
||||
client = AsyncSupersetClient(env)
|
||||
service.sync_environment(env.id, client)
|
||||
runner.run(service.sync_environment(env.id, client))
|
||||
results["synced"].append(env.id)
|
||||
logger.reason(f"Synced environment {env.id}", extra={"src": "trigger_sync_now"})
|
||||
except Exception as e:
|
||||
|
||||
@@ -172,10 +172,10 @@ async def disable_schedule(
|
||||
logger.reason(f"disable_schedule — Job: {job_id}, User: {current_user.username}", extra={"src": "translate_routes"})
|
||||
try:
|
||||
scheduler = TranslationScheduler(db, config_manager, current_user.username)
|
||||
scheduler.set_schedule_active(job_id, is_active=False)
|
||||
schedule = scheduler.set_schedule_active(job_id, is_active=False)
|
||||
from ....dependencies import get_scheduler_service
|
||||
sched_svc = get_scheduler_service()
|
||||
sched_svc.remove_translation_job(schedule_id=job_id)
|
||||
sched_svc.remove_translation_job(schedule_id=schedule.id)
|
||||
return {"status": "disabled", "job_id": job_id}
|
||||
except ValueError as e:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e))
|
||||
@@ -199,10 +199,12 @@ async def delete_schedule(
|
||||
logger.reason(f"delete_schedule — Job: {job_id}, User: {current_user.username}", extra={"src": "translate_routes"})
|
||||
try:
|
||||
scheduler = TranslationScheduler(db, config_manager, current_user.username)
|
||||
schedule = scheduler.get_schedule(job_id)
|
||||
schedule_id = schedule.id
|
||||
scheduler.delete_schedule(job_id)
|
||||
from ....dependencies import get_scheduler_service
|
||||
sched_svc = get_scheduler_service()
|
||||
sched_svc.remove_translation_job(schedule_id=job_id)
|
||||
sched_svc.remove_translation_job(schedule_id=schedule_id)
|
||||
return None
|
||||
except ValueError as e:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e))
|
||||
|
||||
@@ -61,7 +61,7 @@ from .core.database import AuthSessionLocal, init_db
|
||||
from .core.encryption_key import ensure_encryption_key
|
||||
from .core.logger import belief_scope, logger
|
||||
from .core.utils.network import NetworkError
|
||||
from .dependencies import get_scheduler_service, get_task_manager
|
||||
from .dependencies import get_async_job_runner, get_scheduler_service, get_task_manager
|
||||
from .models.auth import Role, User
|
||||
|
||||
|
||||
@@ -108,6 +108,8 @@ async def lifespan(app: FastAPI):
|
||||
_s.close()
|
||||
except Exception as _e:
|
||||
logger.warning(f"Failed to clean up stuck validation runs: {_e}")
|
||||
logger.info("⏰ Initializing AsyncJobRunner...")
|
||||
get_async_job_runner() # Initialize singleton with running event loop BEFORE scheduler starts
|
||||
logger.info("⏰ Starting scheduler...")
|
||||
scheduler = get_scheduler_service()
|
||||
scheduler.start()
|
||||
|
||||
93
backend/src/core/async_job_runner.py
Normal file
93
backend/src/core/async_job_runner.py
Normal file
@@ -0,0 +1,93 @@
|
||||
# #region Core.AsyncJobRunner [C:4] [TYPE Module] [SEMANTICS async,scheduler,bridge,event-loop]
|
||||
# @defgroup Core Async/sync bridge for scheduled jobs.
|
||||
# @BRIEF Centralized bridge between APScheduler (sync thread pool) and async coroutines.
|
||||
# @LAYER Core
|
||||
# @PRE Event loop is running (FastAPI lifespan).
|
||||
# @POST All scheduled async jobs execute through this runner.
|
||||
# @SIDE_EFFECT Submits coroutines to the event loop from sync threads.
|
||||
# @RATIONALE APScheduler BackgroundScheduler runs callbacks in a sync thread pool.
|
||||
# Async coroutines must be dispatched to the FastAPI event loop via
|
||||
# asyncio.run_coroutine_threadsafe(). This runner centralizes that pattern.
|
||||
# @REJECTED asyncio.run() — creates a new event loop each call; breaks when called
|
||||
# from a context with an existing loop; incompatible with AsyncIOScheduler migration.
|
||||
# @REJECTED Direct async def callbacks in BackgroundScheduler — coroutine is created
|
||||
# but never awaited, causing silent no-ops (the root cause of scheduled translation failures).
|
||||
|
||||
import asyncio
|
||||
from typing import Any
|
||||
|
||||
|
||||
# #region AsyncJobRunner [C:4] [TYPE Class] [SEMANTICS async,runner,scheduler]
|
||||
# @ingroup Core
|
||||
# @BRIEF Centralized async/sync bridge for dispatching coroutines from sync threads.
|
||||
# @PRE Event loop is captured at construction time.
|
||||
# @POST run() executes coroutines synchronously (blocking until complete).
|
||||
# @POST run_later() schedules delayed coroutine execution.
|
||||
# @SIDE_EFFECT Submits coroutines to the captured event loop.
|
||||
# @DATA_CONTRACT Input[coroutine] -> Output[result]
|
||||
class AsyncJobRunner:
|
||||
"""Bridge between APScheduler sync threads and the FastAPI async event loop.
|
||||
|
||||
Usage:
|
||||
runner = AsyncJobRunner(asyncio.get_running_loop())
|
||||
result = runner.run(some_async_function())
|
||||
"""
|
||||
|
||||
# #region __init__ [TYPE Function]
|
||||
# @BRIEF Capture the event loop for later coroutine dispatch.
|
||||
# @PRE loop is a running asyncio event loop.
|
||||
# @POST self.loop is set and ready for run()/run_later() calls.
|
||||
def __init__(self, loop: asyncio.AbstractEventLoop) -> None:
|
||||
self.loop = loop
|
||||
# Track pending tasks from run_later() to prevent silent exception loss
|
||||
self._pending_tasks: set[asyncio.Task] = set()
|
||||
# #endregion __init__
|
||||
|
||||
# #region run [TYPE Function]
|
||||
# @BRIEF Execute a coroutine from a sync thread and wait for the result.
|
||||
# @PRE coro is an awaitable (coroutine or Future).
|
||||
# @POST Coroutine has been executed to completion; result or exception is returned.
|
||||
# @SIDE_EFFECT Blocks the calling thread until the coroutine completes.
|
||||
def run(self, coro: Any) -> Any:
|
||||
"""Dispatch a coroutine to the event loop and block until it completes.
|
||||
|
||||
Args:
|
||||
coro: An awaitable (coroutine, Future, or Task).
|
||||
|
||||
Returns:
|
||||
The result of the coroutine.
|
||||
|
||||
Raises:
|
||||
Exception: Any exception raised by the coroutine is re-raised.
|
||||
"""
|
||||
future = asyncio.run_coroutine_threadsafe(coro, self.loop)
|
||||
return future.result() # timeout=None — wait indefinitely
|
||||
# #endregion run
|
||||
|
||||
# #region run_later [TYPE Function]
|
||||
# @BRIEF Schedule a coroutine to execute after a delay (for windowed validation).
|
||||
# @PRE coro is an awaitable; delay_seconds >= 0.
|
||||
# @POST Coroutine is scheduled on the event loop's call_later queue.
|
||||
# @SIDE_EFFECT Non-blocking — returns immediately; coroutine executes later.
|
||||
def run_later(self, coro: Any, delay_seconds: float) -> None:
|
||||
"""Schedule a coroutine to execute after a delay.
|
||||
|
||||
Used for windowed validation: tasks are distributed across a time window
|
||||
using call_later() instead of running them all at once.
|
||||
|
||||
Args:
|
||||
coro: An awaitable to execute after the delay.
|
||||
delay_seconds: Seconds to wait before scheduling the coroutine.
|
||||
"""
|
||||
def _schedule() -> None:
|
||||
task = asyncio.ensure_future(coro, loop=self.loop)
|
||||
# Store task for exception tracking; remove when done
|
||||
self._pending_tasks.add(task)
|
||||
task.add_done_callback(self._pending_tasks.discard)
|
||||
# Use call_soon_threadsafe to schedule from any thread
|
||||
self.loop.call_soon_threadsafe(
|
||||
lambda: self.loop.call_later(delay_seconds, _schedule)
|
||||
)
|
||||
# #endregion run_later
|
||||
# #endregion AsyncJobRunner
|
||||
# #endregion Core.AsyncJobRunner
|
||||
@@ -1,21 +1,21 @@
|
||||
# #region IdMappingServiceModule [C:5] [TYPE Module] [SEMANTICS sqlalchemy, mapping, sync, schedule, superset, resource]
|
||||
# #region IdMappingServiceModule [C:5] [TYPE Module] [SEMANTICS sqlalchemy, mapping, sync, superset, resource]
|
||||
# @defgroup Core Module group.
|
||||
#
|
||||
# @BRIEF Service for tracking and synchronizing Superset Resource IDs (UUID <-> Integer ID)
|
||||
# @LAYER Core
|
||||
# @RELATION DEPENDS_ON -> [MappingModels]
|
||||
# @RELATION DEPENDS_ON -> [LoggerModule]
|
||||
# @RELATION DEPENDS_ON -> [AsyncJobRunner]
|
||||
# @PRE Database session is valid and Superset client factory returns authenticated clients for requested environments.
|
||||
# @POST Mapping synchronization and lookup APIs are available for environment-scoped UUID-to-integer resolution.
|
||||
# @SIDE_EFFECT Reads/writes ResourceMapping rows, emits logs, and schedules periodic sync jobs.
|
||||
# @SIDE_EFFECT Reads/writes ResourceMapping rows, emits logs.
|
||||
# @DATA_CONTRACT Input[environment_id, resource_type, uuid] -> Output[remote_integer_id|None]
|
||||
# @TEST_DATA: mock_superset_resources -> {'chart': [{'id': 42, 'uuid': '1234', 'slice_name': 'test'}], 'dataset': [{'id': 99, 'uuid': '5678', 'table_name': 'data'}]}
|
||||
#
|
||||
# @INVARIANT sync_environment must handle remote API failures gracefully.
|
||||
# @REJECTED BackgroundScheduler — was never started; replaced by AsyncJobRunner for async/sync bridge.
|
||||
from datetime import UTC, datetime
|
||||
|
||||
from apscheduler.schedulers.background import BackgroundScheduler
|
||||
from apscheduler.triggers.cron import CronTrigger
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from src.core.cot_logger import seed_trace_id
|
||||
@@ -27,11 +27,12 @@ from src.models.mapping import ResourceMapping, ResourceType
|
||||
# @defgroup Core Module group.
|
||||
# @BRIEF Service handling the cataloging and retrieval of remote Superset Integer IDs.
|
||||
# @PRE db_session is an active SQLAlchemy Session bound to mapping tables.
|
||||
# @POST Service instance provides scheduler control and environment-scoped mapping synchronization APIs.
|
||||
# @POST Service instance provides environment-scoped mapping synchronization APIs.
|
||||
# @RELATION DEPENDS_ON -> [MappingModels]
|
||||
# @RELATION DEPENDS_ON -> [LoggerModule]
|
||||
# @RELATION DEPENDS_ON -> [AsyncJobRunner]
|
||||
# @INVARIANT self.db remains the authoritative session for all mapping operations.
|
||||
# @SIDE_EFFECT Instantiates an in-process scheduler and performs database writes during sync cycles.
|
||||
# @SIDE_EFFECT Performs database writes during sync cycles.
|
||||
# @DATA_CONTRACT Input[db_session] -> Output[IdMappingService]
|
||||
#
|
||||
# @TEST_CONTRACT IdMappingServiceModel ->
|
||||
@@ -53,47 +54,7 @@ class IdMappingService:
|
||||
# @PURPOSE: Initializes the mapping service.
|
||||
def __init__(self, db_session: Session):
|
||||
self.db = db_session
|
||||
self.scheduler = BackgroundScheduler()
|
||||
self._sync_job = None
|
||||
# #endregion __init__
|
||||
# #region start_scheduler [TYPE Function]
|
||||
# @ingroup Core
|
||||
# @PURPOSE: Starts the background scheduler with a given cron string.
|
||||
# @PARAM cron_string (str) - Cron expression for the sync interval.
|
||||
# @PARAM environments (List[str]) - List of environment IDs to sync.
|
||||
# @PARAM superset_client_factory - Function to get a client for an environment.
|
||||
def start_scheduler(
|
||||
self, cron_string: str, environments: list[str], superset_client_factory
|
||||
):
|
||||
with belief_scope("IdMappingService.start_scheduler"):
|
||||
if self._sync_job:
|
||||
self.scheduler.remove_job(self._sync_job.id)
|
||||
logger.info(
|
||||
"[IdMappingService.start_scheduler][Reflect] Removed existing sync job."
|
||||
)
|
||||
def sync_all():
|
||||
seed_trace_id()
|
||||
for env_id in environments:
|
||||
client = superset_client_factory(env_id)
|
||||
if client:
|
||||
import asyncio
|
||||
asyncio.run(self.sync_environment(env_id, client))
|
||||
self._sync_job = self.scheduler.add_job(
|
||||
sync_all,
|
||||
CronTrigger.from_crontab(cron_string),
|
||||
id="id_mapping_sync_job",
|
||||
replace_existing=True,
|
||||
)
|
||||
if not self.scheduler.running:
|
||||
self.scheduler.start()
|
||||
logger.info(
|
||||
f"[IdMappingService.start_scheduler][Coherence:OK] Started background scheduler with cron: {cron_string}"
|
||||
)
|
||||
else:
|
||||
logger.info(
|
||||
f"[IdMappingService.start_scheduler][Coherence:OK] Updated background scheduler with cron: {cron_string}"
|
||||
)
|
||||
# #endregion start_scheduler
|
||||
# #region sync_environment [TYPE Function]
|
||||
# @ingroup Core
|
||||
# @PURPOSE: Fully synchronizes mapping for a specific environment.
|
||||
|
||||
@@ -3,12 +3,14 @@
|
||||
# @BRIEF Manages scheduled tasks using APScheduler.
|
||||
# @LAYER Core
|
||||
# @RELATION DEPENDS_ON -> TaskManager
|
||||
# @RELATION DEPENDS_ON -> [AsyncJobRunner]
|
||||
import asyncio
|
||||
from datetime import date, datetime, time, timedelta
|
||||
|
||||
from apscheduler.schedulers.background import BackgroundScheduler
|
||||
from apscheduler.triggers.cron import CronTrigger
|
||||
|
||||
from .async_job_runner import AsyncJobRunner
|
||||
from .config_manager import ConfigManager
|
||||
from .cot_logger import seed_trace_id
|
||||
from .database import SessionLocal
|
||||
@@ -29,7 +31,11 @@ class SchedulerService:
|
||||
self.task_manager = task_manager
|
||||
self.config_manager = config_manager
|
||||
self.scheduler = BackgroundScheduler()
|
||||
self.loop = asyncio.get_event_loop()
|
||||
try:
|
||||
self.loop = asyncio.get_running_loop()
|
||||
except RuntimeError:
|
||||
self.loop = asyncio.new_event_loop()
|
||||
self.runner = AsyncJobRunner(self.loop)
|
||||
# #endregion __init__
|
||||
# #region start [TYPE Function]
|
||||
# @ingroup Core
|
||||
@@ -40,7 +46,7 @@ class SchedulerService:
|
||||
with belief_scope("SchedulerService.start"):
|
||||
if not self.scheduler.running:
|
||||
self.scheduler.start()
|
||||
logger.info("Scheduler started.")
|
||||
logger.reason("Scheduler started")
|
||||
self.load_schedules()
|
||||
# #endregion start
|
||||
# #region stop [TYPE Function]
|
||||
@@ -52,7 +58,7 @@ class SchedulerService:
|
||||
with belief_scope("SchedulerService.stop"):
|
||||
if self.scheduler.running:
|
||||
self.scheduler.shutdown()
|
||||
logger.info("Scheduler stopped.")
|
||||
logger.reflect("Scheduler stopped")
|
||||
# #endregion stop
|
||||
# #region load_schedules [C:4] [TYPE Function] [SEMANTICS scheduler,backup,translation,apscheduler]
|
||||
# @ingroup Core
|
||||
@@ -218,7 +224,7 @@ class SchedulerService:
|
||||
def _trigger_backup(self, env_id: str):
|
||||
seed_trace_id()
|
||||
with belief_scope("SchedulerService._trigger_backup", f"env_id={env_id}"):
|
||||
logger.info(f"Triggering scheduled backup for environment {env_id}")
|
||||
logger.reason(f"Triggering scheduled backup for environment {env_id}")
|
||||
# Check if a backup is already running for this environment
|
||||
active_tasks = self.task_manager.get_tasks(limit=100)
|
||||
for task in active_tasks:
|
||||
@@ -227,17 +233,14 @@ class SchedulerService:
|
||||
and task.status in ["PENDING", "RUNNING"]
|
||||
and task.params.get("environment_id") == env_id
|
||||
):
|
||||
logger.warning(
|
||||
f"Backup already running for environment {env_id}. Skipping scheduled run."
|
||||
)
|
||||
logger.explore(f"Backup already running for environment {env_id}, skipping scheduled run",
|
||||
error="Concurrent backup in progress")
|
||||
return
|
||||
# Run the backup task
|
||||
# We need to run this in the event loop since create_task is async
|
||||
asyncio.run_coroutine_threadsafe(
|
||||
# Run the backup task via AsyncJobRunner
|
||||
self.runner.run(
|
||||
self.task_manager.create_task(
|
||||
"superset-backup", {"environment_id": env_id}
|
||||
),
|
||||
self.loop,
|
||||
)
|
||||
)
|
||||
# #endregion _trigger_backup
|
||||
# #region add_validation_job [C:3] [TYPE Function] [SEMANTICS scheduler,validation,cron,apscheduler]
|
||||
@@ -365,20 +368,25 @@ class SchedulerService:
|
||||
current_date=today,
|
||||
)
|
||||
|
||||
now = datetime.now()
|
||||
for idx, dash_id in enumerate(dashboard_ids):
|
||||
sched_time = scheduled_times[idx] if idx < len(scheduled_times) else scheduled_times[-1]
|
||||
delay = (sched_time - now).total_seconds()
|
||||
params: dict = {
|
||||
"dashboard_id": dash_id,
|
||||
"environment_id": policy.environment_id,
|
||||
"provider_id": policy.provider_id or "",
|
||||
"policy_id": policy_id,
|
||||
}
|
||||
asyncio.run_coroutine_threadsafe(
|
||||
self.task_manager.create_task(
|
||||
plugin_id="llm_dashboard_validation", params=params
|
||||
),
|
||||
self.loop,
|
||||
coro = self.task_manager.create_task(
|
||||
plugin_id="llm_dashboard_validation", params=params
|
||||
)
|
||||
if delay <= 0:
|
||||
# Time already passed — run immediately
|
||||
self.runner.run(coro)
|
||||
else:
|
||||
# Delayed execution within the window
|
||||
self.runner.run_later(coro, delay_seconds=delay)
|
||||
logger.reason(
|
||||
f"Scheduled validation for dashboard {dash_id} at {sched_time.isoformat()}"
|
||||
)
|
||||
|
||||
@@ -23,6 +23,7 @@ from fastapi import Depends, HTTPException, Request, status
|
||||
from fastapi.security import OAuth2PasswordBearer
|
||||
from jose import JWTError
|
||||
|
||||
from .core.async_job_runner import AsyncJobRunner
|
||||
from .core.auth.jwt import decode_token, is_token_blacklisted
|
||||
from .core.auth.repository import AuthRepository
|
||||
from .core.config_manager import ConfigManager
|
||||
@@ -58,6 +59,7 @@ config_manager: ConfigManager | None = None
|
||||
plugin_loader: PluginLoader | None = None
|
||||
task_manager: TaskManager | None = None
|
||||
scheduler_service: SchedulerService | None = None
|
||||
async_job_runner: AsyncJobRunner | None = None
|
||||
resource_service: ResourceService | None = None
|
||||
|
||||
|
||||
@@ -155,6 +157,28 @@ def get_scheduler_service() -> SchedulerService:
|
||||
# #endregion get_scheduler_service
|
||||
|
||||
|
||||
# #region get_async_job_runner [C:1] [TYPE Function]
|
||||
# @BRIEF Dependency injector for AsyncJobRunner.
|
||||
# @PRE Event loop is running (FastAPI lifespan).
|
||||
# @POST Returns shared AsyncJobRunner instance.
|
||||
def get_async_job_runner() -> AsyncJobRunner:
|
||||
"""Dependency injector for AsyncJobRunner."""
|
||||
import asyncio
|
||||
global async_job_runner
|
||||
if async_job_runner is None:
|
||||
try:
|
||||
loop = asyncio.get_running_loop()
|
||||
except RuntimeError:
|
||||
loop = asyncio.new_event_loop()
|
||||
asyncio.set_event_loop(loop)
|
||||
async_job_runner = AsyncJobRunner(loop)
|
||||
logger.info("AsyncJobRunner initialized")
|
||||
return async_job_runner
|
||||
|
||||
|
||||
# #endregion get_async_job_runner
|
||||
|
||||
|
||||
# #region get_resource_service [C:1] [TYPE Function]
|
||||
# @BRIEF Dependency injector for ResourceService.
|
||||
# @PRE Global resource_service must be initialized.
|
||||
|
||||
@@ -1,59 +0,0 @@
|
||||
# #region LLMAnalysisScheduler [C:3] [TYPE Module] [SEMANTICS scheduler, llm, validation, task, cron]
|
||||
# @defgroup LLMAnalysis Module group.
|
||||
# @BRIEF Provides helper functions to schedule LLM-based validation tasks.
|
||||
# @LAYER Domain
|
||||
# @RELATION DEPENDS_ON -> [SchedulerService]
|
||||
|
||||
from typing import Any
|
||||
|
||||
from ...core.logger import belief_scope, logger
|
||||
from ...dependencies import get_scheduler_service, get_task_manager
|
||||
|
||||
|
||||
# #region schedule_dashboard_validation [TYPE Function]
|
||||
# @ingroup LLMAnalysis
|
||||
# @BRIEF Schedules a recurring dashboard validation task.
|
||||
# @SIDE_EFFECT Adds a job to the scheduler service.
|
||||
def schedule_dashboard_validation(dashboard_id: str, cron_expression: str, params: dict[str, Any]):
|
||||
with belief_scope("schedule_dashboard_validation", f"dashboard_id={dashboard_id}"):
|
||||
scheduler = get_scheduler_service()
|
||||
task_manager = get_task_manager()
|
||||
|
||||
job_id = f"llm_val_{dashboard_id}"
|
||||
|
||||
async def job_func():
|
||||
await task_manager.create_task(
|
||||
plugin_id="llm_dashboard_validation",
|
||||
params={
|
||||
"dashboard_id": dashboard_id,
|
||||
**params
|
||||
}
|
||||
)
|
||||
|
||||
scheduler.add_job(
|
||||
job_func,
|
||||
"cron",
|
||||
id=job_id,
|
||||
replace_existing=True,
|
||||
**_parse_cron(cron_expression)
|
||||
)
|
||||
logger.info(f"Scheduled validation for dashboard {dashboard_id} with cron {cron_expression}")
|
||||
# #endregion schedule_dashboard_validation
|
||||
|
||||
# #region _parse_cron [TYPE Function]
|
||||
# @BRIEF Basic cron parser placeholder.
|
||||
def _parse_cron(cron: str) -> dict[str, str]:
|
||||
# Basic cron parser placeholder
|
||||
parts = cron.split()
|
||||
if len(parts) != 5:
|
||||
return {}
|
||||
return {
|
||||
"minute": parts[0],
|
||||
"hour": parts[1],
|
||||
"day": parts[2],
|
||||
"month": parts[3],
|
||||
"day_of_week": parts[4]
|
||||
}
|
||||
# #endregion _parse_cron
|
||||
|
||||
# #endregion LLMAnalysisScheduler
|
||||
@@ -7,6 +7,7 @@
|
||||
# @RELATION DEPENDS_ON -> MigrationEngine
|
||||
# @RELATION DEPENDS_ON -> IdMappingService
|
||||
# @RELATION DEPENDS_ON -> [TaskContext]
|
||||
# @RELATION DEPENDS_ON -> [AsyncJobRunner]
|
||||
# @PRE Plugin loader can resolve infrastructure dependencies and execution requests provide validated migration context.
|
||||
# @POST Plugin metadata remains stable and migration execution preserves mapped-environment import guarantees.
|
||||
# @SIDE_EFFECT Reads config, opens database sessions, creates temporary artifacts, and triggers Superset export/import workflows.
|
||||
@@ -24,7 +25,7 @@ from ..core.plugin_base import PluginBase
|
||||
from ..core.superset_client import SupersetClient
|
||||
from ..core.task_manager.context import TaskContext
|
||||
from ..core.utils.fileio import create_temp_file
|
||||
from ..dependencies import get_config_manager
|
||||
from ..dependencies import get_async_job_runner, get_config_manager
|
||||
from ..models.mapping import DatabaseMapping, Environment
|
||||
|
||||
|
||||
@@ -224,7 +225,7 @@ class MigrationPlugin(PluginBase):
|
||||
if not from_c or not to_c:
|
||||
raise ValueError(f"Clients not initialized for environments: {from_env_name}, {to_env_name}")
|
||||
|
||||
_, all_dashboards = from_c.get_dashboards()
|
||||
_, all_dashboards = await from_c.get_dashboards()
|
||||
|
||||
# Selection Logic
|
||||
if selected_ids:
|
||||
@@ -277,7 +278,7 @@ class MigrationPlugin(PluginBase):
|
||||
app_logger.reason(f"Starting pipeline for dashboard '{title}'", extra={"dash_id": dash_id})
|
||||
|
||||
try:
|
||||
exported_content, _ = from_c.export_dashboard(dash_id)
|
||||
exported_content, _ = await from_c.export_dashboard(dash_id)
|
||||
with create_temp_file(content=exported_content, dry_run=True, suffix=".zip") as tmp_zip_path, create_temp_file(suffix=".zip", dry_run=True) as tmp_new_zip:
|
||||
|
||||
success = engine.transform_zip(
|
||||
@@ -318,7 +319,7 @@ class MigrationPlugin(PluginBase):
|
||||
|
||||
if success:
|
||||
app_logger.reason("Pushing transformed ZIP to target Superset")
|
||||
to_c.import_dashboard(file_name=tmp_new_zip, dash_id=dash_id, dash_slug=dash_slug)
|
||||
await to_c.import_dashboard(file_name=tmp_new_zip, dash_id=dash_id, dash_slug=dash_slug)
|
||||
migration_result["migrated_dashboards"].append({"id": dash_id, "title": title})
|
||||
app_logger.reflect("Import successful", extra={"title": title})
|
||||
else:
|
||||
@@ -355,7 +356,7 @@ class MigrationPlugin(PluginBase):
|
||||
|
||||
if passwords:
|
||||
app_logger.reason(f"Retrying import for {title} with injected credentials")
|
||||
to_c.import_dashboard(file_name=tmp_new_zip, dash_id=dash_id, dash_slug=dash_slug, passwords=passwords)
|
||||
await to_c.import_dashboard(file_name=tmp_new_zip, dash_id=dash_id, dash_slug=dash_slug, passwords=passwords)
|
||||
migration_result["migrated_dashboards"].append({"id": dash_id, "title": title})
|
||||
app_logger.reflect("Password injection unblocked import")
|
||||
if "passwords" in task.params:
|
||||
@@ -373,7 +374,8 @@ class MigrationPlugin(PluginBase):
|
||||
app_logger.reason("Executing incremental ID catalog sync on target")
|
||||
db_session = SessionLocal()
|
||||
mapping_service = IdMappingService(db_session)
|
||||
mapping_service.sync_environment(tgt_env.id, to_c, incremental=True)
|
||||
runner = get_async_job_runner()
|
||||
runner.run(mapping_service.sync_environment(tgt_env.id, to_c, incremental=True))
|
||||
db_session.close()
|
||||
app_logger.reflect("Incremental catalog sync closed out cleanly")
|
||||
except Exception as sync_exc:
|
||||
|
||||
@@ -6,6 +6,7 @@
|
||||
# @RELATION DEPENDS_ON -> [SchedulerService]
|
||||
# @RELATION DEPENDS_ON -> [TranslationOrchestrator]
|
||||
# @RELATION DEPENDS_ON -> [TranslationEventLog]
|
||||
# @RELATION DEPENDS_ON -> [AsyncJobRunner]
|
||||
# @PRE Database session and SchedulerService are available.
|
||||
# @POST TranslationSchedule CRUD persisted; APScheduler jobs registered/updated/removed.
|
||||
# @SIDE_EFFECT Registers APScheduler jobs; runs translations on trigger; creates events.
|
||||
@@ -259,6 +260,8 @@ def execute_scheduled_translation(
|
||||
execution_mode: str = "full",
|
||||
) -> None:
|
||||
"""APScheduler job callback for scheduled translations."""
|
||||
from ...dependencies import get_async_job_runner
|
||||
runner = get_async_job_runner()
|
||||
seed_trace_id()
|
||||
db: Session = db_session_maker()
|
||||
try:
|
||||
@@ -381,9 +384,9 @@ def execute_scheduled_translation(
|
||||
payload={"reason": "baseline_expired", "age_days": age.days},
|
||||
)
|
||||
|
||||
# Execute in same thread (APScheduler runs in background thread pool)
|
||||
# Execute via AsyncJobRunner (dispatches coroutine to event loop)
|
||||
try:
|
||||
orch.execute_run(run)
|
||||
runner.run(orch.execute_run(run))
|
||||
if run.status == "FAILED":
|
||||
logger.explore("Scheduled translation completed with all records failed", {
|
||||
"run_id": run.id,
|
||||
@@ -408,9 +411,8 @@ def execute_scheduled_translation(
|
||||
f"Error: {exec_err}\n"
|
||||
f"Time: {datetime.now(UTC).isoformat()}"
|
||||
)
|
||||
import asyncio
|
||||
for provider in notify_service._providers.values():
|
||||
asyncio.run(provider.send(recipient="admin", subject=subject, body=body))
|
||||
runner.run(provider.send(recipient="admin", subject=subject, body=body))
|
||||
except Exception as notify_err:
|
||||
logger.warning(f"Failed to send failure notification: {notify_err}")
|
||||
|
||||
@@ -429,15 +431,15 @@ def execute_scheduled_translation(
|
||||
"status": run.status,
|
||||
})
|
||||
except Exception as e:
|
||||
logger.error(f"[scheduled_translation] Unexpected error: {e}")
|
||||
logger.explore(f"[scheduled_translation] Unexpected error: {e}",
|
||||
error="Scheduled translation crashed outside of main execution block")
|
||||
try:
|
||||
notify_service = NotificationService(db, config_manager)
|
||||
notify_service._initialize_providers()
|
||||
subject = f"Scheduled translation CRASHED: job={job_id}"
|
||||
body = f"Scheduled translation for job '{job_id}' crashed unexpectedly.\nError: {e}\nTime: {datetime.now(UTC).isoformat()}"
|
||||
import asyncio
|
||||
for provider in notify_service._providers.values():
|
||||
asyncio.run(provider.send(recipient="admin", subject=subject, body=body))
|
||||
runner.run(provider.send(recipient="admin", subject=subject, body=body))
|
||||
except Exception:
|
||||
pass
|
||||
finally:
|
||||
|
||||
Reference in New Issue
Block a user