Coverage for src/qdrant_loader/cli/commands/jobs_cmd.py: 38%
104 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-20 10:15 +0000
1"""
2qdrant-loader jobs admin CLI commands.
4Subcommands: list [--status], retry <id>, trigger --source-type --source --mode, cancel <id>.
5"""
7from __future__ import annotations
9import asyncio
10import json
11from pathlib import Path
13import click
14from click.exceptions import ClickException
15from click.types import Choice
16from click.types import Path as ClickPath
18from qdrant_loader.core.worker.job_types import JobType
21def _init_queue(workspace: Path | None, config: Path | None, env: Path | None):
22 """Load config, build StateManager. Returns state_manager."""
23 from qdrant_loader.cli.config_loader import (
24 load_config_with_workspace,
25 setup_workspace,
26 )
27 from qdrant_loader.config import get_global_config
28 from qdrant_loader.config.workspace import validate_workspace_flags
29 from qdrant_loader.core.state.state_manager import StateManager
31 # Validate mutually exclusive flags early
32 try:
33 validate_workspace_flags(workspace, config, env)
34 except ValueError as e:
35 raise ClickException(str(e))
37 if workspace:
38 ws_config = setup_workspace(workspace)
39 load_config_with_workspace(workspace_config=ws_config, skip_validation=True)
40 else:
41 resolved_config = config
42 if resolved_config is None:
43 default_config = Path("config.yaml")
44 resolved_config = default_config if default_config.exists() else None
45 if resolved_config is None:
46 raise ClickException("No config found. Use --workspace or --config.")
47 load_config_with_workspace(
48 workspace_config=None,
49 config_path=resolved_config,
50 env_path=env,
51 skip_validation=True,
52 )
54 global_config = get_global_config()
55 state_manager = StateManager(global_config.state_management)
56 return state_manager
59async def _run_with_queue(workspace, config, env, coro_factory):
60 """Initialize queue, run coro_factory(queue), dispose state manager."""
61 state_manager = _init_queue(workspace, config, env)
62 await state_manager.initialize()
63 queue_instance = None
64 try:
65 from qdrant_loader.core.worker.queue import SQLiteJobQueue
67 queue_instance = SQLiteJobQueue(
68 state_manager.session_factory,
69 db_op_lock=state_manager.queue_db_op_lock,
70 )
71 return await coro_factory(queue_instance)
72 finally:
73 await state_manager.dispose()
76# ────────────────────────────────────────────────
77# Click group
78# ────────────────────────────────────────────────
81@click.group("jobs")
82def jobs_cmd():
83 """Inspect and manage background ingestion jobs."""
86def _common_options(fn):
87 """Decorator that adds --workspace, --config, --env options."""
88 fn = click.option(
89 "--workspace",
90 type=ClickPath(path_type=Path),
91 default=None,
92 help="Workspace directory.",
93 )(fn)
94 fn = click.option(
95 "--config",
96 "config_path",
97 type=ClickPath(path_type=Path),
98 default=None,
99 help="Path to config.yaml.",
100 )(fn)
101 fn = click.option(
102 "--env",
103 "env_path",
104 type=ClickPath(path_type=Path),
105 default=None,
106 help="Path to .env file.",
107 )(fn)
108 return fn
111# ────────────────────────────────────────────────
112# jobs list
113# ────────────────────────────────────────────────
116@jobs_cmd.command("list")
117@click.option(
118 "--status",
119 type=Choice(
120 ["pending", "running", "done", "failed", "cancelled"], case_sensitive=False
121 ),
122 default=None,
123 help="Filter by job status.",
124)
125@click.option(
126 "--limit",
127 type=click.IntRange(min=1, max=1000),
128 default=50,
129 show_default=True,
130 help="Max rows to return.",
131)
132@click.option("--json", "output_json", is_flag=True, help="Output as JSON.")
133@_common_options
134def jobs_list(status, limit, output_json, workspace, config_path, env_path):
135 """List jobs, optionally filtered by status."""
137 async def _list(queue):
138 jobs = await queue.list(status=status, limit=limit)
139 if output_json:
140 rows = []
141 for j in jobs:
142 rows.append(
143 {
144 "id": j.id,
145 "type": j.type,
146 "status": j.status,
147 "attempts": j.attempts,
148 "enqueued_at": (
149 j.enqueued_at.isoformat() if j.enqueued_at else None
150 ),
151 "started_at": (
152 j.started_at.isoformat() if j.started_at else None
153 ),
154 "finished_at": (
155 j.finished_at.isoformat() if j.finished_at else None
156 ),
157 "last_error": j.last_error,
158 "payload": json.loads(j.payload_json) if j.payload_json else {},
159 }
160 )
161 click.echo(json.dumps(rows, indent=2))
162 else:
163 if not jobs:
164 click.echo("No jobs found.")
165 return
166 header = f"{'ID':>6} {'STATUS':<10} {'TYPE':<20} {'ATTEMPTS':>8} {'ENQUEUED_AT':<27} LAST_ERROR"
167 click.echo(header)
168 click.echo("-" * len(header))
169 for j in jobs:
170 enq = j.enqueued_at.isoformat() if j.enqueued_at else ""
171 err = (j.last_error or "")[:40]
172 click.echo(
173 f"{j.id:>6} {j.status:<10} {j.type:<20} {j.attempts:>8} {enq:<27} {err}"
174 )
176 asyncio.run(_run_with_queue(workspace, config_path, env_path, _list))
179# ────────────────────────────────────────────────
180# jobs retry
181# ────────────────────────────────────────────────
184@jobs_cmd.command("retry")
185@click.argument("job_id", type=int)
186@_common_options
187def jobs_retry(job_id, workspace, config_path, env_path):
188 """Reset a failed or done job back to pending."""
190 async def _retry(queue):
191 ok = await queue.reset_to_pending(job_id)
192 if ok:
193 click.echo(f"Job {job_id} reset to pending.")
194 else:
195 raise ClickException(
196 f"Job {job_id} not found or is not in failed/done state."
197 )
199 asyncio.run(_run_with_queue(workspace, config_path, env_path, _retry))
202# ────────────────────────────────────────────────
203# jobs trigger
204# ────────────────────────────────────────────────
207@jobs_cmd.command("trigger")
208@click.option(
209 "--source-type", required=True, help="Source type (e.g. git, confluence)."
210)
211@click.option("--source", required=True, help="Source name as configured.")
212@click.option(
213 "--mode",
214 type=Choice(["bulk", "incremental"], case_sensitive=False),
215 required=True,
216 help="Ingestion mode.",
217)
218@click.option("--project", "project_id", required=True, help="Project ID.")
219@_common_options
220def jobs_trigger(
221 source_type, source, mode, project_id, workspace, config_path, env_path
222):
223 """Enqueue a new ingestion job immediately."""
225 async def _trigger(queue):
226 job_type = (
227 JobType.BULK_INGEST if mode.lower() == "bulk" else JobType.INCREMENTAL_PULL
228 )
229 payload = {
230 "project_id": project_id,
231 "source_type": source_type,
232 "source": source,
233 "source_lock": f"{project_id}:{source_type}:{source}",
234 }
235 job = await queue.enqueue(job_type, payload)
236 click.echo(f"Enqueued job {job.id} (type={job_type}, status={job.status}).")
238 asyncio.run(_run_with_queue(workspace, config_path, env_path, _trigger))
241# ────────────────────────────────────────────────
242# jobs cancel
243# ────────────────────────────────────────────────
246@jobs_cmd.command("cancel")
247@click.argument("job_id", type=int)
248@_common_options
249def jobs_cancel(job_id, workspace, config_path, env_path):
250 """Cancel a pending job."""
252 async def _cancel(queue):
253 ok = await queue.cancel(job_id)
254 if ok:
255 click.echo(f"Job {job_id} cancelled.")
256 else:
257 raise ClickException(f"Job {job_id} not found or is not in pending state.")
259 asyncio.run(_run_with_queue(workspace, config_path, env_path, _cancel))