From f3c657517a117cd22d433f74accd68ad9da5d1b5 Mon Sep 17 00:00:00 2001 From: swaroopakkineni Date: Fri, 20 Feb 2026 11:41:38 -1000 Subject: [PATCH 1/5] FGA_5: list_resources_for_membership, list_memberships_for_resource, list_memberships_for_resource_by_external_id --- src/workos/authorization.py | 355 +++++++++++++++++- ...test_authorization_resource_memberships.py | 344 +++++++++++++++++ 2 files changed, 698 insertions(+), 1 deletion(-) create mode 100644 tests/test_authorization_resource_memberships.py diff --git a/src/workos/authorization.py b/src/workos/authorization.py index 6e12f035..3e886d42 100644 --- a/src/workos/authorization.py +++ b/src/workos/authorization.py @@ -1,4 +1,5 @@ -from typing import Any, Dict, Optional, Protocol, Sequence +from functools import partial +from typing import Any, Dict, Literal, Optional, Protocol, Sequence from pydantic import TypeAdapter @@ -6,8 +7,12 @@ EnvironmentRole, EnvironmentRoleList, ) +from workos.types.authorization.organization_membership import ( + AuthorizationOrganizationMembership, +) from workos.types.authorization.organization_role import OrganizationRole from workos.types.authorization.permission import Permission +from workos.types.authorization.resource import Resource from workos.types.authorization.role import Role, RoleList from workos.types.list_resource import ( ListArgs, @@ -41,6 +46,30 @@ class PermissionListFilters(ListArgs, total=False): ] +class ResourcesForMembershipListFilters(ListArgs, total=False): + permission_slug: str + parent_resource_id: Optional[str] + parent_resource_type_slug: Optional[str] + parent_resource_external_id: Optional[str] + + +ResourcesForMembershipListResource = WorkOSListResource[ + Resource, ResourcesForMembershipListFilters, ListMetadata +] + + +class MembershipsForResourceListFilters(ListArgs, total=False): + permission_slug: str + assignment: Optional[Literal["direct", "indirect"]] + + +MembershipsForResourceListResource = WorkOSListResource[ + AuthorizationOrganizationMembership, + MembershipsForResourceListFilters, + ListMetadata, +] + + class AuthorizationModule(Protocol): """Offers methods through the WorkOS Authorization service.""" @@ -161,6 +190,48 @@ def add_environment_role_permission( permission_slug: str, ) -> SyncOrAsync[EnvironmentRole]: ... + # Resource-Membership Relationships + + def list_resources_for_membership( + self, + organization_membership_id: str, + *, + permission_slug: str, + parent_resource_id: Optional[str] = None, + parent_resource_type_slug: Optional[str] = None, + parent_resource_external_id: Optional[str] = None, + limit: int = DEFAULT_LIST_RESPONSE_LIMIT, + before: Optional[str] = None, + after: Optional[str] = None, + order: PaginationOrder = "desc", + ) -> SyncOrAsync[ResourcesForMembershipListResource]: ... + + 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[MembershipsForResourceListResource]: ... + + 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[MembershipsForResourceListResource]: ... + class Authorization(AuthorizationModule): _http_client: SyncHTTPClient @@ -437,6 +508,147 @@ def add_environment_role_permission( return EnvironmentRole.model_validate(response) + # Resource-Membership Relationships + + def list_resources_for_membership( + self, + organization_membership_id: str, + *, + permission_slug: str, + parent_resource_id: Optional[str] = None, + parent_resource_type_slug: Optional[str] = None, + parent_resource_external_id: Optional[str] = None, + limit: int = DEFAULT_LIST_RESPONSE_LIMIT, + before: Optional[str] = None, + after: Optional[str] = None, + order: PaginationOrder = "desc", + ) -> ResourcesForMembershipListResource: + if parent_resource_id is not None and ( + parent_resource_type_slug is not None + or parent_resource_external_id is not None + ): + raise ValueError( + "Cannot specify both parent_resource_id and " + "parent_resource_type_slug/parent_resource_external_id. " + "Use one identification method." + ) + if (parent_resource_type_slug is None) != (parent_resource_external_id is None): + raise ValueError( + "parent_resource_type_slug and parent_resource_external_id " + "must be provided together." + ) + + list_params: ResourcesForMembershipListFilters = { + "limit": limit, + "before": before, + "after": after, + "order": order, + "permission_slug": permission_slug, + } + if parent_resource_id is not None: + list_params["parent_resource_id"] = parent_resource_id + if parent_resource_type_slug is not None: + list_params["parent_resource_type_slug"] = parent_resource_type_slug + if parent_resource_external_id is not None: + list_params["parent_resource_external_id"] = parent_resource_external_id + + response = self._http_client.request( + f"authorization/organization_memberships/{organization_membership_id}/resources", + method=REQUEST_METHOD_GET, + params=list_params, + ) + + return WorkOSListResource[ + Resource, ResourcesForMembershipListFilters, ListMetadata + ]( + list_method=partial( + self.list_resources_for_membership, organization_membership_id + ), + list_args=list_params, + **ListPage[Resource](**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", + ) -> MembershipsForResourceListResource: + list_params: MembershipsForResourceListFilters = { + "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/{resource_id}/organization_memberships", + method=REQUEST_METHOD_GET, + params=list_params, + ) + + return WorkOSListResource[ + AuthorizationOrganizationMembership, + MembershipsForResourceListFilters, + 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", + ) -> MembershipsForResourceListResource: + list_params: MembershipsForResourceListFilters = { + "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/{organization_id}/resources/{resource_type_slug}/{external_id}/organization_memberships", + method=REQUEST_METHOD_GET, + params=list_params, + ) + + return WorkOSListResource[ + AuthorizationOrganizationMembership, + MembershipsForResourceListFilters, + 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 @@ -712,3 +924,144 @@ async def add_environment_role_permission( ) return EnvironmentRole.model_validate(response) + + # Resource-Membership Relationships + + async def list_resources_for_membership( + self, + organization_membership_id: str, + *, + permission_slug: str, + parent_resource_id: Optional[str] = None, + parent_resource_type_slug: Optional[str] = None, + parent_resource_external_id: Optional[str] = None, + limit: int = DEFAULT_LIST_RESPONSE_LIMIT, + before: Optional[str] = None, + after: Optional[str] = None, + order: PaginationOrder = "desc", + ) -> ResourcesForMembershipListResource: + if parent_resource_id is not None and ( + parent_resource_type_slug is not None + or parent_resource_external_id is not None + ): + raise ValueError( + "Cannot specify both parent_resource_id and " + "parent_resource_type_slug/parent_resource_external_id. " + "Use one identification method." + ) + if (parent_resource_type_slug is None) != (parent_resource_external_id is None): + raise ValueError( + "parent_resource_type_slug and parent_resource_external_id " + "must be provided together." + ) + + list_params: ResourcesForMembershipListFilters = { + "limit": limit, + "before": before, + "after": after, + "order": order, + "permission_slug": permission_slug, + } + if parent_resource_id is not None: + list_params["parent_resource_id"] = parent_resource_id + if parent_resource_type_slug is not None: + list_params["parent_resource_type_slug"] = parent_resource_type_slug + if parent_resource_external_id is not None: + list_params["parent_resource_external_id"] = parent_resource_external_id + + response = await self._http_client.request( + f"authorization/organization_memberships/{organization_membership_id}/resources", + method=REQUEST_METHOD_GET, + params=list_params, + ) + + return WorkOSListResource[ + Resource, ResourcesForMembershipListFilters, ListMetadata + ]( + list_method=partial( + self.list_resources_for_membership, organization_membership_id + ), + list_args=list_params, + **ListPage[Resource](**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", + ) -> MembershipsForResourceListResource: + list_params: MembershipsForResourceListFilters = { + "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/{resource_id}/organization_memberships", + method=REQUEST_METHOD_GET, + params=list_params, + ) + + return WorkOSListResource[ + AuthorizationOrganizationMembership, + MembershipsForResourceListFilters, + 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", + ) -> MembershipsForResourceListResource: + list_params: MembershipsForResourceListFilters = { + "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/{organization_id}/resources/{resource_type_slug}/{external_id}/organization_memberships", + method=REQUEST_METHOD_GET, + params=list_params, + ) + + return WorkOSListResource[ + AuthorizationOrganizationMembership, + MembershipsForResourceListFilters, + 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(), + ) diff --git a/tests/test_authorization_resource_memberships.py b/tests/test_authorization_resource_memberships.py new file mode 100644 index 00000000..067c057a --- /dev/null +++ b/tests/test_authorization_resource_memberships.py @@ -0,0 +1,344 @@ +from typing import Union + +import pytest +from tests.utils.fixtures.mock_resource import MockResource +from tests.utils.list_resource import list_response_of +from tests.utils.syncify import syncify +from workos.authorization import AsyncAuthorization, Authorization + + +def _mock_membership( + membership_id: str = "om_01ABC", + user_id: str = "user_123", + organization_id: str = "org_456", + organization_name: str = "Acme Inc", + status: str = "active", +) -> dict: + return { + "object": "organization_membership", + "id": membership_id, + "user_id": user_id, + "organization_id": organization_id, + "organization_name": organization_name, + "status": status, + "custom_attributes": None, + "created_at": "2024-01-01T00:00:00Z", + "updated_at": "2024-01-01T00:00:00Z", + } + + +@pytest.mark.sync_and_async(Authorization, AsyncAuthorization) +class TestListResourcesForMembership: + @pytest.fixture(autouse=True) + def setup(self, module_instance: Union[Authorization, AsyncAuthorization]): + self.http_client = module_instance._http_client + self.authorization = module_instance + + @pytest.fixture + def mock_resources_list(self): + resources = [MockResource(id=f"res_{i}").dict() for i in range(3)] + return { + "data": resources, + "list_metadata": {"before": None, "after": None}, + "object": "list", + } + + @pytest.fixture + def mock_resources_multiple_pages(self): + resources = [MockResource(id=f"res_{i}").dict() for i in range(40)] + return list_response_of(data=resources) + + def test_list_resources_for_membership( + self, mock_resources_list, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resources_list, 200 + ) + + result = syncify( + self.authorization.list_resources_for_membership( + "om_01ABC", + permission_slug="documents:read", + ) + ) + + assert len(result.data) == 3 + assert request_kwargs["method"] == "get" + assert request_kwargs["url"].endswith( + "/authorization/organization_memberships/om_01ABC/resources" + ) + assert request_kwargs["params"]["permission_slug"] == "documents:read" + + def test_list_resources_for_membership_with_parent_resource_id( + self, mock_resources_list, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resources_list, 200 + ) + + syncify( + self.authorization.list_resources_for_membership( + "om_01ABC", + permission_slug="documents:read", + parent_resource_id="res_parent_01", + ) + ) + + assert request_kwargs["params"]["parent_resource_id"] == "res_parent_01" + assert "parent_resource_type_slug" not in request_kwargs["params"] + assert "parent_resource_external_id" not in request_kwargs["params"] + + def test_list_resources_for_membership_with_parent_external_id( + self, mock_resources_list, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resources_list, 200 + ) + + syncify( + self.authorization.list_resources_for_membership( + "om_01ABC", + permission_slug="documents:read", + parent_resource_type_slug="folder", + parent_resource_external_id="folder_abc", + ) + ) + + assert request_kwargs["params"]["parent_resource_type_slug"] == "folder" + assert request_kwargs["params"]["parent_resource_external_id"] == "folder_abc" + assert "parent_resource_id" not in request_kwargs["params"] + + def test_list_resources_for_membership_rejects_both_parent_id_and_external_id( + self, + ): + with pytest.raises(ValueError, match="Cannot specify both"): + syncify( + self.authorization.list_resources_for_membership( + "om_01ABC", + permission_slug="documents:read", + parent_resource_id="res_parent_01", + parent_resource_external_id="folder_abc", + ) + ) + + def test_list_resources_for_membership_rejects_type_slug_without_external_id( + self, + ): + with pytest.raises(ValueError, match="must be provided together"): + syncify( + self.authorization.list_resources_for_membership( + "om_01ABC", + permission_slug="documents:read", + parent_resource_type_slug="folder", + ) + ) + + def test_list_resources_for_membership_rejects_external_id_without_type_slug( + self, + ): + with pytest.raises(ValueError, match="must be provided together"): + syncify( + self.authorization.list_resources_for_membership( + "om_01ABC", + permission_slug="documents:read", + parent_resource_external_id="folder_abc", + ) + ) + + def test_list_resources_for_membership_auto_pagination( + self, + mock_resources_multiple_pages, + test_auto_pagination, + ): + test_auto_pagination( + http_client=self.http_client, + list_function=self.authorization.list_resources_for_membership, + expected_all_page_data=mock_resources_multiple_pages["data"], + list_function_params={ + "organization_membership_id": "om_01ABC", + "permission_slug": "documents:read", + }, + url_path_keys=["organization_membership_id"], + ) + + +@pytest.mark.sync_and_async(Authorization, AsyncAuthorization) +class TestListMembershipsForResource: + @pytest.fixture(autouse=True) + def setup(self, module_instance: Union[Authorization, AsyncAuthorization]): + self.http_client = module_instance._http_client + self.authorization = module_instance + + @pytest.fixture + def mock_memberships_list(self): + memberships = [_mock_membership(membership_id=f"om_{i}") for i in range(3)] + return { + "data": memberships, + "list_metadata": {"before": None, "after": None}, + "object": "list", + } + + @pytest.fixture + def mock_empty_memberships_list(self): + return { + "data": [], + "list_metadata": {"before": None, "after": None}, + "object": "list", + } + + @pytest.fixture + def mock_memberships_multiple_pages(self): + memberships = [_mock_membership(membership_id=f"om_{i}") for i in range(40)] + return list_response_of(data=memberships) + + def test_list_memberships_for_resource( + self, mock_memberships_list, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_memberships_list, 200 + ) + + result = syncify( + self.authorization.list_memberships_for_resource( + "res_01ABC", + permission_slug="documents:read", + ) + ) + + assert len(result.data) == 3 + assert request_kwargs["method"] == "get" + assert request_kwargs["url"].endswith( + "/authorization/resources/res_01ABC/organization_memberships" + ) + assert request_kwargs["params"]["permission_slug"] == "documents:read" + + def test_list_memberships_for_resource_empty( + self, mock_empty_memberships_list, capture_and_mock_http_client_request + ): + capture_and_mock_http_client_request( + self.http_client, mock_empty_memberships_list, 200 + ) + + result = syncify( + self.authorization.list_memberships_for_resource( + "res_01ABC", + permission_slug="documents:read", + ) + ) + + assert len(result.data) == 0 + + def test_list_memberships_for_resource_with_assignment( + self, mock_memberships_list, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_memberships_list, 200 + ) + + syncify( + self.authorization.list_memberships_for_resource( + "res_01ABC", + permission_slug="documents:read", + assignment="direct", + ) + ) + + assert request_kwargs["params"]["assignment"] == "direct" + + def test_list_memberships_for_resource_auto_pagination( + self, + mock_memberships_multiple_pages, + test_auto_pagination, + ): + test_auto_pagination( + http_client=self.http_client, + list_function=self.authorization.list_memberships_for_resource, + expected_all_page_data=mock_memberships_multiple_pages["data"], + list_function_params={ + "resource_id": "res_01ABC", + "permission_slug": "documents:read", + }, + url_path_keys=["resource_id"], + ) + + +@pytest.mark.sync_and_async(Authorization, AsyncAuthorization) +class TestListMembershipsForResourceByExternalId: + @pytest.fixture(autouse=True) + def setup(self, module_instance: Union[Authorization, AsyncAuthorization]): + self.http_client = module_instance._http_client + self.authorization = module_instance + + @pytest.fixture + def mock_memberships_list(self): + memberships = [_mock_membership(membership_id=f"om_{i}") for i in range(3)] + return { + "data": memberships, + "list_metadata": {"before": None, "after": None}, + "object": "list", + } + + @pytest.fixture + def mock_memberships_multiple_pages(self): + memberships = [_mock_membership(membership_id=f"om_{i}") for i in range(40)] + return list_response_of(data=memberships) + + def test_list_memberships_for_resource_by_external_id( + self, mock_memberships_list, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_memberships_list, 200 + ) + + result = syncify( + self.authorization.list_memberships_for_resource_by_external_id( + "org_456", + "document", + "doc_abc", + permission_slug="documents:read", + ) + ) + + assert len(result.data) == 3 + assert request_kwargs["method"] == "get" + assert request_kwargs["url"].endswith( + "/authorization/organizations/org_456/resources/document/doc_abc/organization_memberships" + ) + assert request_kwargs["params"]["permission_slug"] == "documents:read" + + def test_list_memberships_for_resource_by_external_id_with_assignment( + self, mock_memberships_list, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_memberships_list, 200 + ) + + syncify( + self.authorization.list_memberships_for_resource_by_external_id( + "org_456", + "document", + "doc_abc", + permission_slug="documents:read", + assignment="indirect", + ) + ) + + assert request_kwargs["params"]["assignment"] == "indirect" + + def test_list_memberships_for_resource_by_external_id_auto_pagination( + self, + mock_memberships_multiple_pages, + test_auto_pagination, + ): + test_auto_pagination( + http_client=self.http_client, + list_function=self.authorization.list_memberships_for_resource_by_external_id, + expected_all_page_data=mock_memberships_multiple_pages["data"], + list_function_params={ + "organization_id": "org_456", + "resource_type_slug": "document", + "external_id": "doc_abc", + "permission_slug": "documents:read", + }, + url_path_keys=["organization_id", "resource_type_slug", "external_id"], + ) From 43bae56b6a4b7ba2a211b06b2a686df815e4864e Mon Sep 17 00:00:00 2001 From: swaroopakkineni Date: Mon, 23 Feb 2026 05:47:39 -1000 Subject: [PATCH 2/5] lol --- src/workos/authorization.py | 75 +++++-------------- src/workos/types/authorization/__init__.py | 5 ++ .../authorization/resource_identifier.py | 21 ++++++ ...test_authorization_resource_memberships.py | 73 ++++-------------- 4 files changed, 61 insertions(+), 113 deletions(-) create mode 100644 src/workos/types/authorization/resource_identifier.py diff --git a/src/workos/authorization.py b/src/workos/authorization.py index 3e886d42..834131ef 100644 --- a/src/workos/authorization.py +++ b/src/workos/authorization.py @@ -13,6 +13,7 @@ from workos.types.authorization.organization_role import OrganizationRole from workos.types.authorization.permission import Permission from workos.types.authorization.resource import Resource +from workos.types.authorization.resource_identifier import ParentResourceIdentifier from workos.types.authorization.role import Role, RoleList from workos.types.list_resource import ( ListArgs, @@ -48,9 +49,7 @@ class PermissionListFilters(ListArgs, total=False): class ResourcesForMembershipListFilters(ListArgs, total=False): permission_slug: str - parent_resource_id: Optional[str] - parent_resource_type_slug: Optional[str] - parent_resource_external_id: Optional[str] + parent_resource: ParentResourceIdentifier ResourcesForMembershipListResource = WorkOSListResource[ @@ -197,9 +196,7 @@ def list_resources_for_membership( organization_membership_id: str, *, permission_slug: str, - parent_resource_id: Optional[str] = None, - parent_resource_type_slug: Optional[str] = None, - parent_resource_external_id: Optional[str] = None, + parent_resource: ParentResourceIdentifier, limit: int = DEFAULT_LIST_RESPONSE_LIMIT, before: Optional[str] = None, after: Optional[str] = None, @@ -515,47 +512,30 @@ def list_resources_for_membership( organization_membership_id: str, *, permission_slug: str, - parent_resource_id: Optional[str] = None, - parent_resource_type_slug: Optional[str] = None, - parent_resource_external_id: Optional[str] = None, + parent_resource: ParentResourceIdentifier, limit: int = DEFAULT_LIST_RESPONSE_LIMIT, before: Optional[str] = None, after: Optional[str] = None, order: PaginationOrder = "desc", ) -> ResourcesForMembershipListResource: - if parent_resource_id is not None and ( - parent_resource_type_slug is not None - or parent_resource_external_id is not None - ): - raise ValueError( - "Cannot specify both parent_resource_id and " - "parent_resource_type_slug/parent_resource_external_id. " - "Use one identification method." - ) - if (parent_resource_type_slug is None) != (parent_resource_external_id is None): - raise ValueError( - "parent_resource_type_slug and parent_resource_external_id " - "must be provided together." - ) - list_params: ResourcesForMembershipListFilters = { "limit": limit, "before": before, "after": after, "order": order, "permission_slug": permission_slug, + "parent_resource": parent_resource, + } + + http_params: Dict[str, Any] = { + k: v for k, v in list_params.items() if k != "parent_resource" } - if parent_resource_id is not None: - list_params["parent_resource_id"] = parent_resource_id - if parent_resource_type_slug is not None: - list_params["parent_resource_type_slug"] = parent_resource_type_slug - if parent_resource_external_id is not None: - list_params["parent_resource_external_id"] = parent_resource_external_id + http_params.update(parent_resource) response = self._http_client.request( f"authorization/organization_memberships/{organization_membership_id}/resources", method=REQUEST_METHOD_GET, - params=list_params, + params=http_params, ) return WorkOSListResource[ @@ -932,47 +912,30 @@ async def list_resources_for_membership( organization_membership_id: str, *, permission_slug: str, - parent_resource_id: Optional[str] = None, - parent_resource_type_slug: Optional[str] = None, - parent_resource_external_id: Optional[str] = None, + parent_resource: ParentResourceIdentifier, limit: int = DEFAULT_LIST_RESPONSE_LIMIT, before: Optional[str] = None, after: Optional[str] = None, order: PaginationOrder = "desc", ) -> ResourcesForMembershipListResource: - if parent_resource_id is not None and ( - parent_resource_type_slug is not None - or parent_resource_external_id is not None - ): - raise ValueError( - "Cannot specify both parent_resource_id and " - "parent_resource_type_slug/parent_resource_external_id. " - "Use one identification method." - ) - if (parent_resource_type_slug is None) != (parent_resource_external_id is None): - raise ValueError( - "parent_resource_type_slug and parent_resource_external_id " - "must be provided together." - ) - list_params: ResourcesForMembershipListFilters = { "limit": limit, "before": before, "after": after, "order": order, "permission_slug": permission_slug, + "parent_resource": parent_resource, + } + + http_params: Dict[str, Any] = { + k: v for k, v in list_params.items() if k != "parent_resource" } - if parent_resource_id is not None: - list_params["parent_resource_id"] = parent_resource_id - if parent_resource_type_slug is not None: - list_params["parent_resource_type_slug"] = parent_resource_type_slug - if parent_resource_external_id is not None: - list_params["parent_resource_external_id"] = parent_resource_external_id + http_params.update(parent_resource) response = await self._http_client.request( f"authorization/organization_memberships/{organization_membership_id}/resources", method=REQUEST_METHOD_GET, - params=list_params, + params=http_params, ) return WorkOSListResource[ diff --git a/src/workos/types/authorization/__init__.py b/src/workos/types/authorization/__init__.py index 9eb705a0..a8488218 100644 --- a/src/workos/types/authorization/__init__.py +++ b/src/workos/types/authorization/__init__.py @@ -13,6 +13,11 @@ ) from workos.types.authorization.permission import Permission from workos.types.authorization.resource import Resource +from workos.types.authorization.resource_identifier import ( + ParentResourceIdentifier, + ParentResourceIdentifierByExternalId, + ParentResourceIdentifierById, +) from workos.types.authorization.role import ( Role, RoleList, diff --git a/src/workos/types/authorization/resource_identifier.py b/src/workos/types/authorization/resource_identifier.py new file mode 100644 index 00000000..467bb2ea --- /dev/null +++ b/src/workos/types/authorization/resource_identifier.py @@ -0,0 +1,21 @@ +from typing import Union + +from typing_extensions import TypedDict + + +class ParentResourceIdentifierById(TypedDict): + """Identify a parent resource by its WorkOS-assigned ID.""" + + parent_resource_id: str + + +class ParentResourceIdentifierByExternalId(TypedDict): + """Identify a parent resource by its external ID and resource type.""" + + parent_resource_type_slug: str + parent_resource_external_id: str + + +ParentResourceIdentifier = Union[ + ParentResourceIdentifierById, ParentResourceIdentifierByExternalId +] diff --git a/tests/test_authorization_resource_memberships.py b/tests/test_authorization_resource_memberships.py index 067c057a..b36dcb8e 100644 --- a/tests/test_authorization_resource_memberships.py +++ b/tests/test_authorization_resource_memberships.py @@ -5,6 +5,10 @@ from tests.utils.list_resource import list_response_of from tests.utils.syncify import syncify from workos.authorization import AsyncAuthorization, Authorization +from workos.types.authorization.resource_identifier import ( + ParentResourceIdentifierByExternalId, + ParentResourceIdentifierById, +) def _mock_membership( @@ -48,7 +52,7 @@ def mock_resources_multiple_pages(self): resources = [MockResource(id=f"res_{i}").dict() for i in range(40)] return list_response_of(data=resources) - def test_list_resources_for_membership( + def test_list_resources_for_membership_with_parent_resource_id( self, mock_resources_list, capture_and_mock_http_client_request ): request_kwargs = capture_and_mock_http_client_request( @@ -59,6 +63,9 @@ def test_list_resources_for_membership( self.authorization.list_resources_for_membership( "om_01ABC", permission_slug="documents:read", + parent_resource=ParentResourceIdentifierById( + parent_resource_id="res_parent_01", + ), ) ) @@ -68,22 +75,6 @@ def test_list_resources_for_membership( "/authorization/organization_memberships/om_01ABC/resources" ) assert request_kwargs["params"]["permission_slug"] == "documents:read" - - def test_list_resources_for_membership_with_parent_resource_id( - self, mock_resources_list, capture_and_mock_http_client_request - ): - request_kwargs = capture_and_mock_http_client_request( - self.http_client, mock_resources_list, 200 - ) - - syncify( - self.authorization.list_resources_for_membership( - "om_01ABC", - permission_slug="documents:read", - parent_resource_id="res_parent_01", - ) - ) - assert request_kwargs["params"]["parent_resource_id"] == "res_parent_01" assert "parent_resource_type_slug" not in request_kwargs["params"] assert "parent_resource_external_id" not in request_kwargs["params"] @@ -99,8 +90,10 @@ def test_list_resources_for_membership_with_parent_external_id( self.authorization.list_resources_for_membership( "om_01ABC", permission_slug="documents:read", - parent_resource_type_slug="folder", - parent_resource_external_id="folder_abc", + parent_resource=ParentResourceIdentifierByExternalId( + parent_resource_type_slug="folder", + parent_resource_external_id="folder_abc", + ), ) ) @@ -108,43 +101,6 @@ def test_list_resources_for_membership_with_parent_external_id( assert request_kwargs["params"]["parent_resource_external_id"] == "folder_abc" assert "parent_resource_id" not in request_kwargs["params"] - def test_list_resources_for_membership_rejects_both_parent_id_and_external_id( - self, - ): - with pytest.raises(ValueError, match="Cannot specify both"): - syncify( - self.authorization.list_resources_for_membership( - "om_01ABC", - permission_slug="documents:read", - parent_resource_id="res_parent_01", - parent_resource_external_id="folder_abc", - ) - ) - - def test_list_resources_for_membership_rejects_type_slug_without_external_id( - self, - ): - with pytest.raises(ValueError, match="must be provided together"): - syncify( - self.authorization.list_resources_for_membership( - "om_01ABC", - permission_slug="documents:read", - parent_resource_type_slug="folder", - ) - ) - - def test_list_resources_for_membership_rejects_external_id_without_type_slug( - self, - ): - with pytest.raises(ValueError, match="must be provided together"): - syncify( - self.authorization.list_resources_for_membership( - "om_01ABC", - permission_slug="documents:read", - parent_resource_external_id="folder_abc", - ) - ) - def test_list_resources_for_membership_auto_pagination( self, mock_resources_multiple_pages, @@ -157,8 +113,11 @@ def test_list_resources_for_membership_auto_pagination( list_function_params={ "organization_membership_id": "om_01ABC", "permission_slug": "documents:read", + "parent_resource": ParentResourceIdentifierById( + parent_resource_id="res_parent_01", + ), }, - url_path_keys=["organization_membership_id"], + url_path_keys=["organization_membership_id", "parent_resource"], ) From d66d6ec51f5cca771ef249d23597da95e802fbbb Mon Sep 17 00:00:00 2001 From: swaroopakkineni Date: Mon, 23 Feb 2026 05:59:46 -1000 Subject: [PATCH 3/5] minor nit --- src/workos/types/authorization/resource_identifier.py | 4 ---- 1 file changed, 4 deletions(-) diff --git a/src/workos/types/authorization/resource_identifier.py b/src/workos/types/authorization/resource_identifier.py index 467bb2ea..f3d6edb7 100644 --- a/src/workos/types/authorization/resource_identifier.py +++ b/src/workos/types/authorization/resource_identifier.py @@ -4,14 +4,10 @@ class ParentResourceIdentifierById(TypedDict): - """Identify a parent resource by its WorkOS-assigned ID.""" - parent_resource_id: str class ParentResourceIdentifierByExternalId(TypedDict): - """Identify a parent resource by its external ID and resource type.""" - parent_resource_type_slug: str parent_resource_external_id: str From 9149f71dfdda7d39c2b11d9388d3e7ca61deecfb Mon Sep 17 00:00:00 2001 From: swaroopakkineni Date: Mon, 23 Feb 2026 06:31:01 -1000 Subject: [PATCH 4/5] refactor --- src/workos/authorization.py | 21 +++----- ...test_authorization_resource_memberships.py | 49 +++++++++++++++++++ 2 files changed, 57 insertions(+), 13 deletions(-) diff --git a/src/workos/authorization.py b/src/workos/authorization.py index 834131ef..41cb43ec 100644 --- a/src/workos/authorization.py +++ b/src/workos/authorization.py @@ -49,7 +49,6 @@ class PermissionListFilters(ListArgs, total=False): class ResourcesForMembershipListFilters(ListArgs, total=False): permission_slug: str - parent_resource: ParentResourceIdentifier ResourcesForMembershipListResource = WorkOSListResource[ @@ -524,12 +523,9 @@ def list_resources_for_membership( "after": after, "order": order, "permission_slug": permission_slug, - "parent_resource": parent_resource, } - http_params: Dict[str, Any] = { - k: v for k, v in list_params.items() if k != "parent_resource" - } + http_params: Dict[str, Any] = {**list_params} http_params.update(parent_resource) response = self._http_client.request( @@ -542,7 +538,9 @@ def list_resources_for_membership( Resource, ResourcesForMembershipListFilters, ListMetadata ]( list_method=partial( - self.list_resources_for_membership, organization_membership_id + self.list_resources_for_membership, + organization_membership_id, + parent_resource=parent_resource, ), list_args=list_params, **ListPage[Resource](**response).model_dump(), @@ -905,8 +903,6 @@ async def add_environment_role_permission( return EnvironmentRole.model_validate(response) - # Resource-Membership Relationships - async def list_resources_for_membership( self, organization_membership_id: str, @@ -924,12 +920,9 @@ async def list_resources_for_membership( "after": after, "order": order, "permission_slug": permission_slug, - "parent_resource": parent_resource, } - http_params: Dict[str, Any] = { - k: v for k, v in list_params.items() if k != "parent_resource" - } + http_params: Dict[str, Any] = {**list_params} http_params.update(parent_resource) response = await self._http_client.request( @@ -942,7 +935,9 @@ async def list_resources_for_membership( Resource, ResourcesForMembershipListFilters, ListMetadata ]( list_method=partial( - self.list_resources_for_membership, organization_membership_id + self.list_resources_for_membership, + organization_membership_id, + parent_resource=parent_resource, ), list_args=list_params, **ListPage[Resource](**response).model_dump(), diff --git a/tests/test_authorization_resource_memberships.py b/tests/test_authorization_resource_memberships.py index b36dcb8e..28b0c670 100644 --- a/tests/test_authorization_resource_memberships.py +++ b/tests/test_authorization_resource_memberships.py @@ -69,6 +69,7 @@ def test_list_resources_for_membership_with_parent_resource_id( ) ) + assert result.object == "list" assert len(result.data) == 3 assert request_kwargs["method"] == "get" assert request_kwargs["url"].endswith( @@ -101,6 +102,54 @@ def test_list_resources_for_membership_with_parent_external_id( assert request_kwargs["params"]["parent_resource_external_id"] == "folder_abc" assert "parent_resource_id" not in request_kwargs["params"] + def test_list_resources_for_membership_empty( + self, capture_and_mock_http_client_request + ): + empty_response = { + "data": [], + "list_metadata": {"before": None, "after": None}, + "object": "list", + } + capture_and_mock_http_client_request(self.http_client, empty_response, 200) + + result = syncify( + self.authorization.list_resources_for_membership( + "om_01ABC", + permission_slug="documents:read", + parent_resource=ParentResourceIdentifierById( + parent_resource_id="res_parent_01", + ), + ) + ) + + assert result.object == "list" + assert len(result.data) == 0 + + def test_list_resources_for_membership_passes_pagination_params( + self, mock_resources_list, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resources_list, 200 + ) + + syncify( + self.authorization.list_resources_for_membership( + "om_01ABC", + permission_slug="documents:read", + parent_resource=ParentResourceIdentifierById( + parent_resource_id="res_parent_01", + ), + limit=10, + after="res_cursor123", + order="desc", + ) + ) + + assert request_kwargs["params"]["permission_slug"] == "documents:read" + assert request_kwargs["params"]["limit"] == 10 + assert request_kwargs["params"]["after"] == "res_cursor123" + assert request_kwargs["params"]["order"] == "desc" + def test_list_resources_for_membership_auto_pagination( self, mock_resources_multiple_pages, From e9a69758df9563130d78555a0a90a69081726585 Mon Sep 17 00:00:00 2001 From: swaroopakkineni Date: Mon, 23 Feb 2026 06:43:42 -1000 Subject: [PATCH 5/5] moar --- src/workos/authorization.py | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/src/workos/authorization.py b/src/workos/authorization.py index 41cb43ec..e13d0163 100644 --- a/src/workos/authorization.py +++ b/src/workos/authorization.py @@ -52,7 +52,9 @@ class ResourcesForMembershipListFilters(ListArgs, total=False): ResourcesForMembershipListResource = WorkOSListResource[ - Resource, ResourcesForMembershipListFilters, ListMetadata + Resource, + ResourcesForMembershipListFilters, + ListMetadata, ] @@ -188,8 +190,6 @@ def add_environment_role_permission( permission_slug: str, ) -> SyncOrAsync[EnvironmentRole]: ... - # Resource-Membership Relationships - def list_resources_for_membership( self, organization_membership_id: str, @@ -504,8 +504,6 @@ def add_environment_role_permission( return EnvironmentRole.model_validate(response) - # Resource-Membership Relationships - def list_resources_for_membership( self, organization_membership_id: str,