Provider Examples

"""GCP Secret Manager resource - complete annotated example."""

import json
from typing import Any

from google.api_core.exceptions import AlreadyExists, NotFound
from google.cloud.secretmanager_v1 import AccessSecretVersionResponse, SecretManagerServiceAsyncClient
from google.oauth2 import service_account
from pragma_sdk import Config, Field, ImmutableField, Outputs, Resource


class SecretConfig(Config):
    """Configuration for a GCP Secret Manager secret.

    ``project_id`` and ``secret_id`` locate the secret in GCP, so they are
    immutable: the platform refuses a change to either, and no handler checks
    them. ``data`` changes in place by adding a version. ``credentials`` is
    required because each user operates in their own GCP project with their
    own service account; there are no ambient credentials to fall back on.

    Attributes:
        project_id: GCP project that holds the secret.
        secret_id: Identifier of the secret within the project.
        data: Payload stored as the latest version.
        credentials: Service account key as a dict or a JSON string, usually a
            FieldReference to a ``pragma/secret`` resource.
    """

    project_id: ImmutableField[str]
    secret_id: ImmutableField[str]
    data: Field[str]
    credentials: Field[dict[str, Any] | str]


class SecretOutputs(Outputs):
    """Outputs of a GCP Secret Manager secret.

    Outputs carry identifiers only; the payload stays in the secret.

    Attributes:
        resource_name: Full GCP path, ``projects/{project}/secrets/{id}``.
            Dependents read the secret through GCP APIs with it.
        version_name: Full path of the latest version, so dependents can pin
            it. ``None`` while the secret holds no version.
        version_id: Number of the latest version, for display and logging.
            ``None`` while the secret holds no version.
    """

    resource_name: str
    version_name: str | None
    version_id: str | None


class Secret(Resource[SecretConfig, SecretOutputs]):
    """GCP Secret Manager secret resource.

    Pattern: cloud resource with user-provided credentials. It fits when users
    manage resources in their own cloud accounts, the platform has no ambient
    credentials, and multi-tenant isolation requires per-user authentication.
    """

    def build_client(self) -> SecretManagerServiceAsyncClient:
        """Build an async client authenticated with the user's service account.

        Each handler builds its own client, so credentials never outlive the
        call. The async client lets every GCP call yield to the event loop.

        Returns:
            Secret Manager client using the user's credentials.
        """
        credentials_info = self.config.credentials
        if isinstance(credentials_info, str):
            credentials_info = json.loads(credentials_info)

        credentials = service_account.Credentials.from_service_account_info(credentials_info)
        return SecretManagerServiceAsyncClient(credentials=credentials)

    def build_secret_path(self) -> str:
        """Build the full GCP path of the secret from its locator fields.

        Returns:
            Path in the form ``projects/{project}/secrets/{id}``.
        """
        return f"projects/{self.config.project_id}/secrets/{self.config.secret_id}"

    def build_outputs(self, version_name: str | None) -> SecretOutputs:
        """Build outputs for the secret and its latest version.

        Args:
            version_name: Full path of the latest version, or ``None`` when the
                secret holds no version.

        Returns:
            Outputs for the secret.
        """
        return SecretOutputs(
            resource_name=self.build_secret_path(),
            version_name=version_name,
            version_id=version_name.split("/")[-1] if version_name else None,
        )

    async def fetch_latest_version(
        self, client: SecretManagerServiceAsyncClient
    ) -> AccessSecretVersionResponse | None:
        """Read the latest version of the secret, payload included.

        Args:
            client: Authenticated Secret Manager client.

        Returns:
            The latest version with its resolved name, or ``None`` when the
            secret holds no version.
        """
        try:
            return await client.access_secret_version(name=f"{self.build_secret_path()}/versions/latest")
        except NotFound:
            return None

    async def add_version(self, client: SecretManagerServiceAsyncClient) -> SecretOutputs:
        """Store ``config.data`` as a new version of the secret.

        Args:
            client: Authenticated Secret Manager client.

        Returns:
            Outputs pointing at the new version.
        """
        version = await client.add_secret_version(
            request={
                "parent": self.build_secret_path(),
                "payload": {"data": self.config.data.encode("utf-8")},
            }
        )
        return self.build_outputs(version.name)

    async def on_observe(self) -> SecretOutputs | None:
        """Report the secret at the path its locator fields name.

        Locates the secret from config alone and reads only GCP, never
        ``self.outputs``. The platform acts on what it reports: ``None`` leads
        to ``on_create``, outputs lead to ``on_update``.

        Returns:
            Outputs for the secret and its latest version, or ``None`` when the
            secret does not exist.
        """
        client = self.build_client()
        try:
            await client.get_secret(name=self.build_secret_path())
        except NotFound:
            return None

        latest = await self.fetch_latest_version(client)
        return self.build_outputs(latest.name if latest else None)

    async def on_create(self) -> SecretOutputs:
        """Create the secret with its first version.

        Called only after ``on_observe`` reports the secret absent. A secret
        that appears between the observation and this call is adopted instead
        of failing on ``AlreadyExists``.

        Returns:
            Outputs pointing at the first version.
        """
        client = self.build_client()
        try:
            await client.create_secret(
                request={
                    "parent": f"projects/{self.config.project_id}",
                    "secret_id": self.config.secret_id,
                    "secret": {"replication": {"automatic": {}}},
                }
            )
        except AlreadyExists:
            pass

        return await self.add_version(client)

    async def on_update(self, previous_config: SecretConfig | None) -> SecretOutputs:
        """Converge the latest version on ``config.data``.

        Compares against the live latest version rather than
        ``previous_config``, so the same path handles a changed payload, an
        unchanged one, and ``previous_config=None`` (the secret exists but the
        platform holds no prior configuration for it). Secret versions are
        append-only, so an update adds a version only when the payload differs.

        Args:
            previous_config: Configuration last applied, or ``None``.

        Returns:
            Outputs pointing at the latest version.
        """
        client = self.build_client()
        latest = await self.fetch_latest_version(client)
        if latest and latest.payload.data == self.config.data.encode("utf-8"):
            return self.build_outputs(latest.name)

        return await self.add_version(client)

    async def on_delete(self) -> None:
        """Delete the secret and all its versions.

        Succeeds when the secret is already gone, so a repeated delete after a
        crash converges instead of failing.
        """
        try:
            await self.build_client().delete_secret(name=self.build_secret_path())
        except NotFound:
            return

"""Slack webhook resource - external API integration pattern."""

import httpx
from pragma_sdk import Config, Field, HealthStatus, Outputs, Resource, SensitiveField


class SlackChannelConfig(Config):
    """Configuration for a Slack notification channel.

    ``webhook_url`` is sensitive because users store webhook URLs in
    ``pragma/secret`` resources and reference them. ``channel`` and
    ``icon_emoji`` are optional because Slack webhooks carry their own
    defaults.

    Attributes:
        webhook_url: Incoming webhook URL, usually a FieldReference to a secret.
        channel: Channel that overrides the webhook's default channel.
        username: Display name of the bot.
        icon_emoji: Emoji shown as the bot's icon.
    """

    webhook_url: SensitiveField[str]
    channel: Field[str] | None = None
    username: Field[str] = "Pragmatiks"
    icon_emoji: Field[str] | None = ":robot:"


class SlackChannelOutputs(Outputs):
    """Outputs of a Slack channel setup.

    Outputs carry no sensitive data; the webhook URL stays in config.

    Attributes:
        channel_id: Identifier derived from the resource name and channel.
    """

    channel_id: str


class SlackChannel(Resource[SlackChannelConfig, SlackChannelOutputs]):
    """Slack notification channel resource.

    Pattern: external API without server-side state. Slack keeps no object
    for this configuration, so there is nothing to observe: the type declares
    ``computed = True``, needs no ``on_observe``, and derives its outputs from
    config alone. Whether the webhook still works changes without a platform
    write, so ``health()`` reports it instead of outputs.
    """

    computed = True

    def build_outputs(self) -> SlackChannelOutputs:
        """Build outputs from the resource name and configured channel.

        Returns:
            Outputs holding an identifier unique within the project's resources.
        """
        channel = self.config.channel or "default"
        return SlackChannelOutputs(channel_id=f"slack-{self.name}-{channel}")

    def build_test_message(self) -> dict[str, str]:
        """Build the payload of the test message.

        Returns:
            Slack webhook payload carrying the configured display settings.
        """
        payload = {
            "text": f"Channel check: {self.name}",
            "username": self.config.username,
        }
        if self.config.channel:
            payload["channel"] = self.config.channel
        if self.config.icon_emoji:
            payload["icon_emoji"] = self.config.icon_emoji
        return payload

    async def on_create(self) -> SlackChannelOutputs:
        """Derive the channel outputs from config.

        Returns:
            Outputs holding the channel identifier.
        """
        return self.build_outputs()

    async def on_update(self, previous_config: SlackChannelConfig | None) -> SlackChannelOutputs:
        """Derive the channel outputs from the current config.

        Args:
            previous_config: Configuration last applied, or ``None``.

        Returns:
            Outputs holding the channel identifier.
        """
        return self.build_outputs()

    async def on_delete(self) -> None:
        """Nothing to delete: Slack keeps no object for this configuration."""

    async def health(self) -> HealthStatus:
        """Report whether Slack accepts messages on the webhook.

        Each check posts a test message to the channel. A webhook that worked
        before may have been revoked since, so the result is never cached.

        Returns:
            Healthy when Slack accepted the test message, unhealthy otherwise.
        """
        async with httpx.AsyncClient() as client:
            try:
                response = await client.post(self.config.webhook_url, json=self.build_test_message(), timeout=10.0)
            except httpx.HTTPError as error:
                return HealthStatus(status="unhealthy", message=f"Webhook unreachable: {type(error).__name__}")
        if response.status_code != 200:
            return HealthStatus(status="unhealthy", message=f"Slack rejected the test message: HTTP {response.status_code}")
        return HealthStatus(status="healthy")