74 lines
2.0 KiB
Python
74 lines
2.0 KiB
Python
import asyncio
|
|
import traceback
|
|
from datetime import datetime
|
|
|
|
from app.database import Base, SessionLocal, engine
|
|
from app.linkedin import fetch_linkedin_job
|
|
from app.models import FetchTask, Job
|
|
from app.seed import seed_jobs
|
|
|
|
|
|
async def process_task(task_id: int) -> None:
|
|
db = SessionLocal()
|
|
try:
|
|
task = db.get(FetchTask, task_id)
|
|
if not task:
|
|
return
|
|
task.status = "running"
|
|
task.started_at = datetime.utcnow()
|
|
db.commit()
|
|
parsed = await fetch_linkedin_job(task.url)
|
|
cols = set(Job.__table__.c.keys())
|
|
job = Job(**{k: v for k, v in parsed.items() if k in cols and k != "id"})
|
|
db.add(job)
|
|
db.flush()
|
|
task.job_id = job.id
|
|
task.status = "done"
|
|
task.finished_at = datetime.utcnow()
|
|
task.error = ""
|
|
db.commit()
|
|
print(f"[worker] stored job {job.id} from {task.url}")
|
|
except Exception as exc:
|
|
db.rollback()
|
|
task = db.get(FetchTask, task_id)
|
|
if task:
|
|
task.status = "error"
|
|
task.error = f"{exc}\n{traceback.format_exc()[-800:]}"
|
|
task.finished_at = datetime.utcnow()
|
|
db.commit()
|
|
print(f"[worker] failed task {task_id}: {exc}")
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
async def loop() -> None:
|
|
Base.metadata.create_all(bind=engine)
|
|
db = SessionLocal()
|
|
seed_jobs(db)
|
|
db.close()
|
|
print("[worker] listening for LinkedIn fetch tasks")
|
|
while True:
|
|
db = SessionLocal()
|
|
try:
|
|
task = (
|
|
db.query(FetchTask)
|
|
.filter(FetchTask.status == "queued")
|
|
.order_by(FetchTask.id.asc())
|
|
.first()
|
|
)
|
|
task_id = task.id if task else None
|
|
finally:
|
|
db.close()
|
|
if task_id:
|
|
await process_task(task_id)
|
|
else:
|
|
await asyncio.sleep(2)
|
|
|
|
|
|
def main() -> None:
|
|
asyncio.run(loop())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|