engineering/pig.py
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Fix database method names for import job status updates

Browse source
This commit is contained in:
Russell Ballestrini 2026-01-02 12:35:24 -05:00
parent 2c116c0e84
commit 59f05245d9
2 changed files with 22 additions and 8 deletions
Download patch file Download diff file Expand all files Collapse all files

16
database.py
View file

@ -373,6 +373,22 @@ class Database:
await session.execute(stmt)
await session.commit()
async def set_crawl_job_status(self, job_id: int, status: str, error: str = None) -> None:
"""Set the status of a crawl job."""
async with self.session() as session:
values = {'status': status}
if error:
values['error'] = error
if status == 'failed':
values['completed_at'] = datetime.now(timezone.utc).isoformat()
stmt = update(CrawlJob).where(CrawlJob.id == job_id).values(**values)
await session.execute(stmt)
await session.commit()
async def fail_crawl_job(self, job_id: int, error: str) -> None:
"""Mark a crawl job as failed with error message."""
await self.set_crawl_job_status(job_id, 'failed', error=error)
async def delete_crawl_job(self, job_id: int, purge_data: bool = False) -> Dict[str, Any]:
"""Delete a crawl job and optionally all associated data.

14
serp.py
View file

@ -3806,15 +3806,14 @@ async def complete_resumable_upload(upload_id: str):
# Run import in background
async def run_import():
try:
await db.update_job_status(job_id, "running")
await db.set_crawl_job_status(job_id, "running")
stats = await import_archive_to_db(final_path, job_id)
await db.update_job_stats(job_id, stats)
await db.update_job_status(job_id, "completed")
await db.complete_crawl_job(job_id, stats)
final_path.unlink(missing_ok=True)
logger.info(f"Sandbox import complete: {stats}")
except Exception as e:
logger.error(f"Sandbox import failed: {e}")
await db.update_job_status(job_id, "failed", error=str(e))
await db.fail_crawl_job(job_id, str(e))
task = asyncio.create_task(run_import())
ACTIVE_CRAWL_TASKS[job_id] = task
@ -3871,15 +3870,14 @@ async def upload_archive(file: UploadFile = File(...)):
# Run import in background
async def run_import():
try:
await db.update_job_status(job_id, "running")
await db.set_crawl_job_status(job_id, "running")
stats = await import_archive_to_db(upload_path, job_id)
await db.update_job_stats(job_id, stats)
await db.update_job_status(job_id, "completed")
await db.complete_crawl_job(job_id, stats)
upload_path.unlink(missing_ok=True)
logger.info(f"Sandbox import complete: {stats}")
except Exception as e:
logger.error(f"Sandbox import failed: {e}")
await db.update_job_status(job_id, "failed", error=str(e))
await db.fail_crawl_job(job_id, str(e))
task = asyncio.create_task(run_import())
ACTIVE_CRAWL_TASKS[job_id] = task