Skip to content

FGA_5: list_resources_for_membership, list_memberships_for_resource, list_memberships_for_resource_by_external_id #571

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
swaroopAkkineniWorkos wants to merge 13 commits into
base: ENT-5224-python-sdk-for-fga-worktree-fuck-around
Choose a base branch
from
+1,223 −13
Open

FGA_5: list_resources_for_membership, list_memberships_for_resource, list_memberships_for_resource_by_external_id #571

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
f3c6575
FGA_5: list_resources_for_membership, list_memberships_for_resource, …
swaroopAkkineniWorkos Feb 20, 2026
43bae56
lol
swaroopAkkineniWorkos Feb 23, 2026
d66d6ec
minor nit
swaroopAkkineniWorkos Feb 23, 2026
9149f71
refactor
swaroopAkkineniWorkos Feb 23, 2026
e9a6975
moar
swaroopAkkineniWorkos Feb 23, 2026
4a949d6
enough
swaroopAkkineniWorkos Mar 2, 2026
700d915
lol
swaroopAkkineniWorkos Mar 2, 2026
3412ff0
Merge branch 'ENT-5224-python-sdk-for-fga-worktree-fuck-around' into …
swaroopAkkineniWorkos Mar 2, 2026
e4c8031
final
swaroopAkkineniWorkos Mar 2, 2026
47e2c77
final
swaroopAkkineniWorkos Mar 2, 2026
7453f18
lol
swaroopAkkineniWorkos Mar 2, 2026
adcafb5
lol
swaroopAkkineniWorkos Mar 2, 2026
a973327
lol
swaroopAkkineniWorkos Mar 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
318 changes: 305 additions & 13 deletions src/workos/authorization.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,15 +1,21 @@
from enum import Enum
from typing import Any, Dict, Optional, Protocol, Sequence, Union
from functools import partial
from typing import Any, Dict, Literal, Optional, Protocol, Sequence, Union

from pydantic import TypeAdapter
from typing_extensions import TypedDict

from workos.types.authorization.access_check_response import AccessCheckResponse
from workos.types.authorization.environment_role import (
EnvironmentRole,
EnvironmentRoleList,
)
from workos.types.authorization.organization_membership import (
AuthorizationOrganizationMembership,
)
from workos.types.authorization.organization_role import OrganizationRole
from workos.types.authorization.parent_resource_identifier import (
ParentResourceIdentifier,
)
from workos.types.authorization.permission import Permission
from workos.types.authorization.resource_identifier import ResourceIdentifier
from workos.types.authorization.authorization_resource import AuthorizationResource
Expand Down Expand Up @@ -42,6 +48,7 @@ class _Unset(Enum):
AUTHORIZATION_PERMISSIONS_PATH = "authorization/permissions"
AUTHORIZATION_RESOURCES_PATH = "authorization/resources"
AUTHORIZATION_ORGANIZATIONS_PATH = "authorization/organizations"
AUTHORIZATION_ORGANIZATION_MEMBERSHIPS_PATH = "authorization/organization_memberships"


class ResourceListFilters(ListArgs, total=False):
Expand All @@ -58,16 +65,25 @@ class ResourceListFilters(ListArgs, total=False):
]


class ParentResourceById(TypedDict):
parent_resource_id: str
class ResourcesForMembershipListFilters(ListArgs, total=False):
permission_slug: str


AuthorizationResourcesForMembershipList = WorkOSListResource[
AuthorizationResource, ResourcesForMembershipListFilters, ListMetadata
]


class ParentResourceByExternalId(TypedDict):
parent_resource_external_id: str
parent_resource_type_slug: str
class AuthorizationOrganizationMembershipListFilters(ListArgs, total=False):
permission_slug: str
assignment: Optional[Literal["direct", "indirect"]]


ParentResource = Union[ParentResourceById, ParentResourceByExternalId]
AuthorizationOrganizationMembershipList = WorkOSListResource[
AuthorizationOrganizationMembership,
AuthorizationOrganizationMembershipListFilters,
ListMetadata,
]

_role_adapter: TypeAdapter[Role] = TypeAdapter(Role)

Expand Down Expand Up @@ -213,7 +229,7 @@ def create_resource(
description: Optional[str] = None,
resource_type_slug: str,
organization_id: str,
parent: Optional[ParentResource] = None,
parent: Optional[ParentResourceIdentifier] = None,
) -> SyncOrAsync[AuthorizationResource]: ...

def update_resource(
Expand Down Expand Up @@ -280,6 +296,44 @@ def check(
resource: ResourceIdentifier,
) -> SyncOrAsync[AccessCheckResponse]: ...

def list_resources_for_membership(
self,
organization_membership_id: str,
*,
permission_slug: str,
parent_resource: ParentResourceIdentifier,
limit: int = DEFAULT_LIST_RESPONSE_LIMIT,
before: Optional[str] = None,
after: Optional[str] = None,
order: PaginationOrder = "desc",
) -> SyncOrAsync[AuthorizationResourcesForMembershipList]: ...

def list_memberships_for_resource(
self,
resource_id: str,
*,
permission_slug: str,
assignment: Optional[Literal["direct", "indirect"]] = None,
limit: int = DEFAULT_LIST_RESPONSE_LIMIT,
before: Optional[str] = None,
after: Optional[str] = None,
order: PaginationOrder = "desc",
) -> SyncOrAsync[AuthorizationOrganizationMembershipList]: ...

def list_memberships_for_resource_by_external_id(
self,
organization_id: str,
resource_type_slug: str,
external_id: str,
*,
permission_slug: str,
assignment: Optional[Literal["direct", "indirect"]] = None,
limit: int = DEFAULT_LIST_RESPONSE_LIMIT,
before: Optional[str] = None,
after: Optional[str] = None,
order: PaginationOrder = "desc",
) -> SyncOrAsync[AuthorizationOrganizationMembershipList]: ...


class Authorization(AuthorizationModule):
_http_client: SyncHTTPClient
Expand Down Expand Up @@ -572,7 +626,7 @@ def create_resource(
description: Optional[str] = None,
resource_type_slug: str,
organization_id: str,
parent: Optional[ParentResource] = None,
parent: Optional[ParentResourceIdentifier] = None,
) -> AuthorizationResource:
json: Dict[str, Any] = {
"resource_type_slug": resource_type_slug,
Expand Down Expand Up @@ -747,13 +801,132 @@ def check(
json.update(resource)

response = self._http_client.request(
f"authorization/organization_memberships/{organization_membership_id}/check",
f"{AUTHORIZATION_ORGANIZATION_MEMBERSHIPS_PATH}/{organization_membership_id}/check",
method=REQUEST_METHOD_POST,
json=json,
)

return AccessCheckResponse.model_validate(response)

def list_resources_for_membership(
self,
organization_membership_id: str,
*,
permission_slug: str,
parent_resource: ParentResourceIdentifier,
limit: int = DEFAULT_LIST_RESPONSE_LIMIT,
before: Optional[str] = None,
after: Optional[str] = None,
order: PaginationOrder = "desc",
) -> AuthorizationResourcesForMembershipList:
list_params: ResourcesForMembershipListFilters = {
"limit": limit,
"before": before,
"after": after,
"order": order,
"permission_slug": permission_slug,
}

http_params: Dict[str, Any] = {**list_params}
http_params.update(parent_resource)

response = self._http_client.request(
f"{AUTHORIZATION_ORGANIZATION_MEMBERSHIPS_PATH}/{organization_membership_id}/resources",
method=REQUEST_METHOD_GET,
params=http_params,
)

return AuthorizationResourcesForMembershipList(
list_method=partial(
self.list_resources_for_membership,
organization_membership_id,
parent_resource=parent_resource,
),
list_args=list_params,
**ListPage[AuthorizationResource](**response).model_dump(),
)

def list_memberships_for_resource(
self,
resource_id: str,
*,
permission_slug: str,
assignment: Optional[Literal["direct", "indirect"]] = None,
limit: int = DEFAULT_LIST_RESPONSE_LIMIT,
before: Optional[str] = None,
after: Optional[str] = None,
order: PaginationOrder = "desc",
) -> AuthorizationOrganizationMembershipList:
list_params: AuthorizationOrganizationMembershipListFilters = {
"limit": limit,
"before": before,
"after": after,
"order": order,
"permission_slug": permission_slug,
}
if assignment is not None:
list_params["assignment"] = assignment

response = self._http_client.request(
f"{AUTHORIZATION_RESOURCES_PATH}/{resource_id}/organization_memberships",
method=REQUEST_METHOD_GET,
params=list_params,
)

return WorkOSListResource[
AuthorizationOrganizationMembership,
AuthorizationOrganizationMembershipListFilters,
ListMetadata,
](
list_method=partial(self.list_memberships_for_resource, resource_id),
list_args=list_params,
**ListPage[AuthorizationOrganizationMembership](**response).model_dump(),
)

def list_memberships_for_resource_by_external_id(
self,
organization_id: str,
resource_type_slug: str,
external_id: str,
*,
permission_slug: str,
assignment: Optional[Literal["direct", "indirect"]] = None,
limit: int = DEFAULT_LIST_RESPONSE_LIMIT,
before: Optional[str] = None,
after: Optional[str] = None,
order: PaginationOrder = "desc",
) -> AuthorizationOrganizationMembershipList:
list_params: AuthorizationOrganizationMembershipListFilters = {
"limit": limit,
"before": before,
"after": after,
"order": order,
"permission_slug": permission_slug,
}
if assignment is not None:
list_params["assignment"] = assignment

response = self._http_client.request(
f"{AUTHORIZATION_ORGANIZATIONS_PATH}/{organization_id}/resources/{resource_type_slug}/{external_id}/organization_memberships",
method=REQUEST_METHOD_GET,
params=list_params,
)

return WorkOSListResource[
AuthorizationOrganizationMembership,
AuthorizationOrganizationMembershipListFilters,
ListMetadata,
](
list_method=partial(
self.list_memberships_for_resource_by_external_id,
organization_id,
resource_type_slug,
external_id,
),
list_args=list_params,
**ListPage[AuthorizationOrganizationMembership](**response).model_dump(),
)


class AsyncAuthorization(AuthorizationModule):
_http_client: AsyncHTTPClient
Expand Down Expand Up @@ -1048,7 +1221,7 @@ async def create_resource(
description: Optional[str] = None,
resource_type_slug: str,
organization_id: str,
parent: Optional[ParentResource] = None,
parent: Optional[ParentResourceIdentifier] = None,
) -> AuthorizationResource:
json: Dict[str, Any] = {
"resource_type_slug": resource_type_slug,
Expand Down Expand Up @@ -1223,9 +1396,128 @@ async def check(
json.update(resource)

response = await self._http_client.request(
f"authorization/organization_memberships/{organization_membership_id}/check",
f"{AUTHORIZATION_ORGANIZATION_MEMBERSHIPS_PATH}/{organization_membership_id}/check",
method=REQUEST_METHOD_POST,
json=json,
)

return AccessCheckResponse.model_validate(response)

async def list_resources_for_membership(
self,
organization_membership_id: str,
*,
permission_slug: str,
parent_resource: ParentResourceIdentifier,
limit: int = DEFAULT_LIST_RESPONSE_LIMIT,
before: Optional[str] = None,
after: Optional[str] = None,
order: PaginationOrder = "desc",
) -> AuthorizationResourcesForMembershipList:
list_params: ResourcesForMembershipListFilters = {
"limit": limit,
"before": before,
"after": after,
"order": order,
"permission_slug": permission_slug,
}

http_params: Dict[str, Any] = {**list_params}
http_params.update(parent_resource)

response = await self._http_client.request(
f"{AUTHORIZATION_ORGANIZATION_MEMBERSHIPS_PATH}/{organization_membership_id}/resources",
method=REQUEST_METHOD_GET,
params=http_params,
)

return AuthorizationResourcesForMembershipList(
list_method=partial(
self.list_resources_for_membership,
organization_membership_id,
parent_resource=parent_resource,
),
list_args=list_params,
**ListPage[AuthorizationResource](**response).model_dump(),
)

async def list_memberships_for_resource(
self,
resource_id: str,
*,
permission_slug: str,
assignment: Optional[Literal["direct", "indirect"]] = None,
limit: int = DEFAULT_LIST_RESPONSE_LIMIT,
before: Optional[str] = None,
after: Optional[str] = None,
order: PaginationOrder = "desc",
) -> AuthorizationOrganizationMembershipList:
list_params: AuthorizationOrganizationMembershipListFilters = {
"limit": limit,
"before": before,
"after": after,
"order": order,
"permission_slug": permission_slug,
}
if assignment is not None:
list_params["assignment"] = assignment

response = await self._http_client.request(
f"{AUTHORIZATION_RESOURCES_PATH}/{resource_id}/organization_memberships",
method=REQUEST_METHOD_GET,
params=list_params,
)

return WorkOSListResource[
AuthorizationOrganizationMembership,
AuthorizationOrganizationMembershipListFilters,
ListMetadata,
](
list_method=partial(self.list_memberships_for_resource, resource_id),
list_args=list_params,
**ListPage[AuthorizationOrganizationMembership](**response).model_dump(),
)

async def list_memberships_for_resource_by_external_id(
self,
organization_id: str,
resource_type_slug: str,
external_id: str,
*,
permission_slug: str,
assignment: Optional[Literal["direct", "indirect"]] = None,
limit: int = DEFAULT_LIST_RESPONSE_LIMIT,
before: Optional[str] = None,
after: Optional[str] = None,
order: PaginationOrder = "desc",
) -> AuthorizationOrganizationMembershipList:
list_params: AuthorizationOrganizationMembershipListFilters = {
"limit": limit,
"before": before,
"after": after,
"order": order,
"permission_slug": permission_slug,
}
if assignment is not None:
list_params["assignment"] = assignment

response = await self._http_client.request(
f"{AUTHORIZATION_ORGANIZATIONS_PATH}/{organization_id}/resources/{resource_type_slug}/{external_id}/organization_memberships",
method=REQUEST_METHOD_GET,
params=list_params,
)

return WorkOSListResource[
AuthorizationOrganizationMembership,
AuthorizationOrganizationMembershipListFilters,
ListMetadata,
](
list_method=partial(
self.list_memberships_for_resource_by_external_id,
organization_id,
resource_type_slug,
external_id,
),
list_args=list_params,
**ListPage[AuthorizationOrganizationMembership](**response).model_dump(),
)
Loading