a4-cli 0.42.0

"""Generated by a4 from protocol package `escrow-resolution` 0.1.0 (arete.protocol-spec/v1 `escrowResolution`). Do not edit.
"""

from __future__ import annotations

from typing import Any, Dict, List, Literal, Mapping, NoReturn, Optional, Sequence, TypedDict, Union

import httpx

from arete.auth import AuthConfig
from arete.errors import AreteError
from arete.protocol import (
    ProtocolAccessRejectionCode,
    ProtocolClient,
    ProtocolDefinition,
    ProtocolDeployment,
    ProtocolInstanceHandle,
    ProtocolInteger,
    ProtocolLiveStack,
    ProtocolMetadataFieldDefinition,
    ProtocolPrepared,
    ProtocolRejection,
    ProtocolStateRejectionCode,
    ProtocolStepDefinition,
    ProtocolTransactionFormat,
    VerifiedWebhook,
    decimal_args,
    verify_webhook,
)

EscrowResolutionInstanceState = Literal["awaiting_resolution", "funded", "refunded", "released"]
EscrowResolutionParticipantState = NoReturn
EscrowResolutionEvent = Literal["escrow.awaitingResolution", "escrow.resolved"]


class EscrowResolutionParams(TypedDict):
    amount: ProtocolInteger
    resolveAfter: ProtocolInteger


class EscrowResolutionOpenArgs(TypedDict):
    amount: ProtocolInteger
    resolveAfter: ProtocolInteger
    resolver: str


EscrowResolutionOpenGuardCode = NoReturn


EscrowResolutionOpenResult = Union[ProtocolPrepared, ProtocolRejection[Union[EscrowResolutionOpenGuardCode, ProtocolAccessRejectionCode, ProtocolStateRejectionCode]]]


class EscrowResolutionRefundArgs(TypedDict):
    pass


EscrowResolutionRefundGuardCode = Literal["not_awaiting_resolution", "not_resolver"]


EscrowResolutionRefundResult = Union[ProtocolPrepared, ProtocolRejection[Union[EscrowResolutionRefundGuardCode, ProtocolAccessRejectionCode, ProtocolStateRejectionCode]]]


class EscrowResolutionReleaseArgs(TypedDict):
    pass


EscrowResolutionReleaseGuardCode = Literal["not_awaiting_resolution", "not_resolver"]


EscrowResolutionReleaseResult = Union[ProtocolPrepared, ProtocolRejection[Union[EscrowResolutionReleaseGuardCode, ProtocolAccessRejectionCode, ProtocolStateRejectionCode]]]


class EscrowResolutionInstanceMetadata(TypedDict):
    pass


class EscrowResolutionParticipantMetadata(TypedDict):
    pass


class EscrowResolutionWebhookPayload(TypedDict):
    """The JSON body of a `escrowResolution` webhook delivery."""

    id: str
    event: EscrowResolutionEvent
    deploymentId: str
    instanceId: str
    participantId: Optional[str]
    state: Union[EscrowResolutionInstanceState, EscrowResolutionParticipantState]
    occurredAt: str


STACKS: Dict[str, Any] = {}
PROGRAMS: Dict[str, Any] = {}

ESCROW_RESOLUTION_PROTOCOL: ProtocolDefinition = ProtocolDefinition(
    name="escrowResolution",
    version="0.1.0",
    package="escrow-resolution",
    package_release_hash="arete:h1:package-release:sha256:9999999999999999999999999999999999999999999999999999999999999999",
    protocol_spec_hash="arete:h1:protocol-spec:sha256:b7b4249437959a64742a47da0ce9e133e32d851b53a909d7e47996ae3a789240",
    protocol_release_hash="arete:h1:protocol-release:sha256:ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff",
    deployment=ProtocolDeployment(
        deployment_id="7d0c5a52-2f8e-4c1b-9a37-1b6f3c2a9e10",
        access="public",
        api_url="https://api.example.test/api",
        session_endpoint="https://api.example.test/ws/sessions",
    ),
    instance_visibility="shared",
    steps={
        "open": ProtocolStepDefinition(scope="create", access="secret", guards=(), args=("amount", "resolveAfter", "resolver")),
        "refund": ProtocolStepDefinition(scope="instance", access="secret", guards=("not_resolver", "not_awaiting_resolution"), args=()),
        "release": ProtocolStepDefinition(scope="instance", access="secret", guards=("not_resolver", "not_awaiting_resolution"), args=()),
    },
    events={
        "escrow.awaitingResolution": "instance",
        "escrow.resolved": "instance",
    },
    stacks=STACKS,
    programs=PROGRAMS,
)


class EscrowResolutionPublicSteps:
    """Steps a publishable key may prepare."""

    def __init__(self, client: ProtocolClient) -> None:
        self._client = client


class EscrowResolutionSecretSteps(EscrowResolutionPublicSteps):
    """Every client-callable step; `secret` steps need a secret key."""

    async def open(
        self,
        *,
        wallet: str,
        args: EscrowResolutionOpenArgs,
        transaction_format: ProtocolTransactionFormat = "v1",
        idempotency_key: Optional[str] = None,
        client_context: Optional[Mapping[str, Any]] = None,
    ) -> EscrowResolutionOpenResult:
        """Prepare `open` (create step, secret access)."""
        return await self._client.prepare(
            "open",
            wallet=wallet,
            args=decimal_args(args, ("amount", "resolveAfter")),
            transaction_format=transaction_format,
            idempotency_key=idempotency_key,
            client_context=client_context,
        )

    async def refund(
        self,
        *,
        instance: str,
        wallet: str,
        args: Optional[EscrowResolutionRefundArgs] = None,
        transaction_format: ProtocolTransactionFormat = "v1",
        idempotency_key: Optional[str] = None,
        client_context: Optional[Mapping[str, Any]] = None,
    ) -> EscrowResolutionRefundResult:
        """Prepare `refund` (instance step, secret access)."""
        return await self._client.prepare(
            "refund",
            wallet=wallet,
            instance=instance,
            args=args,
            transaction_format=transaction_format,
            idempotency_key=idempotency_key,
            client_context=client_context,
        )

    async def release(
        self,
        *,
        instance: str,
        wallet: str,
        args: Optional[EscrowResolutionReleaseArgs] = None,
        transaction_format: ProtocolTransactionFormat = "v1",
        idempotency_key: Optional[str] = None,
        client_context: Optional[Mapping[str, Any]] = None,
    ) -> EscrowResolutionReleaseResult:
        """Prepare `release` (instance step, secret access)."""
        return await self._client.prepare(
            "release",
            wallet=wallet,
            instance=instance,
            args=args,
            transaction_format=transaction_format,
            idempotency_key=idempotency_key,
            client_context=client_context,
        )


class EscrowResolutionClient(ProtocolClient):
    """`escrowResolution` with a publishable key (`AuthConfig(publishable_key=...)`): only
    `public` steps are on `steps`.

    `prepare_steps` limits which steps the session may prepare (default:
    every `public` step).
    """

    steps: EscrowResolutionPublicSteps

    def __init__(
        self,
        *,
        auth: Optional[AuthConfig] = None,
        http_client: Optional[httpx.AsyncClient] = None,
        api_url: Optional[str] = None,
        session_endpoint: Optional[str] = None,
        deployment_id: Optional[str] = None,
        stacks: Optional[Mapping[str, Any]] = None,
        programs: Optional[Mapping[str, Any]] = None,
        key_class: Optional[str] = None,
        prepare_steps: Optional[Sequence[str]] = None,
    ) -> None:
        super().__init__(
            ESCROW_RESOLUTION_PROTOCOL,
            auth=auth,
            http_client=http_client,
            api_url=api_url,
            session_endpoint=session_endpoint,
            deployment_id=deployment_id,
            stacks=stacks,
            programs=programs,
            key_class=key_class,
            steps=prepare_steps,
        )
        self.steps = EscrowResolutionPublicSteps(self)

    @classmethod
    def from_session(
        cls, session: Any, *, prepare_steps: Optional[Sequence[str]] = None
    ) -> "EscrowResolutionClient":
        """A client on `session`'s auth and HTTP client whose instance handles
        follow the session's connected stacks."""
        return cls(**session.protocol_client_options(ESCROW_RESOLUTION_PROTOCOL), prepare_steps=prepare_steps)

    def instance_handle(
        self, instance_id: str, *, wallet: Optional[str] = None
    ) -> ProtocolInstanceHandle:
        """A live instance, kept current from its event stream; use it with
        `async with` or `await` it.

        This protocol publishes no live stack view.
        """
        return super().instance_handle(instance_id, wallet=wallet)


class EscrowResolutionSecretClient(ProtocolClient):
    """`escrowResolution` with a secret key (`AuthConfig(secret_key=...)`, or `ARETE_API_KEY`):
    every client-callable step is on `steps`. Raises when the configured
    credential is a publishable key.
    """

    steps: EscrowResolutionSecretSteps

    def __init__(
        self,
        *,
        auth: Optional[AuthConfig] = None,
        http_client: Optional[httpx.AsyncClient] = None,
        api_url: Optional[str] = None,
        session_endpoint: Optional[str] = None,
        deployment_id: Optional[str] = None,
        stacks: Optional[Mapping[str, Any]] = None,
        programs: Optional[Mapping[str, Any]] = None,
    ) -> None:
        super().__init__(
            ESCROW_RESOLUTION_PROTOCOL,
            auth=auth,
            http_client=http_client,
            api_url=api_url,
            session_endpoint=session_endpoint,
            deployment_id=deployment_id,
            stacks=stacks,
            programs=programs,
            key_class="secret",
        )
        if self.key_class != "secret":
            raise AreteError(
                "EscrowResolutionSecretClient requires a secret key; use EscrowResolutionClient with a publishable key",
                "INVALID_CONFIG",
            )
        self.steps = EscrowResolutionSecretSteps(self)

    @classmethod
    def from_session(cls, session: Any) -> "EscrowResolutionSecretClient":
        """A client on `session`'s auth and HTTP client whose instance handles
        follow the session's connected stacks."""
        options = session.protocol_client_options(ESCROW_RESOLUTION_PROTOCOL)
        options.pop("key_class", None)
        return cls(**options)

    def instance_handle(
        self, instance_id: str, *, wallet: Optional[str] = None
    ) -> ProtocolInstanceHandle:
        """A live instance, kept current from its event stream; use it with
        `async with` or `await` it.

        This protocol publishes no live stack view.
        """
        return super().instance_handle(instance_id, wallet=wallet)


def verify_escrow_resolution_webhook(
    secret: Union[str, bytes],
    body: Union[str, bytes],
    headers: Any,
    *,
    now: Optional[int] = None,
    tolerance_seconds: int = 300,
) -> VerifiedWebhook[EscrowResolutionWebhookPayload]:
    """Verify a signed webhook delivery for this protocol's events."""
    return verify_webhook(secret, body, headers, now=now, tolerance_seconds=tolerance_seconds)


__all__ = [
    "ESCROW_RESOLUTION_PROTOCOL",
    "EscrowResolutionClient",
    "EscrowResolutionSecretClient",
    "EscrowResolutionPublicSteps",
    "EscrowResolutionSecretSteps",
    "EscrowResolutionInstanceState",
    "EscrowResolutionParticipantState",
    "EscrowResolutionEvent",
    "EscrowResolutionParams",
    "EscrowResolutionInstanceMetadata",
    "EscrowResolutionParticipantMetadata",
    "EscrowResolutionWebhookPayload",
    "STACKS",
    "PROGRAMS",
    "verify_escrow_resolution_webhook",
    "EscrowResolutionOpenArgs",
    "EscrowResolutionOpenGuardCode",
    "EscrowResolutionOpenResult",
    "EscrowResolutionRefundArgs",
    "EscrowResolutionRefundGuardCode",
    "EscrowResolutionRefundResult",
    "EscrowResolutionReleaseArgs",
    "EscrowResolutionReleaseGuardCode",
    "EscrowResolutionReleaseResult",
]