FGA_1: create/delete/get/update resource

src/workos/authorization.py

@@ -1,13 +1,15 @@
-from typing import Any, Dict, Optional, Protocol, Sequence
+from typing import Any, Dict, Optional, Protocol, Sequence, Union
 
 from pydantic import TypeAdapter
+from typing_extensions import TypedDict
 
 from workos.types.authorization.environment_role import (
     EnvironmentRole,
     EnvironmentRoleList,
 )
 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 +30,19 @@
 )
 
 AUTHORIZATION_PERMISSIONS_PATH = "authorization/permissions"
+AUTHORIZATION_RESOURCES_PATH = "authorization/resources"
+
+
+class ParentResourceById(TypedDict):
+    parent_resource_id: str
+
+
+class ParentResourceByExternalId(TypedDict):
+    parent_resource_external_id: str
+    parent_resource_type_slug: str
+
+
+ParentResource = Union[ParentResourceById, ParentResourceByExternalId]
 
 _role_adapter: TypeAdapter[Role] = TypeAdapter(Role)
 
@@ -161,6 +176,36 @@ def add_environment_role_permission(
         permission_slug: str,
     ) -> SyncOrAsync[EnvironmentRole]: ...
 
+    # Resources
+
+    def get_resource(self, resource_id: str) -> SyncOrAsync[Resource]: ...
+
+    def create_resource(
+        self,
+        *,
+        resource_type_slug: str,
+        organization_id: str,
+        external_id: str,
+        name: str,
+        parent: ParentResource,
+        description: Optional[str] = None,
+    ) -> SyncOrAsync[Resource]: ...
+
+    def update_resource(
+        self,
+        resource_id: str,
+        *,
+        name: Optional[str] = None,
+        description: Optional[str] = None,
+    ) -> SyncOrAsync[Resource]: ...
+
+    def delete_resource(
+        self,
+        resource_id: str,
+        *,
+        cascade_delete: Optional[bool] = None,
+    ) -> SyncOrAsync[None]: ...
+
 
 class Authorization(AuthorizationModule):
     _http_client: SyncHTTPClient
@@ -437,6 +482,82 @@ def add_environment_role_permission(
 
         return EnvironmentRole.model_validate(response)
 
+    # Resources
+
+    def get_resource(self, resource_id: str) -> Resource:
+        response = self._http_client.request(
+            f"{AUTHORIZATION_RESOURCES_PATH}/{resource_id}",
+            method=REQUEST_METHOD_GET,
+        )
+
+        return Resource.model_validate(response)
+
+    def create_resource(
+        self,
+        *,
+        resource_type_slug: str,
+        organization_id: str,
+        external_id: str,
+        name: str,
+        parent: ParentResource,
+        description: Optional[str] = None,
+    ) -> Resource:
+        json: Dict[str, Any] = {
+            "resource_type_slug": resource_type_slug,
+            "organization_id": organization_id,
+            "external_id": external_id,
+            "name": name,
+            **parent,
+        }
+        if description is not None:
+            json["description"] = description
+
+        response = self._http_client.request(
+            AUTHORIZATION_RESOURCES_PATH,
+            method=REQUEST_METHOD_POST,
+            json=json,
+        )
+
+        return Resource.model_validate(response)
+
+    def update_resource(
+        self,
+        resource_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_RESOURCES_PATH}/{resource_id}",
+            method=REQUEST_METHOD_PATCH,
+            json=json,
+        )
+
+        return Resource.model_validate(response)
+
+    def delete_resource(
+        self,
+        resource_id: str,
+        *,
+        cascade_delete: Optional[bool] = None,
+    ) -> None:
+        if cascade_delete is not None:
+            self._http_client.delete_with_body(
+                f"{AUTHORIZATION_RESOURCES_PATH}/{resource_id}",
+                json={"cascade_delete": cascade_delete},
+            )
+        else:
+            self._http_client.request(
+                f"{AUTHORIZATION_RESOURCES_PATH}/{resource_id}",
+                method=REQUEST_METHOD_DELETE,
+            )
+
 
 class AsyncAuthorization(AuthorizationModule):
     _http_client: AsyncHTTPClient
@@ -712,3 +833,79 @@ async def add_environment_role_permission(
         )
 
         return EnvironmentRole.model_validate(response)
+
+    # Resources
+
+    async def get_resource(self, resource_id: str) -> Resource:
+        response = await self._http_client.request(
+            f"{AUTHORIZATION_RESOURCES_PATH}/{resource_id}",
+            method=REQUEST_METHOD_GET,
+        )
+
+        return Resource.model_validate(response)
+
+    async def create_resource(
+        self,
+        *,
+        resource_type_slug: str,
+        organization_id: str,
+        external_id: str,
+        name: str,
+        parent: ParentResource,
+        description: Optional[str] = None,
+    ) -> Resource:
+        json: Dict[str, Any] = {
+            "resource_type_slug": resource_type_slug,
+            "organization_id": organization_id,
+            "external_id": external_id,
+            "name": name,
+            **parent,
+        }
+        if description is not None:
+            json["description"] = description
+
+        response = await self._http_client.request(
+            AUTHORIZATION_RESOURCES_PATH,
+            method=REQUEST_METHOD_POST,
+            json=json,
+        )
+
+        return Resource.model_validate(response)
+
+    async def update_resource(
+        self,
+        resource_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_RESOURCES_PATH}/{resource_id}",
+            method=REQUEST_METHOD_PATCH,
+            json=json,
+        )
+
+        return Resource.model_validate(response)
+
+    async def delete_resource(
+        self,
+        resource_id: str,
+        *,
+        cascade_delete: Optional[bool] = None,
+    ) -> None:
+        if cascade_delete is not None:
+            await self._http_client.delete_with_body(
+                f"{AUTHORIZATION_RESOURCES_PATH}/{resource_id}",
+                json={"cascade_delete": cascade_delete},
+            )
+        else:
+            await self._http_client.request(
+                f"{AUTHORIZATION_RESOURCES_PATH}/{resource_id}",
+                method=REQUEST_METHOD_DELETE,
+            )