-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathcheck_jobs.py
More file actions
50 lines (42 loc) · 1.68 KB
/
check_jobs.py
File metadata and controls
50 lines (42 loc) · 1.68 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
#!/usr/bin/env python3
"""Quick script to check recent jobs in the database."""
from stringsight.database import SessionLocal
from stringsight.db_models.job import Job
from datetime import datetime
db = SessionLocal()
try:
# Get the 10 most recent jobs
jobs = db.query(Job).order_by(Job.created_at.desc()).limit(10).all()
print(f"\n{'='*100}")
print(f"Recent Jobs (most recent first)")
print(f"{'='*100}\n")
if not jobs:
print("No jobs found in database.")
else:
for job in jobs:
print(f"Job ID: {job.id}")
print(f" Status: {job.status}")
print(f" Type: {job.job_type if hasattr(job, 'job_type') else 'N/A'}")
print(f" Progress: {job.progress * 100:.1f}%")
print(f" Result Path: {job.result_path}")
print(f" Created: {job.created_at}")
if job.error_message:
print(f" Error: {job.error_message[:200]}")
print(f" {'-'*98}")
print(f"\n{'='*100}")
print("Jobs with Issues (failed, no result_path, or incomplete)")
print(f"{'='*100}\n")
problem_jobs = [j for j in jobs if j.status == 'failed' or
(j.status == 'completed' and not j.result_path)]
if problem_jobs:
for job in problem_jobs:
print(f"⚠️ Job ID: {job.id}")
print(f" Status: {job.status}")
print(f" Result Path: {job.result_path or 'MISSING'}")
if job.error_message:
print(f" Error: {job.error_message}")
print()
else:
print("✅ No problematic jobs found in recent 10")
finally:
db.close()