Coverage for node / src / stigmem_node / routes / aliases.py: 88%

1"""Entity alias management routes — spec §2.6 Phase 6. 

2 

3POST /v1/aliases — register a user-defined semantic alias 

4GET /v1/aliases — list aliases (filterable by kind / canonical_uri) 

5DELETE /v1/aliases/{raw_uri} — remove a user-defined alias (migration aliases protected) 

6""" 

7 

8from __future__ import annotations 

9 

10from typing import Annotated, Any 

11from urllib.parse import unquote 

12 

13from fastapi import APIRouter, Depends, HTTPException, Query, status 

14 

15from ..auth import Identity, resolve_identity 

16from ..db import db 

17from ..models.aliases import AliasRecord, AliasRequest 

18from ..recall.fuzzy_resolver import register_alias 

19 

20router = APIRouter(prefix="/v1/aliases", tags=["aliases"]) 

21 

22_VALID_KINDS = {"user", "migration"} 

23 

24 

25@router.post("", response_model=AliasRecord, status_code=status.HTTP_201_CREATED) 

26def create_alias( 

27 req: AliasRequest, 

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

29) -> AliasRecord: 

30 """Register a user-defined semantic alias (raw_uri ≡ canonical_uri).""" 

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

32 raise HTTPException( 

33 status_code=status.HTTP_403_FORBIDDEN, detail="write permission required" 

34 ) 

35 

36 with db() as conn: 

37 try: 

38 result = register_alias( 

39 conn, req.raw_uri, req.canonical_uri, kind="user", tenant_id=identity.tenant_id 

40 ) 

41 except ValueError as exc: 

42 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc 

43 

44 return AliasRecord(**result) 

45 

46 

47@router.get("", response_model=list[AliasRecord]) 

48def list_aliases( 

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

50 kind: str | None = Query(None, description="Filter by kind: 'user' or 'migration'"), 

51 canonical_uri: str | None = Query( 

52 None, description="Return all aliases that resolve to this URI" 

53 ), 

54) -> list[AliasRecord]: 

55 """List registered entity aliases.""" 

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

57 raise HTTPException( 

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

59 ) 

60 

61 if kind and kind not in _VALID_KINDS: 

62 raise HTTPException( 

63 status_code=status.HTTP_400_BAD_REQUEST, 

64 detail=f"kind must be one of {sorted(_VALID_KINDS)}", 

65 ) 

66 

67 # Always scope to the caller's tenant (aliases are tenant-isolated). The 

68 # base `tenant_id = ?` predicate is inlined in the SQL literal (not a 

69 # variable) so it is statically verifiable by the tenant-scope CI guard. 

70 params: list[Any] = [identity.tenant_id] 

71 extra = "" 

72 if kind: 

73 extra += " AND kind = ?" 

74 params.append(kind) 

75 if canonical_uri: 

76 extra += " AND canonical_uri = ?" 

77 params.append(canonical_uri) 

78 

79 # Base predicate is a pure literal (incl. `tenant_id = ?` so the CI tenant-scope 

80 # guard sees it); `extra` is built only from literal fragments, values in params. 

81 base_sql = "SELECT raw_uri, canonical_uri, kind, created_at FROM entity_aliases WHERE tenant_id = ?" # noqa: E501 

82 sql = base_sql + extra + " ORDER BY created_at DESC" # noqa: S608 # nosec B608 

83 with db() as conn: 

84 rows = conn.execute(sql, params).fetchall() 

85 

86 return [AliasRecord(**dict(r)) for r in rows] 

87 

88 

89@router.delete("/{raw_uri:path}", status_code=status.HTTP_204_NO_CONTENT) 

90def delete_alias( 

91 raw_uri: str, 

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

93) -> None: 

94 """Remove a user-defined alias. Migration aliases cannot be deleted via API.""" 

95 if not identity.can_write(): 95 ↛ 96line 95 didn't jump to line 96 because the condition on line 95 was never true

96 raise HTTPException( 

97 status_code=status.HTTP_403_FORBIDDEN, detail="write permission required" 

98 ) 

99 

100 decoded = unquote(raw_uri) 

101 

102 with db() as conn: 

103 row = conn.execute( 

104 "SELECT kind FROM entity_aliases WHERE raw_uri = ? AND tenant_id = ?", 

105 (decoded, identity.tenant_id), 

106 ).fetchone() 

107 if row is None: 

108 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="alias not found") 

109 if row["kind"] != "user": 109 ↛ 110line 109 didn't jump to line 110 because the condition on line 109 was never true

110 raise HTTPException( 

111 status_code=status.HTTP_403_FORBIDDEN, 

112 detail=( 

113 "migration aliases are managed by the migration sweep " 

114 "and cannot be deleted via API" 

115 ), 

116 ) 

117 conn.execute( 

118 "DELETE FROM entity_aliases WHERE raw_uri = ? AND tenant_id = ?", 

119 (decoded, identity.tenant_id), 

120 )