Files

586 lines
18 KiB
Python
Generated

# Code generated by api/dev/generate_billing_client.py from saas.openapi.json. DO NOT EDIT.
# dify-saas commit: 2d358b5a0555440b30d8601f5e786a15e28a2345
from datetime import datetime
from typing import Literal, TypeIs, get_args
from pydantic import Field
from services.billing_client.transport import BillingModel, ServiceErrors, Upstream, request
from services.errors.billing import (
BillingUpstreamError,
BillingUpstreamInvalidResponseError,
BillingUpstreamUnavailableError,
UnknownErrorReason,
UpstreamErrorBody,
)
type AccountErrorReason = Literal["", "CODEC", "VALIDATOR", "UNKNOWN", "INVALID_SECRET_KEY"]
class AddDeleteFeedbackReply(BillingModel):
message: str
class AddDeleteFeedbackRequest(BillingModel):
email: str
feedback: str
class AppNetworkAccessGroupBinding(BillingModel):
id: str
tenant_id: str = Field(alias="tenantId")
app_id: str = Field(alias="appId")
group_id: str | None = Field(None, alias="groupId")
version: int
updated_by_account_id: str = Field(alias="updatedByAccountId")
created_at: datetime = Field(alias="createdAt")
updated_at: datetime = Field(alias="updatedAt")
enabled: bool
access_points: list[str] = Field(alias="accessPoints")
class CheckAccountInFreezeReply(BillingModel):
data: bool
freeze_type: Literal["", "freeze", "email_domain_suspended"] = Field(alias="freezeType")
class CleanupAppNetworkAccessGroupBindingReply(BillingModel):
deleted: bool
class CreateNetworkAccessGroupRequest(BillingModel):
tenant_id: str = Field(alias="tenantId")
name: str
description: str
mode: str
allowed_cidrs: list[str] = Field(alias="allowedCidrs")
actor_account_id: str = Field(alias="actorAccountId")
class DeleteAccountReply(BillingModel):
message: str
class DeleteNetworkAccessGroupReply(BillingModel):
deleted: bool
class DismissNotificationReply(BillingModel):
success: bool
class DismissNotificationReq(BillingModel):
notification_id: str = Field(alias="notificationId")
account_id: str = Field(alias="accountId")
class LangContent(BillingModel):
lang: str
title: str
subtitle: str
body: str
title_pic_url: str = Field(alias="titlePicUrl")
class NotificationPayload(BillingModel):
notification_id: str = Field(alias="notificationId")
contents: dict[str, LangContent]
frequency: Literal["once", "every_page_load"]
is_global: bool = Field(alias="isGlobal")
class GetAccountNotificationReply(BillingModel):
should_show: bool = Field(alias="shouldShow")
notifications: list[NotificationPayload]
class GetAppNetworkAccessGroupReply(BillingModel):
tenant_id: str = Field(alias="tenantId")
app_id: str = Field(alias="appId")
entitled: bool
binding: AppNetworkAccessGroupBinding | None
effective_enabled: bool = Field(alias="effectiveEnabled")
class GetAppNetworkAccessGroupStatusReply(BillingModel):
entitled: bool
configured: bool
enabled: bool
access_points: list[str] = Field(alias="accessPoints")
class NetworkAccessGroup(BillingModel):
id: str
tenant_id: str = Field(alias="tenantId")
name: str
description: str
mode: str
allowed_cidrs: list[str] = Field(alias="allowedCidrs")
version: int
updated_by_account_id: str = Field(alias="updatedByAccountId")
created_at: datetime = Field(alias="createdAt")
updated_at: datetime = Field(alias="updatedAt")
used_by_count: int = Field(alias="usedByCount")
used_by_app_ids: list[str] = Field(alias="usedByAppIds")
enforcing_count: int = Field(alias="enforcingCount")
class ListNetworkAccessGroupsReply(BillingModel):
tenant_id: str = Field(alias="tenantId")
entitled: bool
groups: list[NetworkAccessGroup]
class NetworkAccessGroupReply(BillingModel):
group: NetworkAccessGroup | None
class PutAppNetworkAccessGroupReply(BillingModel):
binding: AppNetworkAccessGroupBinding | None
effective_enabled: bool = Field(alias="effectiveEnabled")
class PutAppNetworkAccessGroupRequest(BillingModel):
tenant_id: str = Field(alias="tenantId")
app_id: str = Field(alias="appId")
group_id: str | None = Field(None, alias="groupId")
expected_version: int = Field(alias="expectedVersion")
actor_account_id: str = Field(alias="actorAccountId")
enabled: bool | None = None
access_points: list[str] = Field(alias="accessPoints")
type SaasNetworkAccessPolicyErrorReason = Literal[
"",
"CODEC",
"VALIDATOR",
"UNKNOWN",
"INVALID_SECRET_KEY",
"INVALID_ARGUMENT",
"INTERNAL_ERROR",
"INVALID_NETWORK_ACCESS_INPUT",
"TENANT_NOT_FOUND",
"NETWORK_ACCESS_GROUP_NOT_FOUND",
"APP_NOT_FOUND",
"NETWORK_ACCESS_APP_STILL_EXISTS",
"TENANT_ARCHIVED",
"NETWORK_ACCESS_FORBIDDEN",
"NETWORK_ACCESS_NOT_ENTITLED",
"NETWORK_ACCESS_VERSION_CONFLICT",
"NETWORK_ACCESS_GROUP_NAME_CONFLICT",
"NETWORK_ACCESS_GROUP_BOUND",
"NETWORK_ACCESS_CACHE_UNAVAILABLE",
]
type SaasNotificationErrorReason = Literal["", "CODEC", "VALIDATOR", "UNKNOWN", "INVALID_SECRET_KEY"]
class TenantUtmReply(BillingModel):
result: str
class TenantUtmRequest(BillingModel):
tenant_id: str = Field(alias="tenantId")
utm_source: str = Field(alias="utmSource")
utm_medium: str = Field(alias="utmMedium")
utm_campaign: str = Field(alias="utmCampaign")
utm_content: str = Field(alias="utmContent")
utm_term: str = Field(alias="utmTerm")
class UpdateNetworkAccessGroupRequest(BillingModel):
tenant_id: str = Field(alias="tenantId")
group_id: str = Field(alias="groupId")
name: str
description: str
mode: str
allowed_cidrs: list[str] = Field(alias="allowedCidrs")
expected_version: int = Field(alias="expectedVersion")
actor_account_id: str = Field(alias="actorAccountId")
type UtmErrorReason = Literal["", "CODEC", "VALIDATOR", "UNKNOWN", "INVALID_SECRET_KEY"]
def _is_account_error_reason(value: str) -> TypeIs[AccountErrorReason]:
return value in get_args(AccountErrorReason.__value__)
class AccountUpstreamError(BillingUpstreamError):
"""A failed Account call, with its reason typed by `AccountErrorReason`."""
def __init__(
self,
*args: object,
status_code: int | None = None,
body: UpstreamErrorBody | None = None,
reason: AccountErrorReason | UnknownErrorReason | None = None,
) -> None:
super().__init__(*args, status_code=status_code, body=body)
self.reason = reason
class AccountUpstreamUnavailableError(AccountUpstreamError, BillingUpstreamUnavailableError):
pass
class AccountUpstreamInvalidResponseError(AccountUpstreamError, BillingUpstreamInvalidResponseError):
pass
ACCOUNT_ERRORS = ServiceErrors(
is_reason=_is_account_error_reason,
unavailable=AccountUpstreamUnavailableError,
invalid_response=AccountUpstreamInvalidResponseError,
)
def _is_saas_notification_error_reason(value: str) -> TypeIs[SaasNotificationErrorReason]:
return value in get_args(SaasNotificationErrorReason.__value__)
class SaasNotificationUpstreamError(BillingUpstreamError):
"""A failed SaasNotification call, with its reason typed by `SaasNotificationErrorReason`."""
def __init__(
self,
*args: object,
status_code: int | None = None,
body: UpstreamErrorBody | None = None,
reason: SaasNotificationErrorReason | UnknownErrorReason | None = None,
) -> None:
super().__init__(*args, status_code=status_code, body=body)
self.reason = reason
class SaasNotificationUpstreamUnavailableError(SaasNotificationUpstreamError, BillingUpstreamUnavailableError):
pass
class SaasNotificationUpstreamInvalidResponseError(SaasNotificationUpstreamError, BillingUpstreamInvalidResponseError):
pass
SAAS_NOTIFICATION_ERRORS = ServiceErrors(
is_reason=_is_saas_notification_error_reason,
unavailable=SaasNotificationUpstreamUnavailableError,
invalid_response=SaasNotificationUpstreamInvalidResponseError,
)
def _is_utm_error_reason(value: str) -> TypeIs[UtmErrorReason]:
return value in get_args(UtmErrorReason.__value__)
class UtmUpstreamError(BillingUpstreamError):
"""A failed Utm call, with its reason typed by `UtmErrorReason`."""
def __init__(
self,
*args: object,
status_code: int | None = None,
body: UpstreamErrorBody | None = None,
reason: UtmErrorReason | UnknownErrorReason | None = None,
) -> None:
super().__init__(*args, status_code=status_code, body=body)
self.reason = reason
class UtmUpstreamUnavailableError(UtmUpstreamError, BillingUpstreamUnavailableError):
pass
class UtmUpstreamInvalidResponseError(UtmUpstreamError, BillingUpstreamInvalidResponseError):
pass
UTM_ERRORS = ServiceErrors(
is_reason=_is_utm_error_reason,
unavailable=UtmUpstreamUnavailableError,
invalid_response=UtmUpstreamInvalidResponseError,
)
def _is_saas_network_access_policy_error_reason(value: str) -> TypeIs[SaasNetworkAccessPolicyErrorReason]:
return value in get_args(SaasNetworkAccessPolicyErrorReason.__value__)
class SaasNetworkAccessPolicyUpstreamError(BillingUpstreamError):
"""A failed SaasNetworkAccessPolicy call, with its reason typed by `SaasNetworkAccessPolicyErrorReason`."""
def __init__(
self,
*args: object,
status_code: int | None = None,
body: UpstreamErrorBody | None = None,
reason: SaasNetworkAccessPolicyErrorReason | UnknownErrorReason | None = None,
) -> None:
super().__init__(*args, status_code=status_code, body=body)
self.reason = reason
class SaasNetworkAccessPolicyUpstreamUnavailableError(
SaasNetworkAccessPolicyUpstreamError, BillingUpstreamUnavailableError
):
pass
class SaasNetworkAccessPolicyUpstreamInvalidResponseError(
SaasNetworkAccessPolicyUpstreamError, BillingUpstreamInvalidResponseError
):
pass
SAAS_NETWORK_ACCESS_POLICY_ERRORS = ServiceErrors(
is_reason=_is_saas_network_access_policy_error_reason,
unavailable=SaasNetworkAccessPolicyUpstreamUnavailableError,
invalid_response=SaasNetworkAccessPolicyUpstreamInvalidResponseError,
)
def account_delete_account(*, account_id: str) -> DeleteAccountReply:
"""DELETE /v1/account (Account_DeleteAccount)."""
return request(
Upstream.SAAS,
"DELETE",
"/account",
response_model=DeleteAccountReply,
errors=ACCOUNT_ERRORS,
path_params={},
params={"accountId": account_id},
body=None,
)
def account_check_account_in_freeze(*, email: str) -> CheckAccountInFreezeReply:
"""GET /v1/account/in-freeze (Account_CheckAccountInFreeze)."""
return request(
Upstream.SAAS,
"GET",
"/account/in-freeze",
response_model=CheckAccountInFreezeReply,
errors=ACCOUNT_ERRORS,
path_params={},
params={"email": email},
body=None,
)
def account_add_delete_feedback(*, body: AddDeleteFeedbackRequest) -> AddDeleteFeedbackReply:
"""POST /v1/account/delete-feedback (Account_AddDeleteFeedback)."""
return request(
Upstream.SAAS,
"POST",
"/account/delete-feedback",
response_model=AddDeleteFeedbackReply,
errors=ACCOUNT_ERRORS,
path_params={},
params={},
body=body,
)
def notification_get_account_notification(*, account_id: str) -> GetAccountNotificationReply:
"""GET /v1/notifications/active (SaasNotification_GetAccountNotification)."""
return request(
Upstream.SAAS,
"GET",
"/notifications/active",
response_model=GetAccountNotificationReply,
errors=SAAS_NOTIFICATION_ERRORS,
path_params={},
params={"accountId": account_id},
body=None,
)
def notification_dismiss_notification(
*, notification_id: str, body: DismissNotificationReq
) -> DismissNotificationReply:
"""POST /v1/notifications/{notificationId}/dismiss (SaasNotification_DismissNotification)."""
return request(
Upstream.SAAS,
"POST",
"/notifications/{notification_id}/dismiss",
response_model=DismissNotificationReply,
errors=SAAS_NOTIFICATION_ERRORS,
path_params={"notification_id": notification_id},
params={},
body=body,
)
def utm_record_tenant_utm(*, body: TenantUtmRequest) -> TenantUtmReply:
"""POST /v1/tenant_utms (Utm_RecordTenantUtm)."""
return request(
Upstream.SAAS,
"POST",
"/tenant_utms",
response_model=TenantUtmReply,
errors=UTM_ERRORS,
path_params={},
params={},
body=body,
)
def network_access_policy_list_network_access_groups(
*, tenant_id: str, actor_account_id: str
) -> ListNetworkAccessGroupsReply:
"""GET /v1/tenants/{tenantId}/network-access-groups (SaasNetworkAccessPolicy_ListNetworkAccessGroups)."""
return request(
Upstream.NETWORK_ACCESS,
"GET",
"/tenants/{tenant_id}/network-access-groups",
response_model=ListNetworkAccessGroupsReply,
errors=SAAS_NETWORK_ACCESS_POLICY_ERRORS,
path_params={"tenant_id": tenant_id},
params={"actorAccountId": actor_account_id},
body=None,
)
def network_access_policy_create_network_access_group(
*, tenant_id: str, body: CreateNetworkAccessGroupRequest
) -> NetworkAccessGroupReply:
"""POST /v1/tenants/{tenantId}/network-access-groups (SaasNetworkAccessPolicy_CreateNetworkAccessGroup)."""
return request(
Upstream.NETWORK_ACCESS,
"POST",
"/tenants/{tenant_id}/network-access-groups",
response_model=NetworkAccessGroupReply,
errors=SAAS_NETWORK_ACCESS_POLICY_ERRORS,
path_params={"tenant_id": tenant_id},
params={},
body=body,
)
def network_access_policy_get_network_access_group(
*, tenant_id: str, group_id: str, actor_account_id: str
) -> NetworkAccessGroupReply:
"""GET /v1/tenants/{tenantId}/network-access-groups/{groupId} (SaasNetworkAccessPolicy_GetNetworkAccessGroup)."""
return request(
Upstream.NETWORK_ACCESS,
"GET",
"/tenants/{tenant_id}/network-access-groups/{group_id}",
response_model=NetworkAccessGroupReply,
errors=SAAS_NETWORK_ACCESS_POLICY_ERRORS,
path_params={"tenant_id": tenant_id, "group_id": group_id},
params={"actorAccountId": actor_account_id},
body=None,
)
def network_access_policy_update_network_access_group(
*, tenant_id: str, group_id: str, body: UpdateNetworkAccessGroupRequest
) -> NetworkAccessGroupReply:
"""PUT /v1/tenants/{tenantId}/network-access-groups/{groupId} (SaasNetworkAccessPolicy_UpdateNetworkAccessGroup)."""
return request(
Upstream.NETWORK_ACCESS,
"PUT",
"/tenants/{tenant_id}/network-access-groups/{group_id}",
response_model=NetworkAccessGroupReply,
errors=SAAS_NETWORK_ACCESS_POLICY_ERRORS,
path_params={"tenant_id": tenant_id, "group_id": group_id},
params={},
body=body,
)
def network_access_policy_delete_network_access_group(
*, tenant_id: str, group_id: str, expected_version: str, actor_account_id: str
) -> DeleteNetworkAccessGroupReply:
"""DELETE /v1/tenants/{tenantId}/network-access-groups/{groupId}
SaasNetworkAccessPolicy_DeleteNetworkAccessGroup
"""
return request(
Upstream.NETWORK_ACCESS,
"DELETE",
"/tenants/{tenant_id}/network-access-groups/{group_id}",
response_model=DeleteNetworkAccessGroupReply,
errors=SAAS_NETWORK_ACCESS_POLICY_ERRORS,
path_params={"tenant_id": tenant_id, "group_id": group_id},
params={"expectedVersion": expected_version, "actorAccountId": actor_account_id},
body=None,
)
def network_access_policy_get_app_network_access_group(
*, tenant_id: str, app_id: str, actor_account_id: str
) -> GetAppNetworkAccessGroupReply:
"""GET /v1/tenants/{tenantId}/apps/{appId}/network-access-group
SaasNetworkAccessPolicy_GetAppNetworkAccessGroup
"""
return request(
Upstream.NETWORK_ACCESS,
"GET",
"/tenants/{tenant_id}/apps/{app_id}/network-access-group",
response_model=GetAppNetworkAccessGroupReply,
errors=SAAS_NETWORK_ACCESS_POLICY_ERRORS,
path_params={"tenant_id": tenant_id, "app_id": app_id},
params={"actorAccountId": actor_account_id},
body=None,
)
def network_access_policy_put_app_network_access_group(
*, tenant_id: str, app_id: str, body: PutAppNetworkAccessGroupRequest
) -> PutAppNetworkAccessGroupReply:
"""PUT /v1/tenants/{tenantId}/apps/{appId}/network-access-group
SaasNetworkAccessPolicy_PutAppNetworkAccessGroup
"""
return request(
Upstream.NETWORK_ACCESS,
"PUT",
"/tenants/{tenant_id}/apps/{app_id}/network-access-group",
response_model=PutAppNetworkAccessGroupReply,
errors=SAAS_NETWORK_ACCESS_POLICY_ERRORS,
path_params={"tenant_id": tenant_id, "app_id": app_id},
params={},
body=body,
)
def network_access_policy_get_app_network_access_group_status(
*, tenant_id: str, app_id: str, actor_account_id: str
) -> GetAppNetworkAccessGroupStatusReply:
"""GET /v1/tenants/{tenantId}/apps/{appId}/network-access-group/status
SaasNetworkAccessPolicy_GetAppNetworkAccessGroupStatus
"""
return request(
Upstream.NETWORK_ACCESS,
"GET",
"/tenants/{tenant_id}/apps/{app_id}/network-access-group/status",
response_model=GetAppNetworkAccessGroupStatusReply,
errors=SAAS_NETWORK_ACCESS_POLICY_ERRORS,
path_params={"tenant_id": tenant_id, "app_id": app_id},
params={"actorAccountId": actor_account_id},
body=None,
)
def network_access_policy_cleanup_app_network_access_group_binding(
*, tenant_id: str, app_id: str
) -> CleanupAppNetworkAccessGroupBindingReply:
"""DELETE /v1/tenants/{tenantId}/apps/{appId}/network-access-group-binding
SaasNetworkAccessPolicy_CleanupAppNetworkAccessGroupBinding
"""
return request(
Upstream.NETWORK_ACCESS,
"DELETE",
"/tenants/{tenant_id}/apps/{app_id}/network-access-group-binding",
response_model=CleanupAppNetworkAccessGroupBindingReply,
errors=SAAS_NETWORK_ACCESS_POLICY_ERRORS,
path_params={"tenant_id": tenant_id, "app_id": app_id},
params={},
body=None,
)