Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
31 changes: 31 additions & 0 deletions descope/management/_outbound_scim_base.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,31 @@
from __future__ import annotations

from typing import Any, Optional


class OutboundSCIMBase:
@staticmethod
def _compose_create_body(
app_id: str,
configuration: Optional[dict] = None,
) -> dict:
body: dict[str, Any] = {
"appId": app_id,
}
if configuration is not None:
body["configuration"] = configuration
return body

@staticmethod
def _compose_update_body(
app_id: str,
version: int,
configuration: Optional[dict] = None,
) -> dict:
body: dict[str, Any] = {
"appId": app_id,
"version": version,
}
if configuration is not None:
body["configuration"] = configuration
return body
7 changes: 7 additions & 0 deletions descope/management/common.py
Original file line number Diff line number Diff line change
Expand Up @@ -128,6 +128,13 @@ class MgmtV1:
outbound_application_batch_upload_user_tokens_path = "/v1/mgmt/outbound/app/user/oauthtoken/batch/upload"
outbound_application_batch_upload_tenant_tokens_path = "/v1/mgmt/outbound/app/tenant/oauthtoken/batch/upload"

# outbound scim
outbound_scim_create_path = "/v1/mgmt/outbound/scim/create"
outbound_scim_update_path = "/v1/mgmt/outbound/scim/update"
outbound_scim_delete_path = "/v1/mgmt/outbound/scim/delete"
outbound_scim_load_path = "/v1/mgmt/outbound/scim"
outbound_scim_set_enabled_path = "/v1/mgmt/outbound/scim/enabled/set"

# engine
engine_create_path = "/v1/mgmt/engine/create"
engine_update_path = "/v1/mgmt/engine/update"
Expand Down
132 changes: 132 additions & 0 deletions descope/management/outbound_scim.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,132 @@
from __future__ import annotations

from typing import Optional

from descope._http_base import HTTPBase
from descope.management._outbound_scim_base import OutboundSCIMBase
from descope.management.common import MgmtV1


class OutboundSCIM(OutboundSCIMBase, HTTPBase):
def create_configuration(
self,
app_id: str,
configuration: Optional[dict] = None,
) -> dict:
"""
Create a new outbound SCIM configuration on the federated SSO application
identified by app_id. The connector name is derived server-side from the app.

Args:
app_id (str): The federated SSO application id this SCIM configuration binds to.
configuration (dict): Optional provider-specific SCIM configuration dictionary.

Return value (dict):
Return dict in the format
{"configuration": {"appId": <app_id>, "configuration": {...},
"enabled": <bool>, "lastExportTime": <int>,
"lastProcessingTime": <int>, "failures": <int>,
"version": <int>}}

Raise:
AuthException: raised if create operation fails
"""
response = self._http.post(
MgmtV1.outbound_scim_create_path,
body=OutboundSCIMBase._compose_create_body(app_id, configuration),
)
return response.json()

def update_configuration(
self,
app_id: str,
version: int,
configuration: Optional[dict] = None,
) -> dict:
"""
Update the outbound SCIM configuration attached to the given federated SSO app.
The version must match the currently stored version — otherwise the update is
rejected as a conflict.

Args:
app_id (str): The federated SSO application id.
version (int): The currently stored version, used for optimistic concurrency.
configuration (dict): Optional updated provider-specific SCIM configuration.

Return value (dict):
Return dict in the format
{"configuration": {"appId": <app_id>, ...}}

Raise:
AuthException: raised if update operation fails
"""
response = self._http.post(
MgmtV1.outbound_scim_update_path,
body=OutboundSCIMBase._compose_update_body(app_id, version, configuration),
)
return response.json()

def delete_configuration(
self,
app_id: str,
):
"""
Delete the outbound SCIM configuration attached to the given federated SSO app.
IMPORTANT: This action is irreversible. Use carefully.

Args:
app_id (str): The federated SSO application id.

Raise:
AuthException: raised if deletion operation fails
"""
self._http.post(MgmtV1.outbound_scim_delete_path, body={"appId": app_id})

def load_configuration(
self,
app_id: str,
) -> dict:
"""
Load the outbound SCIM configuration attached to the given federated SSO app.

Args:
app_id (str): The federated SSO application id.

Return value (dict):
Return dict in the format
{"configuration": {"appId": <app_id>, "configuration": {...},
"enabled": <bool>, "lastExportTime": <int>,
"lastProcessingTime": <int>, "failures": <int>,
"version": <int>}}

Raise:
AuthException: raised if load operation fails
"""
response = self._http.get(f"{MgmtV1.outbound_scim_load_path}/{app_id}")
return response.json()

def set_enabled(
self,
app_id: str,
enabled: bool,
) -> dict:
"""
Enable or disable the outbound SCIM configuration attached to the given
federated SSO app.

Args:
app_id (str): The federated SSO application id.
enabled (bool): Whether the SCIM configuration should be enabled.

Return value (dict):
Return dict in the format
{"configuration": {"appId": <app_id>, "enabled": <bool>, ...}}

Raise:
AuthException: raised if the operation fails
"""
response = self._http.post(
MgmtV1.outbound_scim_set_enabled_path,
body={"appId": app_id, "enabled": enabled},
)
return response.json()
134 changes: 134 additions & 0 deletions descope/management/outbound_scim_async.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,134 @@
from __future__ import annotations

from typing import Optional

from descope._http_base import AsyncHTTPBase
from descope.management._outbound_scim_base import OutboundSCIMBase
from descope.management.common import MgmtV1


class OutboundSCIMAsync(OutboundSCIMBase, AsyncHTTPBase):
"""Async counterpart of OutboundSCIM — all HTTP calls are coroutines."""

async def create_configuration(
self,
app_id: str,
configuration: Optional[dict] = None,
) -> dict:
"""
Create a new outbound SCIM configuration on the federated SSO application
identified by app_id. The connector name is derived server-side from the app.

Args:
app_id (str): The federated SSO application id this SCIM configuration binds to.
configuration (dict): Optional provider-specific SCIM configuration dictionary.

Return value (dict):
Return dict in the format
{"configuration": {"appId": <app_id>, "configuration": {...},
"enabled": <bool>, "lastExportTime": <int>,
"lastProcessingTime": <int>, "failures": <int>,
"version": <int>}}

Raise:
AuthException: raised if create operation fails
"""
response = await self._http.post(
MgmtV1.outbound_scim_create_path,
body=OutboundSCIMBase._compose_create_body(app_id, configuration),
)
return response.json()

async def update_configuration(
self,
app_id: str,
version: int,
configuration: Optional[dict] = None,
) -> dict:
"""
Update the outbound SCIM configuration attached to the given federated SSO app.
The version must match the currently stored version — otherwise the update is
rejected as a conflict.

Args:
app_id (str): The federated SSO application id.
version (int): The currently stored version, used for optimistic concurrency.
configuration (dict): Optional updated provider-specific SCIM configuration.

Return value (dict):
Return dict in the format
{"configuration": {"appId": <app_id>, ...}}

Raise:
AuthException: raised if update operation fails
"""
response = await self._http.post(
MgmtV1.outbound_scim_update_path,
body=OutboundSCIMBase._compose_update_body(app_id, version, configuration),
)
return response.json()

async def delete_configuration(
self,
app_id: str,
):
"""
Delete the outbound SCIM configuration attached to the given federated SSO app.
IMPORTANT: This action is irreversible. Use carefully.

Args:
app_id (str): The federated SSO application id.

Raise:
AuthException: raised if deletion operation fails
"""
await self._http.post(MgmtV1.outbound_scim_delete_path, body={"appId": app_id})

async def load_configuration(
self,
app_id: str,
) -> dict:
"""
Load the outbound SCIM configuration attached to the given federated SSO app.

Args:
app_id (str): The federated SSO application id.

Return value (dict):
Return dict in the format
{"configuration": {"appId": <app_id>, "configuration": {...},
"enabled": <bool>, "lastExportTime": <int>,
"lastProcessingTime": <int>, "failures": <int>,
"version": <int>}}

Raise:
AuthException: raised if load operation fails
"""
response = await self._http.get(f"{MgmtV1.outbound_scim_load_path}/{app_id}")
return response.json()

async def set_enabled(
self,
app_id: str,
enabled: bool,
) -> dict:
"""
Enable or disable the outbound SCIM configuration attached to the given
federated SSO app.

Args:
app_id (str): The federated SSO application id.
enabled (bool): Whether the SCIM configuration should be enabled.

Return value (dict):
Return dict in the format
{"configuration": {"appId": <app_id>, "enabled": <bool>, ...}}

Raise:
AuthException: raised if the operation fails
"""
response = await self._http.post(
MgmtV1.outbound_scim_set_enabled_path,
body={"appId": app_id, "enabled": enabled},
)
return response.json()
7 changes: 7 additions & 0 deletions descope/mgmt.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
OutboundApplication,
OutboundApplicationByToken,
)
from descope.management.outbound_scim import OutboundSCIM
from descope.management.password import Password
from descope.management.permission import Permission
from descope.management.project import Project
Expand Down Expand Up @@ -59,6 +60,7 @@ def __init__(self, http_client: HTTPClient, auth: Auth, fga_cache_url: Optional[
self._management_key = ManagementKey(http_client)
self._outbound_application = OutboundApplication(http_client)
self._outbound_application_by_token = OutboundApplicationByToken(http_client)
self._outbound_scim = OutboundSCIM(http_client)
self._password = Password(http_client)
self._permission = Permission(http_client)
self._project = Project(http_client)
Expand Down Expand Up @@ -163,6 +165,11 @@ def outbound_application_by_token(self):
# No management key check for outbound_app_token (as authentication for those methods is done by inbound app token)
return self._outbound_application_by_token

@property
def outbound_scim(self):
self._ensure_management_key("outbound_scim")
return self._outbound_scim

@property
def descoper(self):
self._ensure_management_key("descoper")
Expand Down
7 changes: 7 additions & 0 deletions descope/mgmt_async.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
OutboundApplicationAsync,
OutboundApplicationByTokenAsync,
)
from descope.management.outbound_scim_async import OutboundSCIMAsync
from descope.management.password_async import PasswordAsync
from descope.management.permission_async import PermissionAsync
from descope.management.project_async import ProjectAsync
Expand Down Expand Up @@ -61,6 +62,7 @@ def __init__(self, http_client: HTTPClientAsync, auth: AuthAsync, fga_cache_url:
self._management_key = ManagementKeyAsync(http_client)
self._outbound_application = OutboundApplicationAsync(http_client)
self._outbound_application_by_token = OutboundApplicationByTokenAsync(http_client)
self._outbound_scim = OutboundSCIMAsync(http_client)
self._password = PasswordAsync(http_client)
self._permission = PermissionAsync(http_client)
self._project = ProjectAsync(http_client)
Expand Down Expand Up @@ -165,6 +167,11 @@ def outbound_application_by_token(self) -> OutboundApplicationByTokenAsync:
# No management key check for outbound_app_token (as authentication for those methods is done by inbound app token)
return self._outbound_application_by_token

@property
def outbound_scim(self) -> OutboundSCIMAsync:
self._ensure_management_key("outbound_scim")
return self._outbound_scim

@property
def descoper(self) -> DescoperAsync:
self._ensure_management_key("descoper")
Expand Down
Loading
Loading