Skip to content

Commit 624ea8f

Browse files
refactor: make specific data classes inherit from generic
1 parent dfc8c96 commit 624ea8f

4 files changed

Lines changed: 107 additions & 71 deletions

File tree

openedx_authz/api/data.py

Lines changed: 36 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -25,35 +25,32 @@ class PolicyIndex(Enum):
2525
# The rest of the fields are optional and can be ignored for now
2626

2727

28-
@define
29-
class UserData:
30-
"""A user is a subject that can be assigned roles and permissions.
31-
32-
Attributes:
33-
username: The username. Automatically prefixed with 'user:' if not present.
34-
"""
35-
36-
username: str
37-
38-
def __attrs_post_init__(self):
39-
"""Ensure username has 'user:' namespace prefix."""
40-
if not self.username.startswith("user:"):
41-
object.__setattr__(self, "username", f"user:{self.username}")
42-
43-
4428
@define
4529
class ScopeData:
4630
"""A scope is a context in which roles and permissions are assigned.
4731
4832
Attributes:
49-
scope_id: The scope identifier (e.g., 'course-v1:edX+DemoX+2021_T1').
33+
scope_id: The scope identifier (e.g., 'org:Demo').
5034
5135
This class assumes that the scope is already namespaced appropriately
5236
before being passed in, as scopes can vary widely (e.g., courses, organizations).
5337
"""
54-
# TODO: figure out namespace for scopes
5538
scope_id: str
5639

40+
@define
41+
class ContentLibraryData(ScopeData):
42+
"""A content library is a collection of content items.
43+
44+
Attributes:
45+
library_id: The content library identifier (e.g., 'library-v1:edX+DemoX+2021_T1').
46+
"""
47+
48+
library_id: str
49+
50+
def __attrs_post_init__(self):
51+
"""Ensure scope ID has 'lib:' namespace prefix."""
52+
if not self.scope_id.startswith("lib:"):
53+
self.scope_id = f"lib:{self.library_id}"
5754

5855
@define
5956
class SubjectData:
@@ -67,8 +64,26 @@ class SubjectData:
6764
users, groups, or other entities.
6865
"""
6966

70-
subject_id: str
67+
subject_id: str = ""
68+
69+
@define
70+
class UserData(SubjectData):
71+
"""A user is a subject that can be assigned roles and permissions.
7172
73+
Attributes:
74+
username: The username for the user (e.g., 'john_doe').
75+
76+
This class automatically adds the 'user:' namespace prefix to the subject ID.
77+
Can be initialized with either username= or subject_id= parameter.
78+
"""
79+
80+
username: str = ""
81+
82+
def __attrs_post_init__(self):
83+
"""Ensure subject ID has 'user:' namespace prefix."""
84+
# If username was provided, use it to set subject_id
85+
if not self.subject_id.startswith("user:"):
86+
self.subject_id = f"user:{self.username}"
7287

7388
@define
7489
class ActionData:
@@ -83,7 +98,7 @@ class ActionData:
8398
def __attrs_post_init__(self):
8499
"""Ensure action name has 'act:' namespace prefix."""
85100
if not self.action_id.startswith("act:"):
86-
object.__setattr__(self, "action_id", f"act:{self.action_id}")
101+
self.action_id = f"act:{self.action_id}"
87102

88103

89104
@define
@@ -132,7 +147,7 @@ class RoleData:
132147
def __attrs_post_init__(self):
133148
"""Ensure role name has 'role:' namespace prefix."""
134149
if not self.name.startswith("role:"):
135-
object.__setattr__(self, "name", f"role:{self.name}")
150+
self.name = f"role:{self.name}"
136151

137152

138153
@define

openedx_authz/api/roles.py

Lines changed: 34 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -33,9 +33,9 @@
3333
"batch_assign_role_to_subjects_in_scope",
3434
"unassign_role_from_subject_in_scope",
3535
"batch_unassign_role_from_subjects_in_scope",
36-
"get_role_assignments_for_subject_in_scope",
36+
"get_subject_role_assignments_in_scope",
3737
"get_role_assignments_for_role_in_scope",
38-
"get_role_assignments_for_subject",
38+
"get_subject_role_assignments",
3939
]
4040

4141
# TODO: these are the concerns we still have to address:
@@ -48,7 +48,7 @@
4848

4949

5050
def get_permissions_for_roles(
51-
role_names: list[str] | str,
51+
roles: list[RoleData] | RoleData,
5252
) -> dict[str, dict[str, list[PermissionData | str]]]:
5353
"""Get the permissions (actions) for a list of roles.
5454
@@ -59,18 +59,18 @@ def get_permissions_for_roles(
5959
dict[str, list[PermissionData]]: A dictionary mapping role names to their permissions and scopes.
6060
"""
6161
permissions_by_role = {}
62-
if not role_names:
62+
if not roles:
6363
return permissions_by_role
6464

65-
if isinstance(role_names, str):
66-
role_names = [role_names]
65+
if isinstance(roles, RoleData):
66+
roles = [roles]
6767

68-
for role_name in role_names:
69-
policies = enforcer.get_implicit_permissions_for_user(role_name)
68+
for role in roles:
69+
policies = enforcer.get_implicit_permissions_for_user(role.name)
7070

71-
assert role_name not in permissions_by_role, "Duplicate role names found"
71+
assert role.name not in permissions_by_role, "Duplicate role names found"
7272

73-
permissions_by_role[role_name] = {
73+
permissions_by_role[role.name] = {
7474
"permissions": [get_permission_from_policy(policy) for policy in policies],
7575
"scopes": list(set(policy[PolicyIndex.SCOPE.value] for policy in policies)),
7676
}
@@ -79,7 +79,7 @@ def get_permissions_for_roles(
7979

8080

8181
def get_permissions_for_active_roles_in_scope(
82-
scope: ScopeData, role_name: str = None
82+
scope: ScopeData, role: RoleData | None = None
8383
) -> dict[str, dict[str, list[PermissionData | str]]]:
8484
"""Retrieve all permissions granted by the specified roles within the given scope.
8585
@@ -110,15 +110,15 @@ def get_permissions_for_active_roles_in_scope(
110110
GroupingPolicyIndex.SCOPE.value, scope.scope_id
111111
)
112112

113-
if role_name:
113+
if role:
114114
filtered_policy = [
115115
policy
116116
for policy in filtered_policy
117-
if policy[GroupingPolicyIndex.ROLE.value] == role_name
117+
if policy[GroupingPolicyIndex.ROLE.value] == role.name
118118
]
119119

120120
return get_permissions_for_roles(
121-
[policy[GroupingPolicyIndex.ROLE.value] for policy in filtered_policy]
121+
[RoleData(name=policy[GroupingPolicyIndex.ROLE.value]) for policy in filtered_policy]
122122
)
123123

124124

@@ -180,7 +180,7 @@ def assign_role_to_subject_in_scope(
180180
role: The role to assign.
181181
"""
182182
assert (
183-
get_role_assignments_for_subject_in_scope(subject.subject_id, scope.scope_id)
183+
get_subject_role_assignments_in_scope(subject, scope)
184184
== []
185185
), "Subject already has a role in the scope"
186186

@@ -199,9 +199,7 @@ def batch_assign_role_to_subjects_in_scope(
199199
for subject in subjects:
200200

201201
assert (
202-
get_role_assignments_for_subject_in_scope(
203-
subject.subject_id, scope.scope_id
204-
)
202+
get_subject_role_assignments_in_scope(subject, scope)
205203
== []
206204
), "Subject already has a role in the scope"
207205

@@ -239,7 +237,7 @@ def batch_unassign_role_from_subjects_in_scope(
239237
enforcer.delete_roles_for_user_in_domain(subject, role.name, scope.scope_id)
240238

241239

242-
def get_role_assignments_for_subject(subject: SubjectData) -> list[RoleAssignmentData]:
240+
def get_subject_role_assignments(subject: SubjectData) -> list[RoleAssignmentData]:
243241
"""Get all the roles for a subject across all scopes.
244242
245243
Args:
@@ -250,20 +248,21 @@ def get_role_assignments_for_subject(subject: SubjectData) -> list[RoleAssignmen
250248
"""
251249
role_assignments = []
252250
for policy in enforcer.get_filtered_grouping_policy(
253-
GroupingPolicyIndex.SUBJECT.value, subject
251+
GroupingPolicyIndex.SUBJECT.value, subject.subject_id
254252
):
255253

256254
assert policy[GroupingPolicyIndex.ROLE.value] not in [
257255
role.role.name for role in role_assignments
258256
], "Duplicate role names found"
259257

260-
permissions = get_permissions_for_roles(policy[GroupingPolicyIndex.ROLE.value])[
261-
policy[GroupingPolicyIndex.ROLE.value]
258+
role_name = policy[GroupingPolicyIndex.ROLE.value]
259+
permissions = get_permissions_for_roles(RoleData(name=role_name))[
260+
role_name
262261
]["permissions"]
263262

264263
role_assignments.append(
265264
RoleAssignmentData(
266-
subject=SubjectData(subject_id=subject),
265+
subject=subject,
267266
role=RoleData(
268267
name=policy[GroupingPolicyIndex.ROLE.value],
269268
permissions=permissions,
@@ -274,8 +273,8 @@ def get_role_assignments_for_subject(subject: SubjectData) -> list[RoleAssignmen
274273
return role_assignments
275274

276275

277-
def get_role_assignments_for_subject_in_scope(
278-
subject: str, scope: str
276+
def get_subject_role_assignments_in_scope(
277+
subject: SubjectData, scope: ScopeData
279278
) -> list[RoleAssignmentData]:
280279
"""Get the roles for a subject in a specific scope.
281280
@@ -288,49 +287,49 @@ def get_role_assignments_for_subject_in_scope(
288287
"""
289288
# TODO: we still need to get the remaining data for the role like email, etc
290289
role_assignments = []
291-
for role_name in enforcer.get_roles_for_user_in_domain(subject, scope):
290+
for role_name in enforcer.get_roles_for_user_in_domain(subject.subject_id, scope.scope_id):
292291
role_assignments.append(
293292
RoleAssignmentData(
294-
subject=SubjectData(subject_id=subject),
293+
subject=subject,
295294
role=RoleData(
296295
name=role_name,
297-
permissions=get_permissions_for_roles(role_name)[role_name][
296+
permissions=get_permissions_for_roles(RoleData(name=role_name))[role_name][
298297
"permissions"
299298
],
300299
),
301-
scope=ScopeData(scope_id=scope),
300+
scope=scope,
302301
)
303302
)
304303
return role_assignments
305304

306305

307306
def get_role_assignments_for_role_in_scope(
308-
role_name: str, scope: str
307+
role: RoleData, scope: ScopeData
309308
) -> list[RoleAssignmentData]:
310309
"""Get the subjects assigned to a specific role in a specific scope.
311310
312311
Args:
313-
role_name: The name of the role.
312+
role: The role data.
314313
scope: The scope to filter subjects (e.g., 'library:123' or '*' for global).
315314
316315
Returns:
317316
list[RoleAssignment]: A list of subjects assigned to the specified role in the specified scope.
318317
"""
319318
role_assignments = []
320-
for subject in enforcer.get_users_for_role_in_domain(role_name, scope):
319+
for subject in enforcer.get_users_for_role_in_domain(role.name, scope.scope_id):
321320
if subject.startswith("role:"):
322321
# Skip roles that are also subjects
323322
continue
324323
role_assignments.append(
325324
RoleAssignmentData(
326325
subject=SubjectData(subject_id=subject),
327326
role=RoleData(
328-
name=role_name,
329-
permissions=get_permissions_for_roles(role_name)[role_name][
327+
name=role.name,
328+
permissions=get_permissions_for_roles(role)[role.name][
330329
"permissions"
331330
],
332331
),
333-
scope=ScopeData(scope_id=scope),
332+
scope=scope,
334333
)
335334
)
336335
return role_assignments

openedx_authz/api/users.py

Lines changed: 16 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -14,12 +14,22 @@
1414
assign_role_to_subject_in_scope,
1515
batch_assign_role_to_subjects_in_scope,
1616
batch_unassign_role_from_subjects_in_scope,
17-
get_role_assignments_for_subject,
18-
get_role_assignments_for_subject_in_scope,
17+
get_subject_role_assignments,
18+
get_subject_role_assignments_in_scope,
1919
unassign_role_from_subject_in_scope,
2020
)
2121

2222

23+
__all__ = [
24+
"assign_role_to_user_in_scope",
25+
"batch_assign_role_to_users",
26+
"unassign_role_from_user",
27+
"batch_unassign_role_from_users",
28+
"get_user_role_assignments",
29+
"get_user_role_assignments_in_scope",
30+
]
31+
32+
2333
def assign_role_to_user_in_scope(username: str, role_name: str, scope_id: str) -> bool:
2434
"""Assign a role to a user in a specific scope.
2535
@@ -94,7 +104,7 @@ def batch_unassign_role_from_users(
94104
)
95105

96106

97-
def get_role_assignments_for_user(username: str) -> list[dict]:
107+
def get_user_role_assignments(username: str) -> list[dict]:
98108
"""Get all roles for a user across all scopes.
99109
100110
Args:
@@ -103,10 +113,10 @@ def get_role_assignments_for_user(username: str) -> list[dict]:
103113
Returns:
104114
list[dict]: A list of role names and all their metadata assigned to the user.
105115
"""
106-
return get_role_assignments_for_subject(UserData(username=username))
116+
return get_subject_role_assignments(UserData(username=username))
107117

108118

109-
def get_role_assignments_for_user_in_scope(username: str, scope_id: str) -> list[str]:
119+
def get_user_role_assignments_in_scope(username: str, scope_id: str) -> list[str]:
110120
"""Get the roles assigned to a user in a specific scope.
111121
112122
Args:
@@ -116,6 +126,6 @@ def get_role_assignments_for_user_in_scope(username: str, scope_id: str) -> list
116126
Returns:
117127
list: A list of role names assigned to the user in the specified scope.
118128
"""
119-
return get_role_assignments_for_subject_in_scope(
129+
return get_subject_role_assignments_in_scope(
120130
UserData(username=username), ScopeData(scope_id=scope_id)
121131
)

0 commit comments

Comments
 (0)