a4-cli 0.44.2

"""Generated by a4 from protocol package `local-tip-jar` 0.1.0 (arete.protocol-spec/v1 `localTipJar`). Do not edit.

Local development fixture: two templated System Program transfers.
"""

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,
)

LocalTipJarInstanceState = Literal["closed", "open"]
LocalTipJarParticipantState = NoReturn
LocalTipJarEvent = Literal["tipJar.closed"]


class LocalTipJarParams(TypedDict):
    amount: ProtocolInteger


class LocalTipJarCloseArgs(TypedDict):
    amount: ProtocolInteger


LocalTipJarCloseGuardCode = Literal["not_open", "not_recipient"]


LocalTipJarCloseResult = Union[ProtocolPrepared, ProtocolRejection[Union[LocalTipJarCloseGuardCode, ProtocolAccessRejectionCode, ProtocolStateRejectionCode]]]


class LocalTipJarOpenArgs(TypedDict):
    amount: ProtocolInteger
    recipient: str


LocalTipJarOpenGuardCode = Literal["amount_zero"]


LocalTipJarOpenResult = Union[ProtocolPrepared, ProtocolRejection[Union[LocalTipJarOpenGuardCode, ProtocolAccessRejectionCode, ProtocolStateRejectionCode]]]


class LocalTipJarInstanceMetadata(TypedDict):
    pass


class LocalTipJarParticipantMetadata(TypedDict):
    pass


class LocalTipJarWebhookPayload(TypedDict):
    """The JSON body of a `localTipJar` webhook delivery."""

    id: str
    event: LocalTipJarEvent
    deploymentId: str
    instanceId: str
    participantId: Optional[str]
    state: Union[LocalTipJarInstanceState, LocalTipJarParticipantState]
    occurredAt: str


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

LOCAL_TIP_JAR_PROTOCOL: ProtocolDefinition = ProtocolDefinition(
    name="localTipJar",
    version="0.1.0",
    package="local-tip-jar",
    package_release_hash="arete:h1:protocol-release:sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
    protocol_spec_hash="arete:h1:protocol-spec:sha256:f1a6580e1d2259ca7da870e33e0a1920ea0704cf2014efccf29116c4abc06032",
    protocol_release_hash="arete:h1:protocol-release:sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
    deployment=ProtocolDeployment(
        deployment_id="dep-local-1",
        access="private",
        api_url="https://api.example.test/api",
        session_endpoint="https://api.example.test/ws/sessions",
    ),
    instance_visibility="shared",
    steps={
        "close": ProtocolStepDefinition(scope="instance", access="secret", guards=("not_recipient", "not_open"), args=("amount",)),
        "open": ProtocolStepDefinition(scope="create", access="secret", guards=("amount_zero",), args=("amount", "recipient")),
    },
    events={
        "tipJar.closed": "instance",
    },
    stacks=STACKS,
    programs=PROGRAMS,
)


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

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


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

    async def close(
        self,
        *,
        instance: str,
        wallet: str,
        args: LocalTipJarCloseArgs,
        transaction_format: ProtocolTransactionFormat = "v1",
        idempotency_key: Optional[str] = None,
        client_context: Optional[Mapping[str, Any]] = None,
    ) -> LocalTipJarCloseResult:
        """Close the tip jar. Prepare `close` (instance step, secret access)."""
        return await self._client.prepare(
            "close",
            wallet=wallet,
            instance=instance,
            args=decimal_args(args, ("amount",)),
            transaction_format=transaction_format,
            idempotency_key=idempotency_key,
            client_context=client_context,
        )

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


class LocalTipJarClient(ProtocolClient):
    """`localTipJar` 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: LocalTipJarPublicSteps

    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__(
            LOCAL_TIP_JAR_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 = LocalTipJarPublicSteps(self)

    @classmethod
    def from_session(
        cls, session: Any, *, prepare_steps: Optional[Sequence[str]] = None
    ) -> "LocalTipJarClient":
        """A client on `session`'s auth and HTTP client whose instance handles
        follow the session's connected stacks."""
        return cls(**session.protocol_client_options(LOCAL_TIP_JAR_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 LocalTipJarSecretClient(ProtocolClient):
    """`localTipJar` 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: LocalTipJarSecretSteps

    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__(
            LOCAL_TIP_JAR_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(
                "LocalTipJarSecretClient requires a secret key; use LocalTipJarClient with a publishable key",
                "INVALID_CONFIG",
            )
        self.steps = LocalTipJarSecretSteps(self)

    @classmethod
    def from_session(cls, session: Any) -> "LocalTipJarSecretClient":
        """A client on `session`'s auth and HTTP client whose instance handles
        follow the session's connected stacks."""
        options = session.protocol_client_options(LOCAL_TIP_JAR_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_local_tip_jar_webhook(
    secret: Union[str, bytes],
    body: Union[str, bytes],
    headers: Any,
    *,
    now: Optional[int] = None,
    tolerance_seconds: int = 300,
) -> VerifiedWebhook[LocalTipJarWebhookPayload]:
    """Verify a signed webhook delivery for this protocol's events."""
    return verify_webhook(secret, body, headers, now=now, tolerance_seconds=tolerance_seconds)


__all__ = [
    "LOCAL_TIP_JAR_PROTOCOL",
    "LocalTipJarClient",
    "LocalTipJarSecretClient",
    "LocalTipJarPublicSteps",
    "LocalTipJarSecretSteps",
    "LocalTipJarInstanceState",
    "LocalTipJarParticipantState",
    "LocalTipJarEvent",
    "LocalTipJarParams",
    "LocalTipJarInstanceMetadata",
    "LocalTipJarParticipantMetadata",
    "LocalTipJarWebhookPayload",
    "STACKS",
    "PROGRAMS",
    "verify_local_tip_jar_webhook",
    "LocalTipJarCloseArgs",
    "LocalTipJarCloseGuardCode",
    "LocalTipJarCloseResult",
    "LocalTipJarOpenArgs",
    "LocalTipJarOpenGuardCode",
    "LocalTipJarOpenResult",
]