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 = ? AND tenant_id = ?", 

49 (fact_id, identity.tenant_id), 

50 ).fetchone() 

51 if alias is None: 

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

53 resolved_fact_id = alias["fact_id"] 

54 

55 with db() as conn: 

56 row = conn.execute( 

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

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

59 (resolved_fact_id, identity.tenant_id), 

60 ).fetchone() 

61 if row is None: 

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

63 

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

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

66 

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

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

69 

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

71 row_keys = row.keys() 

72 garden_id = ( 

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

74 ) 

75 if garden_id is not None: 

76 with db() as conn: 

77 garden_row = conn.execute( 

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

79 (garden_id, identity.tenant_id), 

80 ).fetchone() 

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

82 require_garden_read(dict(garden_row), identity) 

83 

84 with db() as conn: 

85 sibling_count: int = conn.execute( 

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

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

88 ).fetchone()[0] 

89 enforce_read_path_cid(row) 

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

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

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

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

94 with db() as conn: 

95 record_read_scopes( 

96 conn, 

97 identity=identity, 

98 session_id=session_id, 

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

100 ) 

101 return pipeline_results[0] 

102 # Pending-quarantine facts return 404 to normal callers 

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