Coverage for node / src / stigmem_node / jobs.py: 85%

1"""Async job store — spec §14.5 / §15.4. 

2 

3Jobs are created by lint/decay routes when the scope exceeds the async threshold 

4(default 100,000 facts). Callers poll GET /v1/{lint,decay}/jobs/:job_id until 

5status is "done" or "failed". 

6""" 

7 

8from __future__ import annotations 

9 

10import json 

11import uuid 

12from datetime import UTC, datetime 

13from typing import Any 

14 

15from .db import db 

16 

17 

18def create_job(job_type: str, scope: str | None, estimated_s: int, tenant_id: str) -> str: 

19 job_id = str(uuid.uuid4()) 

20 now = datetime.now(UTC).isoformat() 

21 with db() as conn: 

22 conn.execute( 

23 "INSERT INTO jobs (id, job_type, status, scope, estimated_s, created_at, tenant_id)" 

24 " VALUES (?, ?, 'pending', ?, ?, ?, ?)", 

25 (job_id, job_type, scope, estimated_s, now, tenant_id), 

26 ) 

27 return job_id 

28 

29 

30def get_job(job_id: str, job_type: str, tenant_id: str) -> dict[str, Any] | None: 

31 """Return the job record, or None if not found (wrong type, or another tenant). 

32 

33 F-SBOLA2: the lookup is tenant-scoped, so a caller who learns a job UUID for 

34 another tenant gets None (404 at the route) rather than that tenant's record. 

35 """ 

36 with db() as conn: 

37 row = conn.execute( 

38 "SELECT * FROM jobs WHERE id = ? AND job_type = ? AND tenant_id = ?", 

39 (job_id, job_type, tenant_id), 

40 ).fetchone() 

41 if row is None: 

42 return None 

43 out: dict[str, Any] = { 

44 "job_id": row["id"], 

45 "status": row["status"], 

46 "scope": row["scope"], 

47 "estimated_s": row["estimated_s"], 

48 "created_at": row["created_at"], 

49 } 

50 if row["started_at"]: 50 ↛ 52line 50 didn't jump to line 52 because the condition on line 50 was always true

51 out["started_at"] = row["started_at"] 

52 if row["completed_at"]: 52 ↛ 54line 52 didn't jump to line 54 because the condition on line 52 was always true

53 out["completed_at"] = row["completed_at"] 

54 if row["result_json"]: 54 ↛ 56line 54 didn't jump to line 56 because the condition on line 54 was always true

55 out.update(json.loads(row["result_json"])) 

56 if row["error"]: 56 ↛ 57line 56 didn't jump to line 57 because the condition on line 56 was never true

57 out["error"] = row["error"] 

58 return out 

59 

60 

61def mark_running(job_id: str) -> None: 

62 with db() as conn: 

63 conn.execute( 

64 "UPDATE jobs SET status='running', started_at=? WHERE id=?", 

65 (datetime.now(UTC).isoformat(), job_id), 

66 ) 

67 

68 

69def mark_done(job_id: str, result: dict[str, Any]) -> None: 

70 with db() as conn: 

71 conn.execute( 

72 "UPDATE jobs SET status='done', completed_at=?, result_json=? WHERE id=?", 

73 (datetime.now(UTC).isoformat(), json.dumps(result), job_id), 

74 ) 

75 

76 

77def mark_failed(job_id: str, error: str) -> None: 

78 with db() as conn: 

79 conn.execute( 

80 "UPDATE jobs SET status='failed', completed_at=?, error=? WHERE id=?", 

81 (datetime.now(UTC).isoformat(), error, job_id), 

82 )