"""GCP Secret Manager resource - complete annotated example."""
from __future__ import annotations
import json
from typing import Any, ClassVar
from google.api_core.exceptions import AlreadyExists, NotFound
from google.cloud import secretmanager
from google.oauth2 import service_account
from pragma_sdk import Config, Outputs, Resource
# -----------------------------------------------------------------------------
# Config: What users provide
# -----------------------------------------------------------------------------
class SecretConfig(Config):
"""Configuration for a GCP Secret Manager secret.
Design decisions:
- project_id and secret_id are required because they define the secret's
identity in GCP. These cannot change after creation.
- data is the secret payload. Simple string type handles most use cases.
- credentials is required (no ADC fallback) because this is a multi-tenant
SaaS platform. Each user operates in their own GCP project with their
own service account.
"""
# Identity fields (immutable after creation)
project_id: str
secret_id: str
# Payload
data: str
# Authentication - accepts dict or JSON string for flexibility
# Users typically pass this via a FieldReference to a pragma/secret
credentials: dict[str, Any] | str
# -----------------------------------------------------------------------------
# Outputs: What dependents can reference
# -----------------------------------------------------------------------------
class SecretOutputs(Outputs):
"""Outputs from GCP Secret Manager secret creation.
Design decisions:
- resource_name is the full GCP path. Dependents need this to access
the secret via GCP APIs.
- version_name includes the version so dependents can pin to specific
versions if needed.
- version_id is just the number for display/logging purposes.
"""
resource_name: str # projects/{project}/secrets/{id}
version_name: str # projects/{project}/secrets/{id}/versions/{version}
version_id: str # Just the version number ("1", "2", etc.)
# -----------------------------------------------------------------------------
# Resource: The lifecycle implementation
# -----------------------------------------------------------------------------
class Secret(Resource[SecretConfig, SecretOutputs]):
"""GCP Secret Manager secret resource.
Pattern: Cloud resource with user-provided credentials
This pattern is common when:
- Users manage resources in their own cloud accounts
- The platform doesn't have ambient credentials (no Workload Identity)
- Multi-tenant isolation requires per-user authentication
"""
# Class-level metadata for provider registration
provider: ClassVar[str] = "gcp"
resource: ClassVar[str] = "secret"
# -------------------------------------------------------------------------
# Helper Methods
# -------------------------------------------------------------------------
def _get_client(self) -> secretmanager.SecretManagerServiceClient:
"""Get authenticated client using user-provided credentials.
Why a method instead of an instance variable?
- Clients are created fresh for each lifecycle call
- Avoids serialization issues with stored credentials
- Handles both dict and JSON string credentials formats
"""
creds_data = self.config.credentials
# Support JSON-encoded credentials (common when passed via env vars)
if isinstance(creds_data, str):
creds_data = json.loads(creds_data)
credentials = service_account.Credentials.from_service_account_info(
creds_data
)
return secretmanager.SecretManagerServiceClient(credentials=credentials)
def _secret_path(self) -> str:
"""Build the full GCP resource path.
Extracted to a method because it's used in multiple lifecycle handlers.
"""
return f"projects/{self.config.project_id}/secrets/{self.config.secret_id}"
# -------------------------------------------------------------------------
# Lifecycle: on_create
# -------------------------------------------------------------------------
async def on_create(self) -> SecretOutputs:
"""Create GCP secret with initial version.
Idempotency strategy: "Create-or-get"
- Try to create the secret
- If it already exists (AlreadyExists exception), get the existing one
- Always add a new version with the payload
This handles retry scenarios where:
1. First call: Creates secret + version 1
2. Retry (if first call crashed before ack): Gets existing + version 2
The version increment on retry is acceptable because secret versions
are append-only in GCP anyway.
"""
client = self._get_client()
parent = f"projects/{self.config.project_id}"
# Step 1: Create or get the secret
try:
secret = client.create_secret(
request={
"parent": parent,
"secret_id": self.config.secret_id,
"secret": {"replication": {"automatic": {}}},
}
)
except AlreadyExists:
# Secret exists from a previous attempt - this is fine
secret = client.get_secret(name=self._secret_path())
# Step 2: Add version with payload (always runs)
version = client.add_secret_version(
request={
"parent": secret.name,
"payload": {"data": self.config.data.encode("utf-8")},
}
)
return SecretOutputs(
resource_name=secret.name,
version_name=version.name,
version_id=version.name.split("/")[-1],
)
# -------------------------------------------------------------------------
# Lifecycle: on_update
# -------------------------------------------------------------------------
async def on_update(self, previous_config: SecretConfig) -> SecretOutputs:
"""Update secret by creating new version if data changed.
Update strategies used here:
1. Immutable field check: project_id and secret_id cannot change
because they define the resource's identity in GCP. Changing them
would require delete + recreate.
2. No-op optimization: If data hasn't changed, return existing outputs
without making any API calls. This is important because on_update
may be called even when only unrelated fields change.
3. Append-only update: Secrets use versioning, so "update" means
adding a new version rather than modifying in place.
"""
# Validate immutable fields
if previous_config.project_id != self.config.project_id:
raise ValueError(
"Cannot change project_id; delete and recreate resource"
)
if previous_config.secret_id != self.config.secret_id:
raise ValueError(
"Cannot change secret_id; delete and recreate resource"
)
# No-op if data unchanged (important for efficiency)
if previous_config.data == self.config.data and self.outputs is not None:
return self.outputs
# Add new version
client = self._get_client()
version = client.add_secret_version(
request={
"parent": self._secret_path(),
"payload": {"data": self.config.data.encode("utf-8")},
}
)
return SecretOutputs(
resource_name=self._secret_path(),
version_name=version.name,
version_id=version.name.split("/")[-1],
)
# -------------------------------------------------------------------------
# Lifecycle: on_delete
# -------------------------------------------------------------------------
async def on_delete(self) -> None:
"""Delete secret and all versions.
Idempotency strategy: "Delete-if-exists"
- Try to delete
- If not found (NotFound exception), succeed anyway
This handles:
- Normal deletion: Deletes the secret
- Retry after successful delete: NotFound, returns success
- Resource never existed: NotFound, returns success
"""
client = self._get_client()
try:
client.delete_secret(name=self._secret_path())
except NotFound:
# Already deleted - idempotent success
pass
