dify/api/services/billing_client/billing.py

# Code generated by api/dev/generate_billing_client.py from billing.openapi.json. DO NOT EDIT.
# dify-saas commit: 2d358b5a0555440b30d8601f5e786a15e28a2345

from datetime import datetime
from typing import Literal, TypeIs, get_args

from services.billing_client.transport import BillingModel, ServiceErrors, Upstream, request
from services.errors.billing import (
    BillingUpstreamError,
    BillingUpstreamInvalidResponseError,
    BillingUpstreamUnavailableError,
    UnknownErrorReason,
    UpstreamErrorBody,
)


class AutocompleteEducationInstitutionReply(BillingModel):
    data: list[str]
    curr_page: int
    has_next: bool


class SubscriptionPlanItem(BillingModel):
    plan: Literal["sandbox", "professional", "team"]
    expiration_date: int


class BatchGetSubscriptionPlanReply(BillingModel):
    data: dict[str, SubscriptionPlanItem]


class BatchGetSubscriptionPlanReq(BillingModel):
    tenant_ids: list[str]


type BillingComplianceErrorReason = Literal[
    "",
    "CODEC",
    "VALIDATOR",
    "UNKNOWN",
    "INVALID_SECRET_KEY",
    "REQUEST_IN_PROGRESS",
    "MISSING_TENANT_ID",
    "INVALID_TENANT_ID",
    "MISSING_ACCOUNT_ID",
    "INVALID_ACCOUNT_ID",
    "MISSING_PREFILLED_EMAIL",
    "INVALID_COMPLIANCE_DOWNLOAD_REQUEST",
    "MISSING_DOC_NAME",
    "MISSING_IP_ADDRESS",
    "MISSING_DEVICE_INFO",
    "COMPLIANCE_DOCUMENT_NOT_FOUND",
    "COMPLIANCE_DOWNLOAD_FORBIDDEN",
    "COMPLIANCE_DOWNLOAD_FAILED",
]


type BillingEducationErrorReason = Literal[
    "",
    "CODEC",
    "VALIDATOR",
    "UNKNOWN",
    "INVALID_SECRET_KEY",
    "REQUEST_IN_PROGRESS",
    "MISSING_TENANT_ID",
    "INVALID_TENANT_ID",
    "MISSING_ACCOUNT_ID",
    "INVALID_ACCOUNT_ID",
    "MISSING_PREFILLED_EMAIL",
    "EDUCATION_VERIFY_FAILED",
    "EDUCATION_CREATE_FAILED",
    "EDUCATION_STATUS_FAILED",
]


type BillingInvoiceErrorReason = Literal[
    "",
    "CODEC",
    "VALIDATOR",
    "UNKNOWN",
    "INVALID_SECRET_KEY",
    "REQUEST_IN_PROGRESS",
    "MISSING_TENANT_ID",
    "INVALID_TENANT_ID",
    "MISSING_ACCOUNT_ID",
    "INVALID_ACCOUNT_ID",
    "MISSING_PREFILLED_EMAIL",
    "INVALID_INVOICE_REQUEST",
    "CREATE_INVOICE_PORTAL_FAILED",
]


type BillingModelProviderErrorReason = Literal[
    "",
    "CODEC",
    "VALIDATOR",
    "UNKNOWN",
    "INVALID_SECRET_KEY",
    "REQUEST_IN_PROGRESS",
    "MISSING_TENANT_ID",
    "INVALID_TENANT_ID",
    "MISSING_ACCOUNT_ID",
    "INVALID_ACCOUNT_ID",
    "MISSING_PREFILLED_EMAIL",
    "INVALID_MODEL_PROVIDER_PAYMENT_LINK_REQUEST",
    "MISSING_PROVIDER_NAME",
    "INVALID_MODEL_PROVIDER",
    "CREATE_MODEL_PROVIDER_PAYMENT_LINK_FAILED",
]


type BillingQuotaErrorReason = Literal[
    "",
    "CODEC",
    "VALIDATOR",
    "UNKNOWN",
    "INVALID_SECRET_KEY",
    "REQUEST_IN_PROGRESS",
    "MISSING_TENANT_ID",
    "INVALID_TENANT_ID",
    "MISSING_ACCOUNT_ID",
    "INVALID_ACCOUNT_ID",
    "MISSING_PREFILLED_EMAIL",
]


type BillingSubscriptionErrorReason = Literal[
    "",
    "CODEC",
    "VALIDATOR",
    "UNKNOWN",
    "INVALID_SECRET_KEY",
    "REQUEST_IN_PROGRESS",
    "MISSING_TENANT_ID",
    "INVALID_TENANT_ID",
    "MISSING_ACCOUNT_ID",
    "INVALID_ACCOUNT_ID",
    "MISSING_PREFILLED_EMAIL",
    "INVALID_PAYMENT_LINK_REQUEST",
    "MISSING_PLAN",
    "MISSING_INTERVAL",
    "INVALID_PLAN_INTERVAL",
    "CREATE_SUBSCRIPTION_PAYMENT_LINK_FAILED",
    "INVALID_KNOWLEDGE_RATE_LIMIT_REQUEST",
    "TENANT_IDS_COUNT_TOO_LARGE",
    "DOCUMENT_USAGE_UNAVAILABLE",
]


class BillingSubscriptionPlan(BillingModel):
    plan: Literal["sandbox", "professional", "team"]
    interval: Literal["", "month", "year"]
    education: bool
    status: str
    cancel_at_period_end: bool
    paid_period_end: int


class CreateEducationIdentityReply(BillingModel):
    message: str


class CreateEducationIdentityReq(BillingModel):
    account_id: str
    curr_tenant_id: str
    institution: str
    role: str
    token: str


class DownloadComplianceDocumentReply(BillingModel):
    url: str


class DownloadComplianceDocumentReq(BillingModel):
    doc_name: str
    account_id: str
    tenant_id: str
    ip_address: str
    device_info: str


class GetEducationIdentityStatusReply(BillingModel):
    result: bool
    is_student: bool
    expire_at: datetime
    allow_refresh: bool


class GetInvoicesReply(BillingModel):
    url: str


class GetKnowledgeRateLimitReply(BillingModel):
    subscription_plan: Literal["sandbox", "professional", "team"]
    limit: int


class GetModelProviderPaymentLinkReply(BillingModel):
    url: str


class KnowledgeRateLimit(BillingModel):
    size: int
    limit: int


class SubscriptionQuota(BillingModel):
    size: int
    limit: int


class VectorSpaceQuota(BillingModel):
    size: float
    limit: int
    usage_unknown: bool | None = None


class GetSubscriptionInfoReply(BillingModel):
    enabled: bool
    subscription: BillingSubscriptionPlan
    members: SubscriptionQuota
    apps: SubscriptionQuota
    vector_space: VectorSpaceQuota | None
    knowledge_rate_limit: KnowledgeRateLimit
    documents_upload_quota: SubscriptionQuota
    annotation_quota_limit: SubscriptionQuota
    docs_processing: Literal["standard", "priority", "top-priority"]
    can_replace_logo: bool
    model_load_balancing_enabled: bool
    knowledge_pipeline_publish_enabled: bool
    next_credit_reset_date: int


class GetSubscriptionPaymentLinkReply(BillingModel):
    url: str


class SubscriptionCleanupWhitelist(BillingModel):
    id: str
    tenant_id: str
    contact: str
    expired_at: datetime
    created_at: datetime
    updated_at: datetime


class ListSubscriptionCleanupWhitelistReply(BillingModel):
    data: list[SubscriptionCleanupWhitelist]


class QuotaCommitReply(BillingModel):
    available: int
    reserved: int
    refunded: int


class QuotaCommitReq(BillingModel):
    tenant_id: str
    feature_key: str
    reservation_id: str
    actual_amount: int
    meta: dict[str, str]
    bucket: str


class QuotaConsumeCappedReply(BillingModel):
    deducted: int
    available: int
    reserved: int
    quota: int
    usage: int


class QuotaConsumeCappedReq(BillingModel):
    tenant_id: str
    feature_key: str
    request_id: str
    amount: int
    meta: dict[str, str]
    bucket: str


class QuotaFeatureInfo(BillingModel):
    usage: int
    limit: int
    reset_date: int


class QuotaGetBalanceReply(BillingModel):
    available: int
    reserved: int
    quota: int
    usage: int
    exhausted_at: int


class QuotaGetInfoReply(BillingModel):
    trigger_event: QuotaFeatureInfo
    api_rate_limit: QuotaFeatureInfo


class QuotaReleaseReply(BillingModel):
    available: int
    reserved: int
    released: int


class QuotaReleaseReq(BillingModel):
    tenant_id: str
    feature_key: str
    reservation_id: str
    bucket: str


class QuotaReserveReply(BillingModel):
    reservation_id: str
    available: int
    reserved: int


class QuotaReserveReq(BillingModel):
    tenant_id: str
    feature_key: str
    request_id: str
    amount: int
    meta: dict[str, str]
    bucket: str


class VerifyEducationIdentityReply(BillingModel):
    token: str
    status: Literal["eligible", "not_eligible"]


def _is_billing_subscription_error_reason(value: str) -> TypeIs[BillingSubscriptionErrorReason]:
    return value in get_args(BillingSubscriptionErrorReason.__value__)


class BillingSubscriptionUpstreamError(BillingUpstreamError):
    """A failed BillingSubscription call, with its reason typed by `BillingSubscriptionErrorReason`."""

    def __init__(
        self,
        *args: object,
        status_code: int | None = None,
        body: UpstreamErrorBody | None = None,
        reason: BillingSubscriptionErrorReason | UnknownErrorReason | None = None,
    ) -> None:
        super().__init__(*args, status_code=status_code, body=body)
        self.reason = reason


class BillingSubscriptionUpstreamUnavailableError(BillingSubscriptionUpstreamError, BillingUpstreamUnavailableError):
    pass


class BillingSubscriptionUpstreamInvalidResponseError(
    BillingSubscriptionUpstreamError, BillingUpstreamInvalidResponseError
):
    pass


BILLING_SUBSCRIPTION_ERRORS = ServiceErrors(
    is_reason=_is_billing_subscription_error_reason,
    unavailable=BillingSubscriptionUpstreamUnavailableError,
    invalid_response=BillingSubscriptionUpstreamInvalidResponseError,
)


def _is_billing_model_provider_error_reason(value: str) -> TypeIs[BillingModelProviderErrorReason]:
    return value in get_args(BillingModelProviderErrorReason.__value__)


class BillingModelProviderUpstreamError(BillingUpstreamError):
    """A failed BillingModelProvider call, with its reason typed by `BillingModelProviderErrorReason`."""

    def __init__(
        self,
        *args: object,
        status_code: int | None = None,
        body: UpstreamErrorBody | None = None,
        reason: BillingModelProviderErrorReason | UnknownErrorReason | None = None,
    ) -> None:
        super().__init__(*args, status_code=status_code, body=body)
        self.reason = reason


class BillingModelProviderUpstreamUnavailableError(BillingModelProviderUpstreamError, BillingUpstreamUnavailableError):
    pass


class BillingModelProviderUpstreamInvalidResponseError(
    BillingModelProviderUpstreamError, BillingUpstreamInvalidResponseError
):
    pass


BILLING_MODEL_PROVIDER_ERRORS = ServiceErrors(
    is_reason=_is_billing_model_provider_error_reason,
    unavailable=BillingModelProviderUpstreamUnavailableError,
    invalid_response=BillingModelProviderUpstreamInvalidResponseError,
)


def _is_billing_invoice_error_reason(value: str) -> TypeIs[BillingInvoiceErrorReason]:
    return value in get_args(BillingInvoiceErrorReason.__value__)


class BillingInvoiceUpstreamError(BillingUpstreamError):
    """A failed BillingInvoice call, with its reason typed by `BillingInvoiceErrorReason`."""

    def __init__(
        self,
        *args: object,
        status_code: int | None = None,
        body: UpstreamErrorBody | None = None,
        reason: BillingInvoiceErrorReason | UnknownErrorReason | None = None,
    ) -> None:
        super().__init__(*args, status_code=status_code, body=body)
        self.reason = reason


class BillingInvoiceUpstreamUnavailableError(BillingInvoiceUpstreamError, BillingUpstreamUnavailableError):
    pass


class BillingInvoiceUpstreamInvalidResponseError(BillingInvoiceUpstreamError, BillingUpstreamInvalidResponseError):
    pass


BILLING_INVOICE_ERRORS = ServiceErrors(
    is_reason=_is_billing_invoice_error_reason,
    unavailable=BillingInvoiceUpstreamUnavailableError,
    invalid_response=BillingInvoiceUpstreamInvalidResponseError,
)


def _is_billing_education_error_reason(value: str) -> TypeIs[BillingEducationErrorReason]:
    return value in get_args(BillingEducationErrorReason.__value__)


class BillingEducationUpstreamError(BillingUpstreamError):
    """A failed BillingEducation call, with its reason typed by `BillingEducationErrorReason`."""

    def __init__(
        self,
        *args: object,
        status_code: int | None = None,
        body: UpstreamErrorBody | None = None,
        reason: BillingEducationErrorReason | UnknownErrorReason | None = None,
    ) -> None:
        super().__init__(*args, status_code=status_code, body=body)
        self.reason = reason


class BillingEducationUpstreamUnavailableError(BillingEducationUpstreamError, BillingUpstreamUnavailableError):
    pass


class BillingEducationUpstreamInvalidResponseError(BillingEducationUpstreamError, BillingUpstreamInvalidResponseError):
    pass


BILLING_EDUCATION_ERRORS = ServiceErrors(
    is_reason=_is_billing_education_error_reason,
    unavailable=BillingEducationUpstreamUnavailableError,
    invalid_response=BillingEducationUpstreamInvalidResponseError,
)


def _is_billing_compliance_error_reason(value: str) -> TypeIs[BillingComplianceErrorReason]:
    return value in get_args(BillingComplianceErrorReason.__value__)


class BillingComplianceUpstreamError(BillingUpstreamError):
    """A failed BillingCompliance call, with its reason typed by `BillingComplianceErrorReason`."""

    def __init__(
        self,
        *args: object,
        status_code: int | None = None,
        body: UpstreamErrorBody | None = None,
        reason: BillingComplianceErrorReason | UnknownErrorReason | None = None,
    ) -> None:
        super().__init__(*args, status_code=status_code, body=body)
        self.reason = reason


class BillingComplianceUpstreamUnavailableError(BillingComplianceUpstreamError, BillingUpstreamUnavailableError):
    pass


class BillingComplianceUpstreamInvalidResponseError(
    BillingComplianceUpstreamError, BillingUpstreamInvalidResponseError
):
    pass


BILLING_COMPLIANCE_ERRORS = ServiceErrors(
    is_reason=_is_billing_compliance_error_reason,
    unavailable=BillingComplianceUpstreamUnavailableError,
    invalid_response=BillingComplianceUpstreamInvalidResponseError,
)


def _is_billing_quota_error_reason(value: str) -> TypeIs[BillingQuotaErrorReason]:
    return value in get_args(BillingQuotaErrorReason.__value__)


class BillingQuotaUpstreamError(BillingUpstreamError):
    """A failed BillingQuota call, with its reason typed by `BillingQuotaErrorReason`."""

    def __init__(
        self,
        *args: object,
        status_code: int | None = None,
        body: UpstreamErrorBody | None = None,
        reason: BillingQuotaErrorReason | UnknownErrorReason | None = None,
    ) -> None:
        super().__init__(*args, status_code=status_code, body=body)
        self.reason = reason


class BillingQuotaUpstreamUnavailableError(BillingQuotaUpstreamError, BillingUpstreamUnavailableError):
    pass


class BillingQuotaUpstreamInvalidResponseError(BillingQuotaUpstreamError, BillingUpstreamInvalidResponseError):
    pass


BILLING_QUOTA_ERRORS = ServiceErrors(
    is_reason=_is_billing_quota_error_reason,
    unavailable=BillingQuotaUpstreamUnavailableError,
    invalid_response=BillingQuotaUpstreamInvalidResponseError,
)


def subscription_get_subscription_info(*, tenant_id: str, exclude_vector_space: bool) -> GetSubscriptionInfoReply:
    """GET /v1/subscription/info (BillingSubscription_GetSubscriptionInfo)."""
    return request(
        Upstream.BILLING,
        "GET",
        "/subscription/info",
        response_model=GetSubscriptionInfoReply,
        errors=BILLING_SUBSCRIPTION_ERRORS,
        path_params={},
        params={"tenant_id": tenant_id, "exclude_vector_space": exclude_vector_space},
        body=None,
    )


def subscription_get_subscription_vector_space(*, tenant_id: str, bypass_cache: bool) -> VectorSpaceQuota:
    """GET /v1/subscription/vector-space (BillingSubscription_GetSubscriptionVectorSpace)."""
    return request(
        Upstream.BILLING,
        "GET",
        "/subscription/vector-space",
        response_model=VectorSpaceQuota,
        errors=BILLING_SUBSCRIPTION_ERRORS,
        path_params={},
        params={"tenant_id": tenant_id, "bypass_cache": bypass_cache},
        body=None,
    )


def subscription_get_knowledge_rate_limit(*, tenant_id: str) -> GetKnowledgeRateLimitReply:
    """GET /v1/subscription/knowledge-rate-limit (BillingSubscription_GetKnowledgeRateLimit)."""
    return request(
        Upstream.BILLING,
        "GET",
        "/subscription/knowledge-rate-limit",
        response_model=GetKnowledgeRateLimitReply,
        errors=BILLING_SUBSCRIPTION_ERRORS,
        path_params={},
        params={"tenant_id": tenant_id},
        body=None,
    )


def subscription_get_subscription_payment_link(
    *, plan: str, interval: str, prefilled_email: str, tenant_id: str
) -> GetSubscriptionPaymentLinkReply:
    """GET /v1/subscription/payment-link (BillingSubscription_GetSubscriptionPaymentLink)."""
    return request(
        Upstream.BILLING,
        "GET",
        "/subscription/payment-link",
        response_model=GetSubscriptionPaymentLinkReply,
        errors=BILLING_SUBSCRIPTION_ERRORS,
        path_params={},
        params={"plan": plan, "interval": interval, "prefilled_email": prefilled_email, "tenant_id": tenant_id},
        body=None,
    )


def subscription_batch_get_subscription_plan(*, body: BatchGetSubscriptionPlanReq) -> BatchGetSubscriptionPlanReply:
    """POST /v1/subscription/plan/batch (BillingSubscription_BatchGetSubscriptionPlan)."""
    return request(
        Upstream.BILLING,
        "POST",
        "/subscription/plan/batch",
        response_model=BatchGetSubscriptionPlanReply,
        errors=BILLING_SUBSCRIPTION_ERRORS,
        path_params={},
        params={},
        body=body,
    )


def subscription_list_subscription_cleanup_whitelist() -> ListSubscriptionCleanupWhitelistReply:
    """GET /v1/subscription/cleanup/whitelist (BillingSubscription_ListSubscriptionCleanupWhitelist)."""
    return request(
        Upstream.BILLING,
        "GET",
        "/subscription/cleanup/whitelist",
        response_model=ListSubscriptionCleanupWhitelistReply,
        errors=BILLING_SUBSCRIPTION_ERRORS,
        path_params={},
        params={},
        body=None,
    )


def model_provider_get_model_provider_payment_link(
    *, provider_name: str, tenant_id: str, account_id: str, prefilled_email: str
) -> GetModelProviderPaymentLinkReply:
    """GET /v1/model-provider/payment-link (BillingModelProvider_GetModelProviderPaymentLink)."""
    return request(
        Upstream.BILLING,
        "GET",
        "/model-provider/payment-link",
        response_model=GetModelProviderPaymentLinkReply,
        errors=BILLING_MODEL_PROVIDER_ERRORS,
        path_params={},
        params={
            "provider_name": provider_name,
            "tenant_id": tenant_id,
            "account_id": account_id,
            "prefilled_email": prefilled_email,
        },
        body=None,
    )


def invoice_get_invoices(*, prefilled_email: str, tenant_id: str) -> GetInvoicesReply:
    """GET /v1/invoices (BillingInvoice_GetInvoices)."""
    return request(
        Upstream.BILLING,
        "GET",
        "/invoices",
        response_model=GetInvoicesReply,
        errors=BILLING_INVOICE_ERRORS,
        path_params={},
        params={"prefilled_email": prefilled_email, "tenant_id": tenant_id},
        body=None,
    )


def education_verify_education_identity(*, account_id: str) -> VerifyEducationIdentityReply:
    """GET /v1/education/verify (BillingEducation_VerifyEducationIdentity)."""
    return request(
        Upstream.BILLING,
        "GET",
        "/education/verify",
        response_model=VerifyEducationIdentityReply,
        errors=BILLING_EDUCATION_ERRORS,
        path_params={},
        params={"account_id": account_id},
        body=None,
    )


def education_get_education_identity_status(*, account_id: str) -> GetEducationIdentityStatusReply:
    """GET /v1/education/status (BillingEducation_GetEducationIdentityStatus)."""
    return request(
        Upstream.BILLING,
        "GET",
        "/education/status",
        response_model=GetEducationIdentityStatusReply,
        errors=BILLING_EDUCATION_ERRORS,
        path_params={},
        params={"account_id": account_id},
        body=None,
    )


def education_create_education_identity(*, body: CreateEducationIdentityReq) -> CreateEducationIdentityReply:
    """POST /v1/education (BillingEducation_CreateEducationIdentity)."""
    return request(
        Upstream.BILLING,
        "POST",
        "/education",
        response_model=CreateEducationIdentityReply,
        errors=BILLING_EDUCATION_ERRORS,
        path_params={},
        params={},
        body=body,
    )


def education_autocomplete_education_institution(
    *, keywords: str, page: int, limit: int
) -> AutocompleteEducationInstitutionReply:
    """GET /v1/education/autocomplete (BillingEducation_AutocompleteEducationInstitution)."""
    return request(
        Upstream.BILLING,
        "GET",
        "/education/autocomplete",
        response_model=AutocompleteEducationInstitutionReply,
        errors=BILLING_EDUCATION_ERRORS,
        path_params={},
        params={"keywords": keywords, "page": page, "limit": limit},
        body=None,
    )


def compliance_download_compliance_document(*, body: DownloadComplianceDocumentReq) -> DownloadComplianceDocumentReply:
    """POST /v1/compliance/download (BillingCompliance_DownloadComplianceDocument)."""
    return request(
        Upstream.BILLING,
        "POST",
        "/compliance/download",
        response_model=DownloadComplianceDocumentReply,
        errors=BILLING_COMPLIANCE_ERRORS,
        path_params={},
        params={},
        body=body,
    )


def quota_get_quota_info(*, tenant_id: str) -> QuotaGetInfoReply:
    """GET /v1/quota/info (BillingQuota_GetQuotaInfo)."""
    return request(
        Upstream.BILLING_QUOTA,
        "GET",
        "/quota/info",
        response_model=QuotaGetInfoReply,
        errors=BILLING_QUOTA_ERRORS,
        path_params={},
        params={"tenant_id": tenant_id},
        body=None,
    )


def quota_reserve(*, body: QuotaReserveReq) -> QuotaReserveReply:
    """POST /v1/quota/reserve (BillingQuota_Reserve)."""
    return request(
        Upstream.BILLING_QUOTA,
        "POST",
        "/quota/reserve",
        response_model=QuotaReserveReply,
        errors=BILLING_QUOTA_ERRORS,
        path_params={},
        params={},
        body=body,
    )


def quota_commit(*, body: QuotaCommitReq) -> QuotaCommitReply:
    """POST /v1/quota/commit (BillingQuota_Commit)."""
    return request(
        Upstream.BILLING_QUOTA,
        "POST",
        "/quota/commit",
        response_model=QuotaCommitReply,
        errors=BILLING_QUOTA_ERRORS,
        path_params={},
        params={},
        body=body,
    )


def quota_release(*, body: QuotaReleaseReq) -> QuotaReleaseReply:
    """POST /v1/quota/release (BillingQuota_Release)."""
    return request(
        Upstream.BILLING_QUOTA,
        "POST",
        "/quota/release",
        response_model=QuotaReleaseReply,
        errors=BILLING_QUOTA_ERRORS,
        path_params={},
        params={},
        body=body,
    )


def quota_get_balance(*, tenant_id: str, feature_key: str, bucket: str) -> QuotaGetBalanceReply:
    """GET /v1/quota/balance (BillingQuota_GetBalance)."""
    return request(
        Upstream.BILLING_QUOTA,
        "GET",
        "/quota/balance",
        response_model=QuotaGetBalanceReply,
        errors=BILLING_QUOTA_ERRORS,
        path_params={},
        params={"tenant_id": tenant_id, "feature_key": feature_key, "bucket": bucket},
        body=None,
    )


def quota_consume_capped(*, body: QuotaConsumeCappedReq) -> QuotaConsumeCappedReply:
    """POST /v1/quota/consume-capped (BillingQuota_ConsumeCapped)."""
    return request(
        Upstream.BILLING_QUOTA,
        "POST",
        "/quota/consume-capped",
        response_model=QuotaConsumeCappedReply,
        errors=BILLING_QUOTA_ERRORS,
        path_params={},
        params={},
        body=body,
    )