Coverage for node / src / stigmem_node / routes / facts / single.py: 90%

1"""Single-fact retrieval route.""" 

2 

3from __future__ import annotations 

4 

5from typing import Annotated 

6 

7from fastapi import Depends, Header, HTTPException, status 

8 

9from ...auth import Identity, resolve_identity 

10from ...cid import is_cid, is_valid_cid 

11from ...db import db 

12from ...garden_acl import require_garden_read 

13from ...models.facts import FactRecord, row_to_record 

14from ...recall.recall_pipeline import apply_recall_pipeline 

15from ...session_graph import record_read_scopes 

16from ..cid_integrity import enforce_read_path_cid 

17from .common import FACT_PROJECTION_JOINS, FACT_PROJECTION_SELECT, router 

18 

19 

20@router.get("/{fact_id}", response_model=FactRecord) 

21def get_fact( 

22 fact_id: str, 

23 identity: Annotated[Identity, Depends(resolve_identity)], 

24 session_id: Annotated[str | None, Header(alias="Stigmem-Session")] = None, 

25) -> FactRecord: 

26 """Retrieve a single fact by UUID or sha256: CID. 

27 

28 Covered by Spec-03-HTTP-API and Spec-21-Content-Addressed-IDs. 

29 """ 

30 if not identity.can_read(): 30 ↛ 31line 30 didn't jump to line 31 because the condition on line 30 was never true

31 raise HTTPException( 

32 status_code=status.HTTP_403_FORBIDDEN, detail="read permission required" 

33 ) # noqa: E501 

34 

35 # §25.5: dual addressing — resolve CID to UUID via alias table 

36 resolved_fact_id = fact_id 

37 if is_cid(fact_id): 

38 if not is_valid_cid(fact_id): 

39 raise HTTPException( 

40 status_code=400, 

41 detail={ 

42 "code": "cid_malformed", 

43 "message": "CID must be 'sha256:' followed by 64 hex chars", 

44 }, # noqa: E501 

45 ) 

46 with db() as conn: 

47 alias = conn.execute( 

48 "SELECT fact_id FROM fact_cid_aliases WHERE cid = ?", (fact_id,) 

49 ).fetchone() 

50 if alias is None: 

51 raise HTTPException(status_code=404, detail="fact not found") 

52 resolved_fact_id = alias["fact_id"] 

53 

54 with db() as conn: 

55 row = conn.execute( 

56 f"SELECT {FACT_PROJECTION_SELECT} FROM facts f {FACT_PROJECTION_JOINS} " # noqa: S608 # nosec B608 

57 "WHERE f.id = ? AND f.tenant_id = ?", 

58 (resolved_fact_id, identity.tenant_id), 

59 ).fetchone() 

60 if row is None: 

61 raise HTTPException(status_code=404, detail="fact not found") 

62 

63 # F-11 §25.6.1/§23.3.3: tombstone indistinguishability — tombstoned facts return 404 

64 from ...lifecycle.tombstone_cache import is_tombstoned as _is_tombstoned_check 

65 

66 if _is_tombstoned_check(row["entity"], identity.tenant_id): 66 ↛ 67line 66 didn't jump to line 67 because the condition on line 66 was never true

67 raise HTTPException(status_code=404, detail="fact not found") 

68 

69 # Garden ACL: fact in a garden is only readable by members (spec §17.3) 

70 row_keys = row.keys() 

71 garden_id = ( 

72 row["projected_garden_id"] if "projected_garden_id" in row_keys else row["garden_id"] 

73 ) 

74 if garden_id is not None: 

75 with db() as conn: 

76 garden_row = conn.execute( 

77 "SELECT * FROM gardens WHERE id = ? AND tenant_id = ?", 

78 (garden_id, identity.tenant_id), 

79 ).fetchone() 

80 if garden_row is not None: 80 ↛ 83line 80 didn't jump to line 83 because the condition on line 80 was always true

81 require_garden_read(dict(garden_row), identity) 

82 

83 with db() as conn: 

84 sibling_count: int = conn.execute( 

85 "SELECT COUNT(*) FROM facts WHERE entity=? AND relation=? AND scope=? AND tenant_id=?", 

86 (row["entity"], row["relation"], row["scope"], identity.tenant_id), 

87 ).fetchone()[0] 

88 enforce_read_path_cid(row) 

89 record = row_to_record(row, contradicted=sibling_count > 1) 

90 # v1.1: recall pipeline (trust multiplier + sanitizer) 

91 pipeline_results = apply_recall_pipeline([record], identity=identity, include_low_trust=True) 

92 if pipeline_results: 92 ↛ 102line 92 didn't jump to line 102 because the condition on line 92 was always true

93 with db() as conn: 

94 record_read_scopes( 

95 conn, 

96 identity=identity, 

97 session_id=session_id, 

98 scopes={pipeline_results[0].scope}, 

99 ) 

100 return pipeline_results[0] 

101 # Pending-quarantine facts return 404 to normal callers 

102 raise HTTPException(status_code=404, detail="fact not found")