Skip to content

Fga p3 #565

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

Closed
swaroopAkkineniWorkos wants to merge 3 commits into from
Closed

Fga p3 #565

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
e623448
FGA implementation pr1
swaroopAkkineniWorkos Feb 20, 2026
a5d3ce2
cleanup
swaroopAkkineniWorkos Feb 20, 2026
da598c8
fga pt3
swaroopAkkineniWorkos Feb 20, 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
266 changes: 265 additions & 1 deletion src/workos/authorization.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,13 +1,16 @@
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.access_evaluation import AccessEvaluation
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,
Expand All @@ -28,6 +31,31 @@
)

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]


class CheckResourceById(TypedDict):
resource_id: str


class CheckResourceByExternalId(TypedDict):
resource_type: str
external_id: str


CheckResource = Union[CheckResourceById, CheckResourceByExternalId]

_role_adapter: TypeAdapter[Role] = TypeAdapter(Role)

Expand Down Expand Up @@ -161,6 +189,46 @@ 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]: ...

# Access Evaluation

def check(
self,
organization_membership_id: str,
*,
resource: CheckResource,
relation: str,
) -> SyncOrAsync[AccessEvaluation]: ...


class Authorization(AuthorizationModule):
_http_client: SyncHTTPClient
Expand Down Expand Up @@ -437,6 +505,104 @@ 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,
)

# Access Evaluation

def check(
self,
organization_membership_id: str,
*,
resource: CheckResource,
relation: str,
) -> AccessEvaluation:
json: Dict[str, Any] = {
"resource": resource,
"relation": relation,
}

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

return AccessEvaluation.model_validate(response)


class AsyncAuthorization(AuthorizationModule):
_http_client: AsyncHTTPClient
Expand Down Expand Up @@ -712,3 +878,101 @@ 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,
)

# Access Evaluation

async def check(
self,
organization_membership_id: str,
*,
resource: CheckResource,
relation: str,
) -> AccessEvaluation:
json: Dict[str, Any] = {
"resource": resource,
"relation": relation,
}

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

return AccessEvaluation.model_validate(response)
2 changes: 1 addition & 1 deletion src/workos/types/authorization/resource.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
from typing import Literal, Optional
from typing import Any, Literal, Mapping, Optional

from workos.types.workos_model import WorkOSModel

Expand Down
Loading
Loading