spaire-sdk
0.28.4Spaire SDK for Python
License Sources
| Source | License | Class |
|---|---|---|
Licensie (detected) | Pending | - |
PyPI (reported) | Not reported | - |
License detection is still in progress for this version.
Loading dependencies…
License File
"""Code generated by Speakeasy (https://speakeasy.com). DO NOT EDIT."""
from .basesdk import BaseSDK
from jsonpath import JSONPath
from spaire_sdk import models, utils
from spaire_sdk._hooks import HookContext
from spaire_sdk.types import BaseModel, OptionalNullable, UNSET
from spaire_sdk.utils.unmarshal_json_response import unmarshal_json_response
from typing import Any, Dict, List, Mapping, Optional, Union, cast
class LicenseKeys(BaseSDK):
def list(
self,
*,
organization_id: OptionalNullable[
Union[
models.LicenseKeysListQueryParamOrganizationIDFilter,
models.LicenseKeysListQueryParamOrganizationIDFilterTypedDict,
]
] = UNSET,
benefit_id: OptionalNullable[
Union[
models.QueryParamBenefitIDFilter,
models.QueryParamBenefitIDFilterTypedDict,
]
] = UNSET,
page: Optional[int] = 1,
limit: Optional[int] = 10,
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> Optional[models.LicenseKeysListResponse]:
r"""List License Keys
Get license keys connected to the given organization & filters.
**Scopes**: `license_keys:read` `license_keys:write`
:param organization_id: Filter by organization ID.
:param benefit_id: Filter by benefit ID.
:param page: Page number, defaults to 1.
:param limit: Size of a page, defaults to 10. Maximum is 100.
:param retries: Override the default retry configuration for this method
:param server_url: Override the default server URL for this method
:param timeout_ms: Override the default request timeout configuration for this method in milliseconds
:param http_headers: Additional headers to set or replace on requests.
"""
base_url = None
url_variables = None
if timeout_ms is None:
timeout_ms = self.sdk_configuration.timeout_ms
if server_url is not None:
base_url = server_url
else:
base_url = self._get_url(base_url, url_variables)
request = models.LicenseKeysListRequest(
organization_id=organization_id,
benefit_id=benefit_id,
page=page,
limit=limit,
)
req = self._build_request(
method="GET",
path="/v1/license-keys/",
base_url=base_url,
url_variables=url_variables,
request=request,
request_body_required=False,
request_has_path_params=False,
request_has_query_params=True,
user_agent_header="user-agent",
accept_header_value="application/json",
http_headers=http_headers,
security=self.sdk_configuration.security,
timeout_ms=timeout_ms,
)
if retries == UNSET:
if self.sdk_configuration.retry_config is not UNSET:
retries = self.sdk_configuration.retry_config
retry_config = None
if isinstance(retries, utils.RetryConfig):
retry_config = (retries, ["429", "500", "502", "503", "504"])
http_res = self.do_request(
hook_ctx=HookContext(
config=self.sdk_configuration,
base_url=base_url or "",
operation_id="license_keys:list",
oauth2_scopes=None,
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["401", "404", "422", "4XX", "5XX"],
retry_config=retry_config,
)
def next_func() -> Optional[models.LicenseKeysListResponse]:
body = utils.unmarshal_json(http_res.text, Union[Dict[Any, Any], List[Any]])
page = request.page if not request.page is None else 1
next_page = page + 1
num_pages = JSONPath("$.pagination.max_page").parse(body)
if len(num_pages) == 0 or num_pages[0] <= page:
return None
if not http_res.text:
return None
results = JSONPath("$.items").parse(body)
if len(results) == 0 or len(results[0]) == 0:
return None
limit = request.limit if not request.limit is None else 10
if len(results[0]) < limit:
return None
return self.list(
organization_id=organization_id,
benefit_id=benefit_id,
page=next_page,
limit=limit,
retries=retries,
)
response_data: Any = None
if utils.match_response(http_res, "200", "application/json"):
return models.LicenseKeysListResponse(
result=unmarshal_json_response(
models.ListResourceLicenseKeyRead, http_res
),
next=next_func,
)
if utils.match_response(http_res, "401", "application/json"):
response_data = unmarshal_json_response(models.UnauthorizedData, http_res)
raise models.Unauthorized(response_data, http_res)
if utils.match_response(http_res, "404", "application/json"):
response_data = unmarshal_json_response(
models.ResourceNotFoundData, http_res
)
raise models.ResourceNotFound(response_data, http_res)
if utils.match_response(http_res, "422", "application/json"):
response_data = unmarshal_json_response(
models.HTTPValidationErrorData, http_res
)
raise models.HTTPValidationError(response_data, http_res)
if utils.match_response(http_res, "4XX", "*"):
http_res_text = utils.stream_to_text(http_res)
raise models.SDKError("API error occurred", http_res, http_res_text)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = utils.stream_to_text(http_res)
raise models.SDKError("API error occurred", http_res, http_res_text)
raise models.SDKError("Unexpected response received", http_res)
async def list_async(
self,
*,
organization_id: OptionalNullable[
Union[
models.LicenseKeysListQueryParamOrganizationIDFilter,
models.LicenseKeysListQueryParamOrganizationIDFilterTypedDict,
]
] = UNSET,
benefit_id: OptionalNullable[
Union[
models.QueryParamBenefitIDFilter,
models.QueryParamBenefitIDFilterTypedDict,
]
] = UNSET,
page: Optional[int] = 1,
limit: Optional[int] = 10,
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> Optional[models.LicenseKeysListResponse]:
r"""List License Keys
Get license keys connected to the given organization & filters.
**Scopes**: `license_keys:read` `license_keys:write`
:param organization_id: Filter by organization ID.
:param benefit_id: Filter by benefit ID.
:param page: Page number, defaults to 1.
:param limit: Size of a page, defaults to 10. Maximum is 100.
:param retries: Override the default retry configuration for this method
:param server_url: Override the default server URL for this method
:param timeout_ms: Override the default request timeout configuration for this method in milliseconds
:param http_headers: Additional headers to set or replace on requests.
"""
base_url = None
url_variables = None
if timeout_ms is None:
timeout_ms = self.sdk_configuration.timeout_ms
if server_url is not None:
base_url = server_url
else:
base_url = self._get_url(base_url, url_variables)
request = models.LicenseKeysListRequest(
organization_id=organization_id,
benefit_id=benefit_id,
page=page,
limit=limit,
)
req = self._build_request_async(
method="GET",
path="/v1/license-keys/",
base_url=base_url,
url_variables=url_variables,
request=request,
request_body_required=False,
request_has_path_params=False,
request_has_query_params=True,
user_agent_header="user-agent",
accept_header_value="application/json",
http_headers=http_headers,
security=self.sdk_configuration.security,
timeout_ms=timeout_ms,
)
if retries == UNSET:
if self.sdk_configuration.retry_config is not UNSET:
retries = self.sdk_configuration.retry_config
retry_config = None
if isinstance(retries, utils.RetryConfig):
retry_config = (retries, ["429", "500", "502", "503", "504"])
http_res = await self.do_request_async(
hook_ctx=HookContext(
config=self.sdk_configuration,
base_url=base_url or "",
operation_id="license_keys:list",
oauth2_scopes=None,
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["401", "404", "422", "4XX", "5XX"],
retry_config=retry_config,
)
def next_func() -> Optional[models.LicenseKeysListResponse]:
body = utils.unmarshal_json(http_res.text, Union[Dict[Any, Any], List[Any]])
page = request.page if not request.page is None else 1
next_page = page + 1
num_pages = JSONPath("$.pagination.max_page").parse(body)
if len(num_pages) == 0 or num_pages[0] <= page:
return None
if not http_res.text:
return None
results = JSONPath("$.items").parse(body)
if len(results) == 0 or len(results[0]) == 0:
return None
limit = request.limit if not request.limit is None else 10
if len(results[0]) < limit:
return None
return self.list(
organization_id=organization_id,
benefit_id=benefit_id,
page=next_page,
limit=limit,
retries=retries,
)
response_data: Any = None
if utils.match_response(http_res, "200", "application/json"):
return models.LicenseKeysListResponse(
result=unmarshal_json_response(
models.ListResourceLicenseKeyRead, http_res
),
next=next_func,
)
if utils.match_response(http_res, "401", "application/json"):
response_data = unmarshal_json_response(models.UnauthorizedData, http_res)
raise models.Unauthorized(response_data, http_res)
if utils.match_response(http_res, "404", "application/json"):
response_data = unmarshal_json_response(
models.ResourceNotFoundData, http_res
)
raise models.ResourceNotFound(response_data, http_res)
if utils.match_response(http_res, "422", "application/json"):
response_data = unmarshal_json_response(
models.HTTPValidationErrorData, http_res
)
raise models.HTTPValidationError(response_data, http_res)
if utils.match_response(http_res, "4XX", "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise models.SDKError("API error occurred", http_res, http_res_text)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise models.SDKError("API error occurred", http_res, http_res_text)
raise models.SDKError("Unexpected response received", http_res)
def get(
self,
*,
id: str,
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> models.LicenseKeyWithActivations:
r"""Get License Key
Get a license key.
**Scopes**: `license_keys:read` `license_keys:write`
:param id:
:param retries: Override the default retry configuration for this method
:param server_url: Override the default server URL for this method
:param timeout_ms: Override the default request timeout configuration for this method in milliseconds
:param http_headers: Additional headers to set or replace on requests.
"""
base_url = None
url_variables = None
if timeout_ms is None:
timeout_ms = self.sdk_configuration.timeout_ms
if server_url is not None:
base_url = server_url
else:
base_url = self._get_url(base_url, url_variables)
request = models.LicenseKeysGetRequest(
id=id,
)
req = self._build_request(
method="GET",
path="/v1/license-keys/{id}",
base_url=base_url,
url_variables=url_variables,
request=request,
request_body_required=False,
request_has_path_params=True,
request_has_query_params=True,
user_agent_header="user-agent",
accept_header_value="application/json",
http_headers=http_headers,
security=self.sdk_configuration.security,
timeout_ms=timeout_ms,
)
if retries == UNSET:
if self.sdk_configuration.retry_config is not UNSET:
retries = self.sdk_configuration.retry_config
retry_config = None
if isinstance(retries, utils.RetryConfig):
retry_config = (retries, ["429", "500", "502", "503", "504"])
http_res = self.do_request(
hook_ctx=HookContext(
config=self.sdk_configuration,
base_url=base_url or "",
operation_id="license_keys:get",
oauth2_scopes=None,
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["401", "404", "422", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "application/json"):
return unmarshal_json_response(models.LicenseKeyWithActivations, http_res)
if utils.match_response(http_res, "401", "application/json"):
response_data = unmarshal_json_response(models.UnauthorizedData, http_res)
raise models.Unauthorized(response_data, http_res)
if utils.match_response(http_res, "404", "application/json"):
response_data = unmarshal_json_response(
models.ResourceNotFoundData, http_res
)
raise models.ResourceNotFound(response_data, http_res)
if utils.match_response(http_res, "422", "application/json"):
response_data = unmarshal_json_response(
models.HTTPValidationErrorData, http_res
)
raise models.HTTPValidationError(response_data, http_res)
if utils.match_response(http_res, "4XX", "*"):
http_res_text = utils.stream_to_text(http_res)
raise models.SDKError("API error occurred", http_res, http_res_text)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = utils.stream_to_text(http_res)
raise models.SDKError("API error occurred", http_res, http_res_text)
raise models.SDKError("Unexpected response received", http_res)
async def get_async(
self,
*,
id: str,
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> models.LicenseKeyWithActivations:
r"""Get License Key
Get a license key.
**Scopes**: `license_keys:read` `license_keys:write`
:param id:
:param retries: Override the default retry configuration for this method
:param server_url: Override the default server URL for this method
:param timeout_ms: Override the default request timeout configuration for this method in milliseconds
:param http_headers: Additional headers to set or replace on requests.
"""
base_url = None
url_variables = None
if timeout_ms is None:
timeout_ms = self.sdk_configuration.timeout_ms
if server_url is not None:
base_url = server_url
else:
base_url = self._get_url(base_url, url_variables)
request = models.LicenseKeysGetRequest(
id=id,
)
req = self._build_request_async(
method="GET",
path="/v1/license-keys/{id}",
base_url=base_url,
url_variables=url_variables,
request=request,
request_body_required=False,
request_has_path_params=True,
request_has_query_params=True,
user_agent_header="user-agent",
accept_header_value="application/json",
http_headers=http_headers,
security=self.sdk_configuration.security,
timeout_ms=timeout_ms,
)
if retries == UNSET:
if self.sdk_configuration.retry_config is not UNSET:
retries = self.sdk_configuration.retry_config
retry_config = None
if isinstance(retries, utils.RetryConfig):
retry_config = (retries, ["429", "500", "502", "503", "504"])
http_res = await self.do_request_async(
hook_ctx=HookContext(
config=self.sdk_configuration,
base_url=base_url or "",
operation_id="license_keys:get",
oauth2_scopes=None,
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["401", "404", "422", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "application/json"):
return unmarshal_json_response(models.LicenseKeyWithActivations, http_res)
if utils.match_response(http_res, "401", "application/json"):
response_data = unmarshal_json_response(models.UnauthorizedData, http_res)
raise models.Unauthorized(response_data, http_res)
if utils.match_response(http_res, "404", "application/json"):
response_data = unmarshal_json_response(
models.ResourceNotFoundData, http_res
)
raise models.ResourceNotFound(response_data, http_res)
if utils.match_response(http_res, "422", "application/json"):
response_data = unmarshal_json_response(
models.HTTPValidationErrorData, http_res
)
raise models.HTTPValidationError(response_data, http_res)
if utils.match_response(http_res, "4XX", "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise models.SDKError("API error occurred", http_res, http_res_text)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise models.SDKError("API error occurred", http_res, http_res_text)
raise models.SDKError("Unexpected response received", http_res)
def update(
self,
*,
id: str,
license_key_update: Union[
models.LicenseKeyUpdate, models.LicenseKeyUpdateTypedDict
],
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> models.LicenseKeyRead:
r"""Update License Key
Update a license key.
**Scopes**: `license_keys:write`
:param id:
:param license_key_update:
:param retries: Override the default retry configuration for this method
:param server_url: Override the default server URL for this method
:param timeout_ms: Override the default request timeout configuration for this method in milliseconds
:param http_headers: Additional headers to set or replace on requests.
"""
base_url = None
url_variables = None
if timeout_ms is None:
timeout_ms = self.sdk_configuration.timeout_ms
if server_url is not None:
base_url = server_url
else:
base_url = self._get_url(base_url, url_variables)
request = models.LicenseKeysUpdateRequest(
id=id,
license_key_update=utils.get_pydantic_model(
license_key_update, models.LicenseKeyUpdate
),
)
req = self._build_request(
method="PATCH",
path="/v1/license-keys/{id}",
base_url=base_url,
url_variables=url_variables,
request=request,
request_body_required=True,
request_has_path_params=True,
request_has_query_params=True,
user_agent_header="user-agent",
accept_header_value="application/json",
http_headers=http_headers,
security=self.sdk_configuration.security,
get_serialized_body=lambda: utils.serialize_request_body(
request.license_key_update,
False,
False,
"json",
models.LicenseKeyUpdate,
),
timeout_ms=timeout_ms,
)
if retries == UNSET:
if self.sdk_configuration.retry_config is not UNSET:
retries = self.sdk_configuration.retry_config
retry_config = None
if isinstance(retries, utils.RetryConfig):
retry_config = (retries, ["429", "500", "502", "503", "504"])
http_res = self.do_request(
hook_ctx=HookContext(
config=self.sdk_configuration,
base_url=base_url or "",
operation_id="license_keys:update",
oauth2_scopes=None,
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["401", "404", "422", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "application/json"):
return unmarshal_json_response(models.LicenseKeyRead, http_res)
if utils.match_response(http_res, "401", "application/json"):
response_data = unmarshal_json_response(models.UnauthorizedData, http_res)
raise models.Unauthorized(response_data, http_res)
if utils.match_response(http_res, "404", "application/json"):
response_data = unmarshal_json_response(
models.ResourceNotFoundData, http_res
)
raise models.ResourceNotFound(response_data, http_res)
if utils.match_response(http_res, "422", "application/json"):
response_data = unmarshal_json_response(
models.HTTPValidationErrorData, http_res
)
raise models.HTTPValidationError(response_data, http_res)
if utils.match_response(http_res, "4XX", "*"):
http_res_text = utils.stream_to_text(http_res)
raise models.SDKError("API error occurred", http_res, http_res_text)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = utils.stream_to_text(http_res)
raise models.SDKError("API error occurred", http_res, http_res_text)
raise models.SDKError("Unexpected response received", http_res)
async def update_async(
self,
*,
id: str,
license_key_update: Union[
models.LicenseKeyUpdate, models.LicenseKeyUpdateTypedDict
],
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> models.LicenseKeyRead:
r"""Update License Key
Update a license key.
**Scopes**: `license_keys:write`
:param id:
:param license_key_update:
:param retries: Override the default retry configuration for this method
:param server_url: Override the default server URL for this method
:param timeout_ms: Override the default request timeout configuration for this method in milliseconds
:param http_headers: Additional headers to set or replace on requests.
"""
base_url = None
url_variables = None
if timeout_ms is None:
timeout_ms = self.sdk_configuration.timeout_ms
if server_url is not None:
base_url = server_url
else:
base_url = self._get_url(base_url, url_variables)
request = models.LicenseKeysUpdateRequest(
id=id,
license_key_update=utils.get_pydantic_model(
license_key_update, models.LicenseKeyUpdate
),
)
req = self._build_request_async(
method="PATCH",
path="/v1/license-keys/{id}",
base_url=base_url,
url_variables=url_variables,
request=request,
request_body_required=True,
request_has_path_params=True,
request_has_query_params=True,
user_agent_header="user-agent",
accept_header_value="application/json",
http_headers=http_headers,
security=self.sdk_configuration.security,
get_serialized_body=lambda: utils.serialize_request_body(
request.license_key_update,
False,
False,
"json",
models.LicenseKeyUpdate,
),
timeout_ms=timeout_ms,
)
if retries == UNSET:
if self.sdk_configuration.retry_config is not UNSET:
retries = self.sdk_configuration.retry_config
retry_config = None
if isinstance(retries, utils.RetryConfig):
retry_config = (retries, ["429", "500", "502", "503", "504"])
http_res = await self.do_request_async(
hook_ctx=HookContext(
config=self.sdk_configuration,
base_url=base_url or "",
operation_id="license_keys:update",
oauth2_scopes=None,
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["401", "404", "422", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "application/json"):
return unmarshal_json_response(models.LicenseKeyRead, http_res)
if utils.match_response(http_res, "401", "application/json"):
response_data = unmarshal_json_response(models.UnauthorizedData, http_res)
raise models.Unauthorized(response_data, http_res)
if utils.match_response(http_res, "404", "application/json"):
response_data = unmarshal_json_response(
models.ResourceNotFoundData, http_res
)
raise models.ResourceNotFound(response_data, http_res)
if utils.match_response(http_res, "422", "application/json"):
response_data = unmarshal_json_response(
models.HTTPValidationErrorData, http_res
)
raise models.HTTPValidationError(response_data, http_res)
if utils.match_response(http_res, "4XX", "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise models.SDKError("API error occurred", http_res, http_res_text)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise models.SDKError("API error occurred", http_res, http_res_text)
raise models.SDKError("Unexpected response received", http_res)
def get_activation(
self,
*,
id: str,
activation_id: str,
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> models.LicenseKeyActivationRead:
r"""Get Activation
Get a license key activation.
**Scopes**: `license_keys:read` `license_keys:write`
:param id:
:param activation_id:
:param retries: Override the default retry configuration for this method
:param server_url: Override the default server URL for this method
:param timeout_ms: Override the default request timeout configuration for this method in milliseconds
:param http_headers: Additional headers to set or replace on requests.
"""
base_url = None
url_variables = None
if timeout_ms is None:
timeout_ms = self.sdk_configuration.timeout_ms
if server_url is not None:
base_url = server_url
else:
base_url = self._get_url(base_url, url_variables)
request = models.LicenseKeysGetActivationRequest(
id=id,
activation_id=activation_id,
)
req = self._build_request(
method="GET",
path="/v1/license-keys/{id}/activations/{activation_id}",
base_url=base_url,
url_variables=url_variables,
request=request,
request_body_required=False,
request_has_path_params=True,
request_has_query_params=True,
user_agent_header="user-agent",
accept_header_value="application/json",
http_headers=http_headers,
security=self.sdk_configuration.security,
timeout_ms=timeout_ms,
)
if retries == UNSET:
if self.sdk_configuration.retry_config is not UNSET:
retries = self.sdk_configuration.retry_config
retry_config = None
if isinstance(retries, utils.RetryConfig):
retry_config = (retries, ["429", "500", "502", "503", "504"])
http_res = self.do_request(
hook_ctx=HookContext(
config=self.sdk_configuration,
base_url=base_url or "",
operation_id="license_keys:get_activation",
oauth2_scopes=None,
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["401", "404", "422", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "application/json"):
return unmarshal_json_response(models.LicenseKeyActivationRead, http_res)
if utils.match_response(http_res, "401", "application/json"):
response_data = unmarshal_json_response(models.UnauthorizedData, http_res)
raise models.Unauthorized(response_data, http_res)
if utils.match_response(http_res, "404", "application/json"):
response_data = unmarshal_json_response(
models.ResourceNotFoundData, http_res
)
raise models.ResourceNotFound(response_data, http_res)
if utils.match_response(http_res, "422", "application/json"):
response_data = unmarshal_json_response(
models.HTTPValidationErrorData, http_res
)
raise models.HTTPValidationError(response_data, http_res)
if utils.match_response(http_res, "4XX", "*"):
http_res_text = utils.stream_to_text(http_res)
raise models.SDKError("API error occurred", http_res, http_res_text)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = utils.stream_to_text(http_res)
raise models.SDKError("API error occurred", http_res, http_res_text)
raise models.SDKError("Unexpected response received", http_res)
async def get_activation_async(
self,
*,
id: str,
activation_id: str,
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> models.LicenseKeyActivationRead:
r"""Get Activation
Get a license key activation.
**Scopes**: `license_keys:read` `license_keys:write`
:param id:
:param activation_id:
:param retries: Override the default retry configuration for this method
:param server_url: Override the default server URL for this method
:param timeout_ms: Override the default request timeout configuration for this method in milliseconds
:param http_headers: Additional headers to set or replace on requests.
"""
base_url = None
url_variables = None
if timeout_ms is None:
timeout_ms = self.sdk_configuration.timeout_ms
if server_url is not None:
base_url = server_url
else:
base_url = self._get_url(base_url, url_variables)
request = models.LicenseKeysGetActivationRequest(
id=id,
activation_id=activation_id,
)
req = self._build_request_async(
method="GET",
path="/v1/license-keys/{id}/activations/{activation_id}",
base_url=base_url,
url_variables=url_variables,
request=request,
request_body_required=False,
request_has_path_params=True,
request_has_query_params=True,
user_agent_header="user-agent",
accept_header_value="application/json",
http_headers=http_headers,
security=self.sdk_configuration.security,
timeout_ms=timeout_ms,
)
if retries == UNSET:
if self.sdk_configuration.retry_config is not UNSET:
retries = self.sdk_configuration.retry_config
retry_config = None
if isinstance(retries, utils.RetryConfig):
retry_config = (retries, ["429", "500", "502", "503", "504"])
http_res = await self.do_request_async(
hook_ctx=HookContext(
config=self.sdk_configuration,
base_url=base_url or "",
operation_id="license_keys:get_activation",
oauth2_scopes=None,
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["401", "404", "422", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "application/json"):
return unmarshal_json_response(models.LicenseKeyActivationRead, http_res)
if utils.match_response(http_res, "401", "application/json"):
response_data = unmarshal_json_response(models.UnauthorizedData, http_res)
raise models.Unauthorized(response_data, http_res)
if utils.match_response(http_res, "404", "application/json"):
response_data = unmarshal_json_response(
models.ResourceNotFoundData, http_res
)
raise models.ResourceNotFound(response_data, http_res)
if utils.match_response(http_res, "422", "application/json"):
response_data = unmarshal_json_response(
models.HTTPValidationErrorData, http_res
)
raise models.HTTPValidationError(response_data, http_res)
if utils.match_response(http_res, "4XX", "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise models.SDKError("API error occurred", http_res, http_res_text)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise models.SDKError("API error occurred", http_res, http_res_text)
raise models.SDKError("Unexpected response received", http_res)
def validate(
self,
*,
request: Union[models.LicenseKeyValidate, models.LicenseKeyValidateTypedDict],
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> models.ValidatedLicenseKey:
r"""Validate License Key
Validate a license key.
**Scopes**: `license_keys:write`
:param request: The request object to send.
:param retries: Override the default retry configuration for this method
:param server_url: Override the default server URL for this method
:param timeout_ms: Override the default request timeout configuration for this method in milliseconds
:param http_headers: Additional headers to set or replace on requests.
"""
base_url = None
url_variables = None
if timeout_ms is None:
timeout_ms = self.sdk_configuration.timeout_ms
if server_url is not None:
base_url = server_url
else:
base_url = self._get_url(base_url, url_variables)
if not isinstance(request, BaseModel):
request = utils.unmarshal(request, models.LicenseKeyValidate)
request = cast(models.LicenseKeyValidate, request)
req = self._build_request(
method="POST",
path="/v1/license-keys/validate",
base_url=base_url,
url_variables=url_variables,
request=request,
request_body_required=True,
request_has_path_params=False,
request_has_query_params=True,
user_agent_header="user-agent",
accept_header_value="application/json",
http_headers=http_headers,
security=self.sdk_configuration.security,
get_serialized_body=lambda: utils.serialize_request_body(
request, False, False, "json", models.LicenseKeyValidate
),
timeout_ms=timeout_ms,
)
if retries == UNSET:
if self.sdk_configuration.retry_config is not UNSET:
retries = self.sdk_configuration.retry_config
retry_config = None
if isinstance(retries, utils.RetryConfig):
retry_config = (retries, ["429", "500", "502", "503", "504"])
http_res = self.do_request(
hook_ctx=HookContext(
config=self.sdk_configuration,
base_url=base_url or "",
operation_id="license_keys:validate",
oauth2_scopes=None,
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["404", "422", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "application/json"):
return unmarshal_json_response(models.ValidatedLicenseKey, http_res)
if utils.match_response(http_res, "404", "application/json"):
response_data = unmarshal_json_response(
models.ResourceNotFoundData, http_res
)
raise models.ResourceNotFound(response_data, http_res)
if utils.match_response(http_res, "422", "application/json"):
response_data = unmarshal_json_response(
models.HTTPValidationErrorData, http_res
)
raise models.HTTPValidationError(response_data, http_res)
if utils.match_response(http_res, "4XX", "*"):
http_res_text = utils.stream_to_text(http_res)
raise models.SDKError("API error occurred", http_res, http_res_text)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = utils.stream_to_text(http_res)
raise models.SDKError("API error occurred", http_res, http_res_text)
raise models.SDKError("Unexpected response received", http_res)
async def validate_async(
self,
*,
request: Union[models.LicenseKeyValidate, models.LicenseKeyValidateTypedDict],
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> models.ValidatedLicenseKey:
r"""Validate License Key
Validate a license key.
**Scopes**: `license_keys:write`
:param request: The request object to send.
:param retries: Override the default retry configuration for this method
:param server_url: Override the default server URL for this method
:param timeout_ms: Override the default request timeout configuration for this method in milliseconds
:param http_headers: Additional headers to set or replace on requests.
"""
base_url = None
url_variables = None
if timeout_ms is None:
timeout_ms = self.sdk_configuration.timeout_ms
if server_url is not None:
base_url = server_url
else:
base_url = self._get_url(base_url, url_variables)
if not isinstance(request, BaseModel):
request = utils.unmarshal(request, models.LicenseKeyValidate)
request = cast(models.LicenseKeyValidate, request)
req = self._build_request_async(
method="POST",
path="/v1/license-keys/validate",
base_url=base_url,
url_variables=url_variables,
request=request,
request_body_required=True,
request_has_path_params=False,
request_has_query_params=True,
user_agent_header="user-agent",
accept_header_value="application/json",
http_headers=http_headers,
security=self.sdk_configuration.security,
get_serialized_body=lambda: utils.serialize_request_body(
request, False, False, "json", models.LicenseKeyValidate
),
timeout_ms=timeout_ms,
)
if retries == UNSET:
if self.sdk_configuration.retry_config is not UNSET:
retries = self.sdk_configuration.retry_config
retry_config = None
if isinstance(retries, utils.RetryConfig):
retry_config = (retries, ["429", "500", "502", "503", "504"])
http_res = await self.do_request_async(
hook_ctx=HookContext(
config=self.sdk_configuration,
base_url=base_url or "",
operation_id="license_keys:validate",
oauth2_scopes=None,
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["404", "422", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "application/json"):
return unmarshal_json_response(models.ValidatedLicenseKey, http_res)
if utils.match_response(http_res, "404", "application/json"):
response_data = unmarshal_json_response(
models.ResourceNotFoundData, http_res
)
raise models.ResourceNotFound(response_data, http_res)
if utils.match_response(http_res, "422", "application/json"):
response_data = unmarshal_json_response(
models.HTTPValidationErrorData, http_res
)
raise models.HTTPValidationError(response_data, http_res)
if utils.match_response(http_res, "4XX", "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise models.SDKError("API error occurred", http_res, http_res_text)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise models.SDKError("API error occurred", http_res, http_res_text)
raise models.SDKError("Unexpected response received", http_res)
def activate(
self,
*,
request: Union[models.LicenseKeyActivate, models.LicenseKeyActivateTypedDict],
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> models.LicenseKeyActivationRead:
r"""Activate License Key
Activate a license key instance.
**Scopes**: `license_keys:write`
:param request: The request object to send.
:param retries: Override the default retry configuration for this method
:param server_url: Override the default server URL for this method
:param timeout_ms: Override the default request timeout configuration for this method in milliseconds
:param http_headers: Additional headers to set or replace on requests.
"""
base_url = None
url_variables = None
if timeout_ms is None:
timeout_ms = self.sdk_configuration.timeout_ms
if server_url is not None:
base_url = server_url
else:
base_url = self._get_url(base_url, url_variables)
if not isinstance(request, BaseModel):
request = utils.unmarshal(request, models.LicenseKeyActivate)
request = cast(models.LicenseKeyActivate, request)
req = self._build_request(
method="POST",
path="/v1/license-keys/activate",
base_url=base_url,
url_variables=url_variables,
request=request,
request_body_required=True,
request_has_path_params=False,
request_has_query_params=True,
user_agent_header="user-agent",
accept_header_value="application/json",
http_headers=http_headers,
security=self.sdk_configuration.security,
get_serialized_body=lambda: utils.serialize_request_body(
request, False, False, "json", models.LicenseKeyActivate
),
timeout_ms=timeout_ms,
)
if retries == UNSET:
if self.sdk_configuration.retry_config is not UNSET:
retries = self.sdk_configuration.retry_config
retry_config = None
if isinstance(retries, utils.RetryConfig):
retry_config = (retries, ["429", "500", "502", "503", "504"])
http_res = self.do_request(
hook_ctx=HookContext(
config=self.sdk_configuration,
base_url=base_url or "",
operation_id="license_keys:activate",
oauth2_scopes=None,
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["403", "404", "422", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "application/json"):
return unmarshal_json_response(models.LicenseKeyActivationRead, http_res)
if utils.match_response(http_res, "403", "application/json"):
response_data = unmarshal_json_response(models.NotPermittedData, http_res)
raise models.NotPermitted(response_data, http_res)
if utils.match_response(http_res, "404", "application/json"):
response_data = unmarshal_json_response(
models.ResourceNotFoundData, http_res
)
raise models.ResourceNotFound(response_data, http_res)
if utils.match_response(http_res, "422", "application/json"):
response_data = unmarshal_json_response(
models.HTTPValidationErrorData, http_res
)
raise models.HTTPValidationError(response_data, http_res)
if utils.match_response(http_res, "4XX", "*"):
http_res_text = utils.stream_to_text(http_res)
raise models.SDKError("API error occurred", http_res, http_res_text)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = utils.stream_to_text(http_res)
raise models.SDKError("API error occurred", http_res, http_res_text)
raise models.SDKError("Unexpected response received", http_res)
async def activate_async(
self,
*,
request: Union[models.LicenseKeyActivate, models.LicenseKeyActivateTypedDict],
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
) -> models.LicenseKeyActivationRead:
r"""Activate License Key
Activate a license key instance.
**Scopes**: `license_keys:write`
:param request: The request object to send.
:param retries: Override the default retry configuration for this method
:param server_url: Override the default server URL for this method
:param timeout_ms: Override the default request timeout configuration for this method in milliseconds
:param http_headers: Additional headers to set or replace on requests.
"""
base_url = None
url_variables = None
if timeout_ms is None:
timeout_ms = self.sdk_configuration.timeout_ms
if server_url is not None:
base_url = server_url
else:
base_url = self._get_url(base_url, url_variables)
if not isinstance(request, BaseModel):
request = utils.unmarshal(request, models.LicenseKeyActivate)
request = cast(models.LicenseKeyActivate, request)
req = self._build_request_async(
method="POST",
path="/v1/license-keys/activate",
base_url=base_url,
url_variables=url_variables,
request=request,
request_body_required=True,
request_has_path_params=False,
request_has_query_params=True,
user_agent_header="user-agent",
accept_header_value="application/json",
http_headers=http_headers,
security=self.sdk_configuration.security,
get_serialized_body=lambda: utils.serialize_request_body(
request, False, False, "json", models.LicenseKeyActivate
),
timeout_ms=timeout_ms,
)
if retries == UNSET:
if self.sdk_configuration.retry_config is not UNSET:
retries = self.sdk_configuration.retry_config
retry_config = None
if isinstance(retries, utils.RetryConfig):
retry_config = (retries, ["429", "500", "502", "503", "504"])
http_res = await self.do_request_async(
hook_ctx=HookContext(
config=self.sdk_configuration,
base_url=base_url or "",
operation_id="license_keys:activate",
oauth2_scopes=None,
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["403", "404", "422", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "200", "application/json"):
return unmarshal_json_response(models.LicenseKeyActivationRead, http_res)
if utils.match_response(http_res, "403", "application/json"):
response_data = unmarshal_json_response(models.NotPermittedData, http_res)
raise models.NotPermitted(response_data, http_res)
if utils.match_response(http_res, "404", "application/json"):
response_data = unmarshal_json_response(
models.ResourceNotFoundData, http_res
)
raise models.ResourceNotFound(response_data, http_res)
if utils.match_response(http_res, "422", "application/json"):
response_data = unmarshal_json_response(
models.HTTPValidationErrorData, http_res
)
raise models.HTTPValidationError(response_data, http_res)
if utils.match_response(http_res, "4XX", "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise models.SDKError("API error occurred", http_res, http_res_text)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise models.SDKError("API error occurred", http_res, http_res_text)
raise models.SDKError("Unexpected response received", http_res)
def deactivate(
self,
*,
request: Union[
models.LicenseKeyDeactivate, models.LicenseKeyDeactivateTypedDict
],
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
):
r"""Deactivate License Key
Deactivate a license key instance.
**Scopes**: `license_keys:write`
:param request: The request object to send.
:param retries: Override the default retry configuration for this method
:param server_url: Override the default server URL for this method
:param timeout_ms: Override the default request timeout configuration for this method in milliseconds
:param http_headers: Additional headers to set or replace on requests.
"""
base_url = None
url_variables = None
if timeout_ms is None:
timeout_ms = self.sdk_configuration.timeout_ms
if server_url is not None:
base_url = server_url
else:
base_url = self._get_url(base_url, url_variables)
if not isinstance(request, BaseModel):
request = utils.unmarshal(request, models.LicenseKeyDeactivate)
request = cast(models.LicenseKeyDeactivate, request)
req = self._build_request(
method="POST",
path="/v1/license-keys/deactivate",
base_url=base_url,
url_variables=url_variables,
request=request,
request_body_required=True,
request_has_path_params=False,
request_has_query_params=True,
user_agent_header="user-agent",
accept_header_value="application/json",
http_headers=http_headers,
security=self.sdk_configuration.security,
get_serialized_body=lambda: utils.serialize_request_body(
request, False, False, "json", models.LicenseKeyDeactivate
),
timeout_ms=timeout_ms,
)
if retries == UNSET:
if self.sdk_configuration.retry_config is not UNSET:
retries = self.sdk_configuration.retry_config
retry_config = None
if isinstance(retries, utils.RetryConfig):
retry_config = (retries, ["429", "500", "502", "503", "504"])
http_res = self.do_request(
hook_ctx=HookContext(
config=self.sdk_configuration,
base_url=base_url or "",
operation_id="license_keys:deactivate",
oauth2_scopes=None,
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["404", "422", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "204", "*"):
return
if utils.match_response(http_res, "404", "application/json"):
response_data = unmarshal_json_response(
models.ResourceNotFoundData, http_res
)
raise models.ResourceNotFound(response_data, http_res)
if utils.match_response(http_res, "422", "application/json"):
response_data = unmarshal_json_response(
models.HTTPValidationErrorData, http_res
)
raise models.HTTPValidationError(response_data, http_res)
if utils.match_response(http_res, "4XX", "*"):
http_res_text = utils.stream_to_text(http_res)
raise models.SDKError("API error occurred", http_res, http_res_text)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = utils.stream_to_text(http_res)
raise models.SDKError("API error occurred", http_res, http_res_text)
raise models.SDKError("Unexpected response received", http_res)
async def deactivate_async(
self,
*,
request: Union[
models.LicenseKeyDeactivate, models.LicenseKeyDeactivateTypedDict
],
retries: OptionalNullable[utils.RetryConfig] = UNSET,
server_url: Optional[str] = None,
timeout_ms: Optional[int] = None,
http_headers: Optional[Mapping[str, str]] = None,
):
r"""Deactivate License Key
Deactivate a license key instance.
**Scopes**: `license_keys:write`
:param request: The request object to send.
:param retries: Override the default retry configuration for this method
:param server_url: Override the default server URL for this method
:param timeout_ms: Override the default request timeout configuration for this method in milliseconds
:param http_headers: Additional headers to set or replace on requests.
"""
base_url = None
url_variables = None
if timeout_ms is None:
timeout_ms = self.sdk_configuration.timeout_ms
if server_url is not None:
base_url = server_url
else:
base_url = self._get_url(base_url, url_variables)
if not isinstance(request, BaseModel):
request = utils.unmarshal(request, models.LicenseKeyDeactivate)
request = cast(models.LicenseKeyDeactivate, request)
req = self._build_request_async(
method="POST",
path="/v1/license-keys/deactivate",
base_url=base_url,
url_variables=url_variables,
request=request,
request_body_required=True,
request_has_path_params=False,
request_has_query_params=True,
user_agent_header="user-agent",
accept_header_value="application/json",
http_headers=http_headers,
security=self.sdk_configuration.security,
get_serialized_body=lambda: utils.serialize_request_body(
request, False, False, "json", models.LicenseKeyDeactivate
),
timeout_ms=timeout_ms,
)
if retries == UNSET:
if self.sdk_configuration.retry_config is not UNSET:
retries = self.sdk_configuration.retry_config
retry_config = None
if isinstance(retries, utils.RetryConfig):
retry_config = (retries, ["429", "500", "502", "503", "504"])
http_res = await self.do_request_async(
hook_ctx=HookContext(
config=self.sdk_configuration,
base_url=base_url or "",
operation_id="license_keys:deactivate",
oauth2_scopes=None,
security_source=self.sdk_configuration.security,
),
request=req,
error_status_codes=["404", "422", "4XX", "5XX"],
retry_config=retry_config,
)
response_data: Any = None
if utils.match_response(http_res, "204", "*"):
return
if utils.match_response(http_res, "404", "application/json"):
response_data = unmarshal_json_response(
models.ResourceNotFoundData, http_res
)
raise models.ResourceNotFound(response_data, http_res)
if utils.match_response(http_res, "422", "application/json"):
response_data = unmarshal_json_response(
models.HTTPValidationErrorData, http_res
)
raise models.HTTPValidationError(response_data, http_res)
if utils.match_response(http_res, "4XX", "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise models.SDKError("API error occurred", http_res, http_res_text)
if utils.match_response(http_res, "5XX", "*"):
http_res_text = await utils.stream_to_text_async(http_res)
raise models.SDKError("API error occurred", http_res, http_res_text)
raise models.SDKError("Unexpected response received", http_res)
Versions
1 version| Version | License | Published | Status |
|---|---|---|---|
| 0.28.4 Latest Viewing | - | Mar 9, 2026 | Pending |