Coverage for src/lexigram/admin/rbac/roles_sql.py: 89%

1"""SQL-backed implementation of AdminRoleStoreProtocol. 

2 

3Owns all DDL and DML for the ``admin_roles`` table. The service layer 

4depends only on ``AdminRoleStoreProtocol`` — never on this class directly. 

5Permissions and inheritance are stored as JSON text for database-portable 

6array semantics. 

7""" 

8 

9from __future__ import annotations 

10 

11from typing import Any 

12 

13from lexigram.admin.rbac.protocols import AdminRoleStoreProtocol 

14from lexigram.admin.sql_dialect import is_postgres, now_expr 

15from lexigram.contracts.auth import RoleDefinition 

16from lexigram.contracts.data import DatabaseProviderProtocol 

17from lexigram.di.decorators import inject 

18from lexigram.logging import get_logger 

19from lexigram.serialization import dumps_str, loads 

20 

21logger = get_logger(__name__) 

22 

23_TABLE = "admin_roles" 

24 

25 

26def _load_list(value: Any) -> list[str]: 

27 """Parse a stored JSON text column into a list of strings.""" 

28 if value is None: 

29 return [] 

30 if isinstance(value, list): 

31 return [str(v) for v in value] 

32 try: 

33 parsed = loads(str(value)) 

34 except ValueError: 

35 return [] 

36 return [str(v) for v in parsed] if isinstance(parsed, list) else [] 

37 

38 

39@inject 

40class AdminRoleSqlStore(AdminRoleStoreProtocol): 

41 """SQL store for admin roles. 

42 

43 Implements ``AdminRoleStoreProtocol``. Manages the ``admin_roles`` 

44 table including DDL bootstrap. Role names are the primary key; rows 

45 mirror the shape ``AuthorizationService.sync_from_db`` expects from 

46 its legacy ``admin_roles`` fallback table (``name``, ``description``, 

47 ``permissions``, ``inherits``, plus ``is_system`` for UI protection). 

48 """ 

49 

50 def __init__(self, db: DatabaseProviderProtocol) -> None: 

51 """Initialise with a resolved database provider. 

52 

53 Args: 

54 db: Framework database provider exposing ``execute`` and 

55 ``execute_query``. 

56 """ 

57 self._db = db 

58 self._initialized = False 

59 

60 async def ensure_schema(self) -> None: 

61 """Create the roles table if it does not exist (idempotent).""" 

62 if self._initialized: 

63 return 

64 if is_postgres(self._db): 

65 create_sql = f""" 

66 CREATE TABLE IF NOT EXISTS {_TABLE} ( 

67 name VARCHAR(100) PRIMARY KEY, 

68 description TEXT NOT NULL DEFAULT '', 

69 permissions TEXT NOT NULL DEFAULT '[]', 

70 inherits TEXT NOT NULL DEFAULT '[]', 

71 is_system BOOLEAN NOT NULL DEFAULT FALSE, 

72 created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), 

73 updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() 

74 ) 

75 """ 

76 else: 

77 create_sql = f""" 

78 CREATE TABLE IF NOT EXISTS {_TABLE} ( 

79 name VARCHAR(100) PRIMARY KEY, 

80 description TEXT NOT NULL DEFAULT '', 

81 permissions TEXT NOT NULL DEFAULT '[]', 

82 inherits TEXT NOT NULL DEFAULT '[]', 

83 is_system BOOLEAN NOT NULL DEFAULT FALSE, 

84 created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, 

85 updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP 

86 ) 

87 """ 

88 await self._db.execute(create_sql, []) 

89 self._initialized = True 

90 

91 async def list_roles(self) -> list[RoleDefinition]: 

92 """Return all roles ordered by name (see protocol docs).""" 

93 result = await self._db.execute_query( 

94 f"SELECT name, description, permissions, inherits, is_system FROM {_TABLE} ORDER BY name", # noqa: S608 — table name is module constant "admin_roles", never user input 

95 [], 

96 ) 

97 return [self._row_to_role(row) for row in self._rows(result)] 

98 

99 async def get_role(self, name: str) -> RoleDefinition | None: 

100 """Look up a role by name (see protocol docs).""" 

101 result = await self._db.execute_query( 

102 f"SELECT name, description, permissions, inherits, is_system FROM {_TABLE} WHERE name = ?", # noqa: S608 — table name is module constant "admin_roles", never user input 

103 [name], 

104 ) 

105 rows = self._rows(result) 

106 return self._row_to_role(rows[0]) if rows else None 

107 

108 async def create_role(self, role: RoleDefinition) -> None: 

109 """Insert a new role (see protocol docs).""" 

110 await self._db.execute( 

111 f"INSERT INTO {_TABLE} (name, description, permissions, inherits, is_system) VALUES (?, ?, ?, ?, ?)", # noqa: S608 — table name is module constant "admin_roles", never user input 

112 [ 

113 role.name, 

114 role.description, 

115 dumps_str(role.permissions), 

116 dumps_str(role.inherits), 

117 role.is_system, 

118 ], 

119 ) 

120 

121 async def update_role(self, role: RoleDefinition) -> None: 

122 """Update an existing role by name (see protocol docs).""" 

123 await self._db.execute( 

124 f"UPDATE {_TABLE} SET description = ?, permissions = ?, inherits = ?, is_system = ?, " # noqa: S608 — table name is module constant "admin_roles", never user input 

125 f"updated_at = {now_expr(self._db)} WHERE name = ?", 

126 [ 

127 role.description, 

128 dumps_str(role.permissions), 

129 dumps_str(role.inherits), 

130 role.is_system, 

131 role.name, 

132 ], 

133 ) 

134 

135 async def delete_role(self, name: str) -> bool: 

136 """Delete a role by name; ``True`` when a row was removed.""" 

137 result = await self._db.execute( 

138 f"DELETE FROM {_TABLE} WHERE name = ?", # noqa: S608 — table name is module constant "admin_roles", never user input 

139 [name], 

140 ) 

141 row_count = getattr(result, "row_count", None) 

142 if row_count is not None: 

143 return int(row_count) > 0 

144 return True 

145 

146 @staticmethod 

147 def _rows(result: Any) -> list[dict[str, Any]]: 

148 """Normalize execute_query results (object/.rows, list, or dict).""" 

149 if hasattr(result, "rows") and result.rows: 

150 return list(result.rows) 

151 if isinstance(result, list): 

152 return result 

153 if isinstance(result, dict): 

154 return [result] 

155 return [] 

156 

157 @staticmethod 

158 def _row_to_role(row: dict[str, Any]) -> RoleDefinition: 

159 """Build an RoleDefinition from a provider row.""" 

160 return RoleDefinition( 

161 name=str(row.get("name", "")), 

162 description=str(row.get("description", "")), 

163 permissions=_load_list(row.get("permissions")), 

164 inherits=_load_list(row.get("inherits")), 

165 is_system=bool(row.get("is_system", False)), 

166 ) 

167 

168 

169__all__ = ["AdminRoleSqlStore"]