Skip to content

Commit f3c6575

Browse files
swaroopAkkineniWorkosswaroopAkkineniWorkos
committed
FGA_5: list_resources_for_membership, list_memberships_for_resource, list_memberships_for_resource_by_external_id
1 parent 526eb7d commit f3c6575

File tree

2 files changed

+698
-1
lines changed

2 files changed

+698
-1
lines changed

‎src/workos/authorization.py‎

Lines changed: 354 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,13 +1,18 @@
1-
from typing import Any, Dict, Optional, Protocol, Sequence
1+
from functools import partial
2+
from typing import Any, Dict, Literal, Optional, Protocol, Sequence
23

34
from pydantic import TypeAdapter
45

56
from workos.types.authorization.environment_role import (
67
EnvironmentRole,
78
EnvironmentRoleList,
89
)
10+
from workos.types.authorization.organization_membership import (
11+
AuthorizationOrganizationMembership,
12+
)
913
from workos.types.authorization.organization_role import OrganizationRole
1014
from workos.types.authorization.permission import Permission
15+
from workos.types.authorization.resource import Resource
1116
from workos.types.authorization.role import Role, RoleList
1217
from workos.types.list_resource import (
1318
ListArgs,
@@ -41,6 +46,30 @@ class PermissionListFilters(ListArgs, total=False):
4146
]
4247

4348

49+
class ResourcesForMembershipListFilters(ListArgs, total=False):
50+
permission_slug: str
51+
parent_resource_id: Optional[str]
52+
parent_resource_type_slug: Optional[str]
53+
parent_resource_external_id: Optional[str]
54+
55+
56+
ResourcesForMembershipListResource = WorkOSListResource[
57+
Resource, ResourcesForMembershipListFilters, ListMetadata
58+
]
59+
60+
61+
class MembershipsForResourceListFilters(ListArgs, total=False):
62+
permission_slug: str
63+
assignment: Optional[Literal["direct", "indirect"]]
64+
65+
66+
MembershipsForResourceListResource = WorkOSListResource[
67+
AuthorizationOrganizationMembership,
68+
MembershipsForResourceListFilters,
69+
ListMetadata,
70+
]
71+
72+
4473
class AuthorizationModule(Protocol):
4574
"""Offers methods through the WorkOS Authorization service."""
4675

@@ -161,6 +190,48 @@ def add_environment_role_permission(
161190
permission_slug: str,
162191
) -> SyncOrAsync[EnvironmentRole]: ...
163192

193+
# Resource-Membership Relationships
194+
195+
def list_resources_for_membership(
196+
self,
197+
organization_membership_id: str,
198+
*,
199+
permission_slug: str,
200+
parent_resource_id: Optional[str] = None,
201+
parent_resource_type_slug: Optional[str] = None,
202+
parent_resource_external_id: Optional[str] = None,
203+
limit: int = DEFAULT_LIST_RESPONSE_LIMIT,
204+
before: Optional[str] = None,
205+
after: Optional[str] = None,
206+
order: PaginationOrder = "desc",
207+
) -> SyncOrAsync[ResourcesForMembershipListResource]: ...
208+
209+
def list_memberships_for_resource(
210+
self,
211+
resource_id: str,
212+
*,
213+
permission_slug: str,
214+
assignment: Optional[Literal["direct", "indirect"]] = None,
215+
limit: int = DEFAULT_LIST_RESPONSE_LIMIT,
216+
before: Optional[str] = None,
217+
after: Optional[str] = None,
218+
order: PaginationOrder = "desc",
219+
) -> SyncOrAsync[MembershipsForResourceListResource]: ...
220+
221+
def list_memberships_for_resource_by_external_id(
222+
self,
223+
organization_id: str,
224+
resource_type_slug: str,
225+
external_id: str,
226+
*,
227+
permission_slug: str,
228+
assignment: Optional[Literal["direct", "indirect"]] = None,
229+
limit: int = DEFAULT_LIST_RESPONSE_LIMIT,
230+
before: Optional[str] = None,
231+
after: Optional[str] = None,
232+
order: PaginationOrder = "desc",
233+
) -> SyncOrAsync[MembershipsForResourceListResource]: ...
234+
164235

165236
class Authorization(AuthorizationModule):
166237
_http_client: SyncHTTPClient
@@ -437,6 +508,147 @@ def add_environment_role_permission(
437508

438509
return EnvironmentRole.model_validate(response)
439510

511+
# Resource-Membership Relationships
512+
513+
def list_resources_for_membership(
514+
self,
515+
organization_membership_id: str,
516+
*,
517+
permission_slug: str,
518+
parent_resource_id: Optional[str] = None,
519+
parent_resource_type_slug: Optional[str] = None,
520+
parent_resource_external_id: Optional[str] = None,
521+
limit: int = DEFAULT_LIST_RESPONSE_LIMIT,
522+
before: Optional[str] = None,
523+
after: Optional[str] = None,
524+
order: PaginationOrder = "desc",
525+
) -> ResourcesForMembershipListResource:
526+
if parent_resource_id is not None and (
527+
parent_resource_type_slug is not None
528+
or parent_resource_external_id is not None
529+
):
530+
raise ValueError(
531+
"Cannot specify both parent_resource_id and "
532+
"parent_resource_type_slug/parent_resource_external_id. "
533+
"Use one identification method."
534+
)
535+
if (parent_resource_type_slug is None) != (parent_resource_external_id is None):
536+
raise ValueError(
537+
"parent_resource_type_slug and parent_resource_external_id "
538+
"must be provided together."
539+
)
540+
541+
list_params: ResourcesForMembershipListFilters = {
542+
"limit": limit,
543+
"before": before,
544+
"after": after,
545+
"order": order,
546+
"permission_slug": permission_slug,
547+
}
548+
if parent_resource_id is not None:
549+
list_params["parent_resource_id"] = parent_resource_id
550+
if parent_resource_type_slug is not None:
551+
list_params["parent_resource_type_slug"] = parent_resource_type_slug
552+
if parent_resource_external_id is not None:
553+
list_params["parent_resource_external_id"] = parent_resource_external_id
554+
555+
response = self._http_client.request(
556+
f"authorization/organization_memberships/{organization_membership_id}/resources",
557+
method=REQUEST_METHOD_GET,
558+
params=list_params,
559+
)
560+
561+
return WorkOSListResource[
562+
Resource, ResourcesForMembershipListFilters, ListMetadata
563+
](
564+
list_method=partial(
565+
self.list_resources_for_membership, organization_membership_id
566+
),
567+
list_args=list_params,
568+
**ListPage[Resource](**response).model_dump(),
569+
)
570+
571+
def list_memberships_for_resource(
572+
self,
573+
resource_id: str,
574+
*,
575+
permission_slug: str,
576+
assignment: Optional[Literal["direct", "indirect"]] = None,
577+
limit: int = DEFAULT_LIST_RESPONSE_LIMIT,
578+
before: Optional[str] = None,
579+
after: Optional[str] = None,
580+
order: PaginationOrder = "desc",
581+
) -> MembershipsForResourceListResource:
582+
list_params: MembershipsForResourceListFilters = {
583+
"limit": limit,
584+
"before": before,
585+
"after": after,
586+
"order": order,
587+
"permission_slug": permission_slug,
588+
}
589+
if assignment is not None:
590+
list_params["assignment"] = assignment
591+
592+
response = self._http_client.request(
593+
f"authorization/resources/{resource_id}/organization_memberships",
594+
method=REQUEST_METHOD_GET,
595+
params=list_params,
596+
)
597+
598+
return WorkOSListResource[
599+
AuthorizationOrganizationMembership,
600+
MembershipsForResourceListFilters,
601+
ListMetadata,
602+
](
603+
list_method=partial(self.list_memberships_for_resource, resource_id),
604+
list_args=list_params,
605+
**ListPage[AuthorizationOrganizationMembership](**response).model_dump(),
606+
)
607+
608+
def list_memberships_for_resource_by_external_id(
609+
self,
610+
organization_id: str,
611+
resource_type_slug: str,
612+
external_id: str,
613+
*,
614+
permission_slug: str,
615+
assignment: Optional[Literal["direct", "indirect"]] = None,
616+
limit: int = DEFAULT_LIST_RESPONSE_LIMIT,
617+
before: Optional[str] = None,
618+
after: Optional[str] = None,
619+
order: PaginationOrder = "desc",
620+
) -> MembershipsForResourceListResource:
621+
list_params: MembershipsForResourceListFilters = {
622+
"limit": limit,
623+
"before": before,
624+
"after": after,
625+
"order": order,
626+
"permission_slug": permission_slug,
627+
}
628+
if assignment is not None:
629+
list_params["assignment"] = assignment
630+
631+
response = self._http_client.request(
632+
f"authorization/organizations/{organization_id}/resources/{resource_type_slug}/{external_id}/organization_memberships",
633+
method=REQUEST_METHOD_GET,
634+
params=list_params,
635+
)
636+
637+
return WorkOSListResource[
638+
AuthorizationOrganizationMembership,
639+
MembershipsForResourceListFilters,
640+
ListMetadata,
641+
](
642+
list_method=partial(
643+
self.list_memberships_for_resource_by_external_id,
644+
organization_id,
645+
resource_type_slug,
646+
external_id,
647+
),
648+
list_args=list_params,
649+
**ListPage[AuthorizationOrganizationMembership](**response).model_dump(),
650+
)
651+
440652

441653
class AsyncAuthorization(AuthorizationModule):
442654
_http_client: AsyncHTTPClient
@@ -712,3 +924,144 @@ async def add_environment_role_permission(
712924
)
713925

714926
return EnvironmentRole.model_validate(response)
927+
928+
# Resource-Membership Relationships
929+
930+
async def list_resources_for_membership(
931+
self,
932+
organization_membership_id: str,
933+
*,
934+
permission_slug: str,
935+
parent_resource_id: Optional[str] = None,
936+
parent_resource_type_slug: Optional[str] = None,
937+
parent_resource_external_id: Optional[str] = None,
938+
limit: int = DEFAULT_LIST_RESPONSE_LIMIT,
939+
before: Optional[str] = None,
940+
after: Optional[str] = None,
941+
order: PaginationOrder = "desc",
942+
) -> ResourcesForMembershipListResource:
943+
if parent_resource_id is not None and (
944+
parent_resource_type_slug is not None
945+
or parent_resource_external_id is not None
946+
):
947+
raise ValueError(
948+
"Cannot specify both parent_resource_id and "
949+
"parent_resource_type_slug/parent_resource_external_id. "
950+
"Use one identification method."
951+
)
952+
if (parent_resource_type_slug is None) != (parent_resource_external_id is None):
953+
raise ValueError(
954+
"parent_resource_type_slug and parent_resource_external_id "
955+
"must be provided together."
956+
)
957+
958+
list_params: ResourcesForMembershipListFilters = {
959+
"limit": limit,
960+
"before": before,
961+
"after": after,
962+
"order": order,
963+
"permission_slug": permission_slug,
964+
}
965+
if parent_resource_id is not None:
966+
list_params["parent_resource_id"] = parent_resource_id
967+
if parent_resource_type_slug is not None:
968+
list_params["parent_resource_type_slug"] = parent_resource_type_slug
969+
if parent_resource_external_id is not None:
970+
list_params["parent_resource_external_id"] = parent_resource_external_id
971+
972+
response = await self._http_client.request(
973+
f"authorization/organization_memberships/{organization_membership_id}/resources",
974+
method=REQUEST_METHOD_GET,
975+
params=list_params,
976+
)
977+
978+
return WorkOSListResource[
979+
Resource, ResourcesForMembershipListFilters, ListMetadata
980+
](
981+
list_method=partial(
982+
self.list_resources_for_membership, organization_membership_id
983+
),
984+
list_args=list_params,
985+
**ListPage[Resource](**response).model_dump(),
986+
)
987+
988+
async def list_memberships_for_resource(
989+
self,
990+
resource_id: str,
991+
*,
992+
permission_slug: str,
993+
assignment: Optional[Literal["direct", "indirect"]] = None,
994+
limit: int = DEFAULT_LIST_RESPONSE_LIMIT,
995+
before: Optional[str] = None,
996+
after: Optional[str] = None,
997+
order: PaginationOrder = "desc",
998+
) -> MembershipsForResourceListResource:
999+
list_params: MembershipsForResourceListFilters = {
1000+
"limit": limit,
1001+
"before": before,
1002+
"after": after,
1003+
"order": order,
1004+
"permission_slug": permission_slug,
1005+
}
1006+
if assignment is not None:
1007+
list_params["assignment"] = assignment
1008+
1009+
response = await self._http_client.request(
1010+
f"authorization/resources/{resource_id}/organization_memberships",
1011+
method=REQUEST_METHOD_GET,
1012+
params=list_params,
1013+
)
1014+
1015+
return WorkOSListResource[
1016+
AuthorizationOrganizationMembership,
1017+
MembershipsForResourceListFilters,
1018+
ListMetadata,
1019+
](
1020+
list_method=partial(self.list_memberships_for_resource, resource_id),
1021+
list_args=list_params,
1022+
**ListPage[AuthorizationOrganizationMembership](**response).model_dump(),
1023+
)
1024+
1025+
async def list_memberships_for_resource_by_external_id(
1026+
self,
1027+
organization_id: str,
1028+
resource_type_slug: str,
1029+
external_id: str,
1030+
*,
1031+
permission_slug: str,
1032+
assignment: Optional[Literal["direct", "indirect"]] = None,
1033+
limit: int = DEFAULT_LIST_RESPONSE_LIMIT,
1034+
before: Optional[str] = None,
1035+
after: Optional[str] = None,
1036+
order: PaginationOrder = "desc",
1037+
) -> MembershipsForResourceListResource:
1038+
list_params: MembershipsForResourceListFilters = {
1039+
"limit": limit,
1040+
"before": before,
1041+
"after": after,
1042+
"order": order,
1043+
"permission_slug": permission_slug,
1044+
}
1045+
if assignment is not None:
1046+
list_params["assignment"] = assignment
1047+
1048+
response = await self._http_client.request(
1049+
f"authorization/organizations/{organization_id}/resources/{resource_type_slug}/{external_id}/organization_memberships",
1050+
method=REQUEST_METHOD_GET,
1051+
params=list_params,
1052+
)
1053+
1054+
return WorkOSListResource[
1055+
AuthorizationOrganizationMembership,
1056+
MembershipsForResourceListFilters,
1057+
ListMetadata,
1058+
](
1059+
list_method=partial(
1060+
self.list_memberships_for_resource_by_external_id,
1061+
organization_id,
1062+
resource_type_slug,
1063+
external_id,
1064+
),
1065+
list_args=list_params,
1066+
**ListPage[AuthorizationOrganizationMembership](**response).model_dump(),
1067+
)

0 commit comments

Comments
 (0)