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(conn, req.raw_uri, req.canonical_uri, kind="user") 

39 except ValueError as exc: 

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

41 

42 return AliasRecord(**result) 

43 

44 

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

46def list_aliases( 

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

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

49 canonical_uri: str | None = Query( 

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

51 ), 

52) -> list[AliasRecord]: 

53 """List registered entity aliases.""" 

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

55 raise HTTPException( 

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

57 ) 

58 

59 if kind and kind not in _VALID_KINDS: 

60 raise HTTPException( 

61 status_code=status.HTTP_400_BAD_REQUEST, 

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

63 ) 

64 

65 conditions: list[str] = [] 

66 params: list[Any] = [] 

67 if kind: 

68 conditions.append("kind = ?") 

69 params.append(kind) 

70 if canonical_uri: 

71 conditions.append("canonical_uri = ?") 

72 params.append(canonical_uri) 

73 

74 where = ("WHERE " + " AND ".join(conditions)) if conditions else "" 

75 

76 with db() as conn: 

77 rows = conn.execute( 

78 f"SELECT raw_uri, canonical_uri, kind, created_at FROM entity_aliases" # nosec B608 — where is built from literal fragments; values in params 

79 f" {where} ORDER BY created_at DESC", 

80 params, 

81 ).fetchall() 

82 

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

84 

85 

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

87def delete_alias( 

88 raw_uri: str, 

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

90) -> None: 

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

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

93 raise HTTPException( 

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

95 ) 

96 

97 decoded = unquote(raw_uri) 

98 

99 with db() as conn: 

100 row = conn.execute( 

101 "SELECT kind FROM entity_aliases WHERE raw_uri = ?", (decoded,) 

102 ).fetchone() 

103 if row is None: 

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

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

106 raise HTTPException( 

107 status_code=status.HTTP_403_FORBIDDEN, 

108 detail=( 

109 "migration aliases are managed by the migration sweep " 

110 "and cannot be deleted via API" 

111 ), 

112 ) 

113 conn.execute("DELETE FROM entity_aliases WHERE raw_uri = ?", (decoded,))