From 63f2f60bf7e10eaf7f9545ecd1cf3b510d86a57c Mon Sep 17 00:00:00 2001 From: swaroopakkineni Date: Fri, 20 Feb 2026 11:36:11 -1000 Subject: [PATCH 1/3] FGA_2: listResources(), get/update/delete resource_by_external_id --- src/workos/authorization.py | 231 +++++++++++++++ ...test_authorization_resource_external_id.py | 271 ++++++++++++++++++ 2 files changed, 502 insertions(+) create mode 100644 tests/test_authorization_resource_external_id.py diff --git a/src/workos/authorization.py b/src/workos/authorization.py index 6e12f035..145595a0 100644 --- a/src/workos/authorization.py +++ b/src/workos/authorization.py @@ -8,6 +8,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.role import Role, RoleList from workos.types.list_resource import ( ListArgs, @@ -28,6 +29,15 @@ ) AUTHORIZATION_PERMISSIONS_PATH = "authorization/permissions" +AUTHORIZATION_ORGANIZATIONS_PATH = "authorization/organizations" + + +class ResourceListFilters(ListArgs, total=False): + organization_id: str + resource_type_slug: Optional[str] + + +ResourcesListResource = WorkOSListResource[Resource, ResourceListFilters, ListMetadata] _role_adapter: TypeAdapter[Role] = TypeAdapter(Role) @@ -161,6 +171,45 @@ def add_environment_role_permission( permission_slug: str, ) -> SyncOrAsync[EnvironmentRole]: ... + # Resources (External ID) + List + + def get_resource_by_external_id( + self, + organization_id: str, + resource_type: str, + external_id: str, + ) -> SyncOrAsync[Resource]: ... + + def update_resource_by_external_id( + self, + organization_id: str, + resource_type: str, + external_id: str, + *, + name: Optional[str] = None, + description: Optional[str] = None, + ) -> SyncOrAsync[Resource]: ... + + def delete_resource_by_external_id( + self, + organization_id: str, + resource_type: str, + external_id: str, + *, + cascade_delete: Optional[bool] = None, + ) -> SyncOrAsync[None]: ... + + def list_resources( + self, + organization_id: str, + *, + resource_type_slug: Optional[str] = None, + limit: int = DEFAULT_LIST_RESPONSE_LIMIT, + before: Optional[str] = None, + after: Optional[str] = None, + order: PaginationOrder = "desc", + ) -> SyncOrAsync[ResourcesListResource]: ... + class Authorization(AuthorizationModule): _http_client: SyncHTTPClient @@ -437,6 +486,97 @@ def add_environment_role_permission( return EnvironmentRole.model_validate(response) + # Resources (External ID) + List + + def get_resource_by_external_id( + self, + organization_id: str, + resource_type: str, + external_id: str, + ) -> Resource: + response = self._http_client.request( + f"{AUTHORIZATION_ORGANIZATIONS_PATH}/{organization_id}/resources/{resource_type}/{external_id}", + method=REQUEST_METHOD_GET, + ) + + return Resource.model_validate(response) + + def update_resource_by_external_id( + self, + organization_id: str, + resource_type: str, + external_id: str, + *, + name: Optional[str] = None, + description: Optional[str] = None, + ) -> Resource: + json: Dict[str, Any] = {} + if name is not None: + json["name"] = name + if description is not None: + json["description"] = description + + response = self._http_client.request( + f"{AUTHORIZATION_ORGANIZATIONS_PATH}/{organization_id}/resources/{resource_type}/{external_id}", + method=REQUEST_METHOD_PATCH, + json=json, + ) + + return Resource.model_validate(response) + + def delete_resource_by_external_id( + self, + organization_id: str, + resource_type: str, + external_id: str, + *, + cascade_delete: Optional[bool] = None, + ) -> None: + path = f"{AUTHORIZATION_ORGANIZATIONS_PATH}/{organization_id}/resources/{resource_type}/{external_id}" + params: Dict[str, bool] = {} + if cascade_delete is not None: + params["cascade_delete"] = cascade_delete + + self._http_client.request( + path, + method=REQUEST_METHOD_DELETE, + params=params if params else None, + ) + + def list_resources( + self, + organization_id: str, + *, + resource_type_slug: Optional[str] = None, + limit: int = DEFAULT_LIST_RESPONSE_LIMIT, + before: Optional[str] = None, + after: Optional[str] = None, + order: PaginationOrder = "desc", + ) -> ResourcesListResource: + list_params: ResourceListFilters = { + "organization_id": organization_id, + "limit": limit, + "before": before, + "after": after, + "order": order, + } + if resource_type_slug is not None: + list_params["resource_type_slug"] = resource_type_slug + + query_params = {k: v for k, v in list_params.items() if k != "organization_id"} + + response = self._http_client.request( + f"{AUTHORIZATION_ORGANIZATIONS_PATH}/{organization_id}/resources", + method=REQUEST_METHOD_GET, + params=query_params, + ) + + return WorkOSListResource[Resource, ResourceListFilters, ListMetadata]( + list_method=self.list_resources, + list_args=list_params, + **ListPage[Resource](**response).model_dump(), + ) + class AsyncAuthorization(AuthorizationModule): _http_client: AsyncHTTPClient @@ -712,3 +852,94 @@ async def add_environment_role_permission( ) return EnvironmentRole.model_validate(response) + + # Resources (External ID) + List + + async def get_resource_by_external_id( + self, + organization_id: str, + resource_type: str, + external_id: str, + ) -> Resource: + response = await self._http_client.request( + f"{AUTHORIZATION_ORGANIZATIONS_PATH}/{organization_id}/resources/{resource_type}/{external_id}", + method=REQUEST_METHOD_GET, + ) + + return Resource.model_validate(response) + + async def update_resource_by_external_id( + self, + organization_id: str, + resource_type: str, + external_id: str, + *, + name: Optional[str] = None, + description: Optional[str] = None, + ) -> Resource: + json: Dict[str, Any] = {} + if name is not None: + json["name"] = name + if description is not None: + json["description"] = description + + response = await self._http_client.request( + f"{AUTHORIZATION_ORGANIZATIONS_PATH}/{organization_id}/resources/{resource_type}/{external_id}", + method=REQUEST_METHOD_PATCH, + json=json, + ) + + return Resource.model_validate(response) + + async def delete_resource_by_external_id( + self, + organization_id: str, + resource_type: str, + external_id: str, + *, + cascade_delete: Optional[bool] = None, + ) -> None: + path = f"{AUTHORIZATION_ORGANIZATIONS_PATH}/{organization_id}/resources/{resource_type}/{external_id}" + params: Dict[str, bool] = {} + if cascade_delete is not None: + params["cascade_delete"] = cascade_delete + + await self._http_client.request( + path, + method=REQUEST_METHOD_DELETE, + params=params if params else None, + ) + + async def list_resources( + self, + organization_id: str, + *, + resource_type_slug: Optional[str] = None, + limit: int = DEFAULT_LIST_RESPONSE_LIMIT, + before: Optional[str] = None, + after: Optional[str] = None, + order: PaginationOrder = "desc", + ) -> ResourcesListResource: + list_params: ResourceListFilters = { + "organization_id": organization_id, + "limit": limit, + "before": before, + "after": after, + "order": order, + } + if resource_type_slug is not None: + list_params["resource_type_slug"] = resource_type_slug + + query_params = {k: v for k, v in list_params.items() if k != "organization_id"} + + response = await self._http_client.request( + f"{AUTHORIZATION_ORGANIZATIONS_PATH}/{organization_id}/resources", + method=REQUEST_METHOD_GET, + params=query_params, + ) + + return WorkOSListResource[Resource, ResourceListFilters, ListMetadata]( + list_method=self.list_resources, + list_args=list_params, + **ListPage[Resource](**response).model_dump(), + ) diff --git a/tests/test_authorization_resource_external_id.py b/tests/test_authorization_resource_external_id.py new file mode 100644 index 00000000..4a6c7a47 --- /dev/null +++ b/tests/test_authorization_resource_external_id.py @@ -0,0 +1,271 @@ +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 tests.types.test_auto_pagination_function import TestAutoPaginationFunction +from workos.authorization import AsyncAuthorization, Authorization + + +MOCK_ORG_ID = "org_01EHT88Z8J8795GZNQ4ZP1J81T" +MOCK_RESOURCE_TYPE = "document" +MOCK_EXTERNAL_ID = "ext_123" + + +@pytest.mark.sync_and_async(Authorization, AsyncAuthorization) +class TestAuthorizationResourceExternalId: + @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_resource(self): + return MockResource( + id="res_01ABC", + external_id=MOCK_EXTERNAL_ID, + resource_type_slug=MOCK_RESOURCE_TYPE, + organization_id=MOCK_ORG_ID, + ).dict() + + @pytest.fixture + def mock_resources_list(self, mock_resource): + return list_response_of(data=[mock_resource]) + + @pytest.fixture + def mock_resources_empty_list(self): + return list_response_of(data=[]) + + @pytest.fixture + def mock_resources_multiple(self): + resources = [ + MockResource(id=f"res_{i:05d}", external_id=f"ext_{i}").dict() + for i in range(15) + ] + return resources + + # --- get_resource_by_external_id --- + + def test_get_resource_by_external_id( + self, mock_resource, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resource, 200 + ) + + resource = syncify( + self.authorization.get_resource_by_external_id( + MOCK_ORG_ID, MOCK_RESOURCE_TYPE, MOCK_EXTERNAL_ID + ) + ) + + assert resource.id == "res_01ABC" + assert resource.external_id == MOCK_EXTERNAL_ID + assert resource.object == "authorization_resource" + assert request_kwargs["method"] == "get" + assert request_kwargs["url"].endswith( + f"/authorization/organizations/{MOCK_ORG_ID}/resources/{MOCK_RESOURCE_TYPE}/{MOCK_EXTERNAL_ID}" + ) + + def test_get_resource_by_external_id_url_construction( + self, mock_resource, capture_and_mock_http_client_request + ): + org_id = "org_different" + res_type = "folder" + ext_id = "my-folder-123" + + mock_res = MockResource( + id="res_02XYZ", + external_id=ext_id, + resource_type_slug=res_type, + organization_id=org_id, + ).dict() + + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_res, 200 + ) + + resource = syncify( + self.authorization.get_resource_by_external_id(org_id, res_type, ext_id) + ) + + assert resource.id == "res_02XYZ" + assert request_kwargs["url"].endswith( + f"/authorization/organizations/{org_id}/resources/{res_type}/{ext_id}" + ) + + # --- update_resource_by_external_id --- + + def test_update_resource_by_external_id_with_name( + self, mock_resource, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resource, 200 + ) + + resource = syncify( + self.authorization.update_resource_by_external_id( + MOCK_ORG_ID, + MOCK_RESOURCE_TYPE, + MOCK_EXTERNAL_ID, + name="Updated Name", + description="Updated description", + ) + ) + + assert resource.id == "res_01ABC" + assert request_kwargs["method"] == "patch" + assert request_kwargs["url"].endswith( + f"/authorization/organizations/{MOCK_ORG_ID}/resources/{MOCK_RESOURCE_TYPE}/{MOCK_EXTERNAL_ID}" + ) + assert request_kwargs["json"] == { + "name": "Updated Name", + "description": "Updated description", + } + + def test_update_resource_by_external_id_empty( + self, mock_resource, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resource, 200 + ) + + syncify( + self.authorization.update_resource_by_external_id( + MOCK_ORG_ID, MOCK_RESOURCE_TYPE, MOCK_EXTERNAL_ID + ) + ) + + assert request_kwargs["method"] == "patch" + assert request_kwargs["json"] == {} + + # --- delete_resource_by_external_id --- + + def test_delete_resource_by_external_id_without_cascade( + self, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, + status_code=202, + headers={"content-type": "text/plain; charset=utf-8"}, + ) + + response = syncify( + self.authorization.delete_resource_by_external_id( + MOCK_ORG_ID, MOCK_RESOURCE_TYPE, MOCK_EXTERNAL_ID + ) + ) + + assert response is None + assert request_kwargs["method"] == "delete" + assert request_kwargs["url"].endswith( + f"/authorization/organizations/{MOCK_ORG_ID}/resources/{MOCK_RESOURCE_TYPE}/{MOCK_EXTERNAL_ID}" + ) + + def test_delete_resource_by_external_id_with_cascade( + self, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, + status_code=202, + headers={"content-type": "text/plain; charset=utf-8"}, + ) + + response = syncify( + self.authorization.delete_resource_by_external_id( + MOCK_ORG_ID, + MOCK_RESOURCE_TYPE, + MOCK_EXTERNAL_ID, + cascade_delete=True, + ) + ) + + assert response is None + assert request_kwargs["method"] == "delete" + assert request_kwargs["url"].endswith( + f"/authorization/organizations/{MOCK_ORG_ID}/resources/{MOCK_RESOURCE_TYPE}/{MOCK_EXTERNAL_ID}" + ) + assert request_kwargs["params"] == {"cascade_delete": True} + + # --- list_resources --- + + def test_list_resources_with_results( + 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 + ) + + resources_response = syncify(self.authorization.list_resources(MOCK_ORG_ID)) + + assert request_kwargs["method"] == "get" + assert request_kwargs["url"].endswith( + f"/authorization/organizations/{MOCK_ORG_ID}/resources" + ) + assert "organization_id" not in request_kwargs["params"] + assert len(resources_response.data) == 1 + assert resources_response.data[0].id == "res_01ABC" + + def test_list_resources_empty_results( + self, mock_resources_empty_list, capture_and_mock_http_client_request + ): + request_kwargs = capture_and_mock_http_client_request( + self.http_client, mock_resources_empty_list, 200 + ) + + resources_response = syncify(self.authorization.list_resources(MOCK_ORG_ID)) + + assert request_kwargs["method"] == "get" + assert len(resources_response.data) == 0 + + def test_list_resources_with_resource_type_slug_filter( + 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( + MOCK_ORG_ID, resource_type_slug="document" + ) + ) + + assert request_kwargs["method"] == "get" + assert request_kwargs["params"]["resource_type_slug"] == "document" + + def test_list_resources_with_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( + MOCK_ORG_ID, + limit=5, + after="res_cursor_abc", + before="res_cursor_xyz", + order="asc", + ) + ) + + assert request_kwargs["params"]["limit"] == 5 + assert request_kwargs["params"]["after"] == "res_cursor_abc" + assert request_kwargs["params"]["before"] == "res_cursor_xyz" + assert request_kwargs["params"]["order"] == "asc" + + def test_list_resources_auto_pagination( + self, + mock_resources_multiple, + test_auto_pagination: TestAutoPaginationFunction, + ): + test_auto_pagination( + http_client=self.http_client, + list_function=self.authorization.list_resources, + expected_all_page_data=mock_resources_multiple, + list_function_params={"organization_id": MOCK_ORG_ID}, + url_path_keys=["organization_id"], + ) From 30bc17ea84155a0fc786681313725b20bdd4aae9 Mon Sep 17 00:00:00 2001 From: swaroopakkineni Date: Tue, 24 Feb 2026 09:23:22 -1000 Subject: [PATCH 2/3] moar --- .claude/settings.local.json | 7 ------- 1 file changed, 7 deletions(-) delete mode 100644 .claude/settings.local.json diff --git a/.claude/settings.local.json b/.claude/settings.local.json deleted file mode 100644 index 32f4ee44..00000000 --- a/.claude/settings.local.json +++ /dev/null @@ -1,7 +0,0 @@ -{ - "permissions": { - "allow": [ - "Bash(uv run mypy:*)" - ] - } -} From fa295f51328e37b8fd2e8ea3b032a6150135d9c0 Mon Sep 17 00:00:00 2001 From: swaroopakkineni Date: Tue, 24 Feb 2026 10:43:48 -1000 Subject: [PATCH 3/3] moar --- src/workos/authorization.py | 90 ++++++++++++++++++------------------- 1 file changed, 45 insertions(+), 45 deletions(-) diff --git a/src/workos/authorization.py b/src/workos/authorization.py index 015fa5f6..3d545054 100644 --- a/src/workos/authorization.py +++ b/src/workos/authorization.py @@ -622,6 +622,51 @@ def delete_resource( method=REQUEST_METHOD_DELETE, ) + def list_resources( + self, + *, + organization_id: Optional[str] = None, + resource_type_slug: Optional[str] = None, + parent_resource_id: Optional[str] = None, + parent_resource_type_slug: Optional[str] = None, + parent_external_id: Optional[str] = None, + search: Optional[str] = None, + limit: int = DEFAULT_LIST_RESPONSE_LIMIT, + before: Optional[str] = None, + after: Optional[str] = None, + order: PaginationOrder = "desc", + ) -> ResourcesListResource: + list_params: ResourceListFilters = { + "limit": limit, + "before": before, + "after": after, + "order": order, + } + if organization_id is not None: + list_params["organization_id"] = organization_id + if resource_type_slug is not None: + list_params["resource_type_slug"] = resource_type_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_external_id is not None: + list_params["parent_external_id"] = parent_external_id + if search is not None: + list_params["search"] = search + + response = self._http_client.request( + AUTHORIZATION_RESOURCES_PATH, + method=REQUEST_METHOD_GET, + params=list_params, + ) + + return WorkOSListResource[Resource, ResourceListFilters, ListMetadata]( + list_method=self.list_resources, + list_args=list_params, + **ListPage[Resource](**response).model_dump(), + ) + def get_resource_by_external_id( self, organization_id: str, @@ -677,51 +722,6 @@ def delete_resource_by_external_id( params=params if params else None, ) - def list_resources( - self, - *, - organization_id: Optional[str] = None, - resource_type_slug: Optional[str] = None, - parent_resource_id: Optional[str] = None, - parent_resource_type_slug: Optional[str] = None, - parent_external_id: Optional[str] = None, - search: Optional[str] = None, - limit: int = DEFAULT_LIST_RESPONSE_LIMIT, - before: Optional[str] = None, - after: Optional[str] = None, - order: PaginationOrder = "desc", - ) -> ResourcesListResource: - list_params: ResourceListFilters = { - "limit": limit, - "before": before, - "after": after, - "order": order, - } - if organization_id is not None: - list_params["organization_id"] = organization_id - if resource_type_slug is not None: - list_params["resource_type_slug"] = resource_type_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_external_id is not None: - list_params["parent_external_id"] = parent_external_id - if search is not None: - list_params["search"] = search - - response = self._http_client.request( - AUTHORIZATION_RESOURCES_PATH, - method=REQUEST_METHOD_GET, - params=list_params, - ) - - return WorkOSListResource[Resource, ResourceListFilters, ListMetadata]( - list_method=self.list_resources, - list_args=list_params, - **ListPage[Resource](**response).model_dump(), - ) - class AsyncAuthorization(AuthorizationModule): _http_client: AsyncHTTPClient