Metadata-Version: 2.3
Name: pragmatiks-sdk
Version: 21.1.0
Summary: Python SDK for the Pragmatiks platform
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3.13
Classifier: Typing :: Typed
Requires-Dist: httpx>=0.28.1
Requires-Dist: packaging>=21.0
Requires-Dist: pydantic>=2.11.4
Requires-Dist: pydantic-settings>=2.9.1
Requires-Dist: pyyaml>=6.0
Requires-Dist: structlog>=24.1.0
Requires-Python: >=3.13
Description-Content-Type: text/markdown

<p align="center">
  <img src="assets/wordmark.png" alt="Pragmatiks" width="800">
</p>

# Pragma SDK

[![Ask DeepWiki](https://deepwiki.com/badge.svg)](https://deepwiki.com/pragmatiks/pragma-sdk)
[![PyPI version](https://img.shields.io/pypi/v/pragmatiks-sdk.svg)](https://pypi.org/project/pragmatiks-sdk/)
[![Python 3.13+](https://img.shields.io/badge/python-3.13+-blue.svg)](https://www.python.org/downloads/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)
[![Code style: ruff](https://img.shields.io/badge/code%20style-ruff-000000.svg)](https://github.com/astral-sh/ruff)

**[Documentation](https://docs.pragmatiks.io/sdk/overview)** | **[CLI](https://github.com/pragmatiks/pragma-cli)** | **[Providers](https://github.com/pragmatiks/pragma-providers)**

Build providers and interact with Pragmatiks programmatically.

## Quick Start

### Synchronous Client

```python
from pragma_sdk import PragmaClient, Resource

with PragmaClient() as client:
    # Apply a resource
    client.apply_resource(
        Resource(
            provider="gcp", resource="storage", name="my-bucket", config={"location": "US", "storage_class": "STANDARD"}
        )
    )

    # Get resource status
    bucket = client.get_resource("gcp", "storage", "my-bucket")
    print(bucket.outputs)
```

### Asynchronous Client

```python
import asyncio
from pragma_sdk import AsyncPragmaClient, Resource


async def main():
    async with AsyncPragmaClient() as client:
        # Apply a resource
        await client.apply_resource(
            Resource(
                provider="gcp",
                resource="storage",
                name="my-bucket",
                config={"location": "US", "storage_class": "STANDARD"},
            )
        )

        # Get resource status
        bucket = await client.get_resource("gcp", "storage", "my-bucket")
        print(bucket.outputs)


asyncio.run(main())
```

## Installation

```bash
pip install pragmatiks-sdk
```

Or with uv:

```bash
uv add pragmatiks-sdk
```

## Features

- **HTTP Clients** - Sync and async clients for the Pragmatiks API
- **Provider Authoring** - Build custom providers with type-safe Config and Outputs
- **Provider Publishing and Installation** - Publish provider versions, install, upgrade and downgrade them, and read an installation's history
- **Dead-Letter Queue** - Inspect and delete failed events for debugging and recovery
- **Field References** - Reference outputs from other resources dynamically
- **Testing Harness** - Test provider lifecycle methods locally without deployment
- **Auto-discovery** - Automatic credential resolution from environment or config files

## Building Providers

Define resources with typed configuration and lifecycle methods:

```python
from pragma_sdk import Provider, Resource, Config, Outputs, Field

gcp = Provider()


class BucketConfig(Config):
    location: Field[str]
    storage_class: Field[str] = "STANDARD"


class BucketOutputs(Outputs):
    url: str
    created_at: str


@gcp.resource("storage")
class Bucket(Resource[BucketConfig, BucketOutputs]):
    async def on_create(self) -> BucketOutputs:
        """Provision the bucket."""
        return BucketOutputs(url=f"gs://{self.name}", created_at="...")

    async def on_observe(self) -> BucketOutputs | None:
        """Look the bucket up by identity; None when it does not exist."""
        return BucketOutputs(url=f"gs://{self.name}", created_at="...")

    async def on_update(self, previous_config: BucketConfig | None) -> BucketOutputs:
        """Converge to self.config; previous_config is None when unknown."""
        return BucketOutputs(url=f"gs://{self.name}", created_at="...")

    async def on_delete(self) -> None:
        """Delete the bucket; succeeds when it is already gone."""
```

### Declaring the provider

A provider wheel declares itself with one entry point in the `pragma.provider` group, named after the provider and pointing at its importable package:

```toml
[project.entry-points."pragma.provider"]
gcp = "gcp_provider"
```

The Pragmatiks platform finds the provider's resource types through this entry point. Only resource classes defined inside that package belong to the provider; resource classes it imports from another provider stay that provider's.

A resource always names the provider whose package defines its class. When a provider builds or applies a resource of a type it imports from another provider, that resource, its identity and its owner references name the other provider, and that provider manages it. Declare that provider as a dependency, so it is embedded, and install it in the organization too: the organization's installation of it is what manages those resources.

To check what the platform will read from an installed wheel, run introspection in its environment:

```bash
PRAGMA_PROVIDER_DISTRIBUTION=<distribution> python -m pragma_sdk.introspection --output report.json
```

`<distribution>` is the `name` in the wheel's `[project]` table. When you publish, the wheel is renamed to `<organization>-<provider name you publish under>` before it is uploaded, and the platform loads it under that name.

The report lists the resource types with their schemas, or why the provider failed to load.

### Reading other versions' data

`Outputs` ignore properties the class does not declare, so outputs a newer version of the provider recorded still read. `Config` stays strict for the resource being processed. A dependency's config is read with the unknown top-level fields of a strict `Config` ignored, so a dependency managed by a newer version of its provider still resolves.

## Field References

Reference outputs from other resources:

```python
from pragma_sdk import FieldReference

config = AppConfig(
    database_url=FieldReference(provider="postgres", resource="database", name="my-db", field="outputs.connection_url")
)
```

## Testing Providers

Test lifecycle methods locally with `ProviderHarness`:

```python
from pragma_sdk.provider import ProviderHarness


async def test_bucket_creation():
    harness = ProviderHarness()

    result = await harness.invoke_create(Bucket, name="test-bucket", config=BucketConfig(location="US"))

    assert result.success
    assert "gs://test-bucket" in result.outputs.url
```

Inside the harness, a resource of a class from the tested class's top-level package carries the harness's `provider_name` (`test/local` by default). When the tested class builds resources of an embedded provider's types, name that provider's package the way the platform does, or reading their `provider` or `id` raises:

```python
harness = ProviderHarness(package_catalog_names={"gcp_provider": "pragmatiks/gcp"})
```

## Authentication

Credentials are discovered automatically in this order:

1. Explicit `auth_token` parameter
2. Context-specific environment variable: `PRAGMA_AUTH_TOKEN_{CONTEXT}` (e.g., `PRAGMA_AUTH_TOKEN_PRODUCTION`)
3. Generic environment variable: `PRAGMA_AUTH_TOKEN`
4. Credentials file: `~/.config/pragma/credentials`

The context is determined by: explicit `context` parameter > `PRAGMA_CONTEXT` env var > CLI config > `"default"`.

```python
# Auto-discover credentials (uses default context)
client = PragmaClient()

# Explicit token
client = PragmaClient(auth_token="sk_...")

# Use a specific context (checks PRAGMA_AUTH_TOKEN_PRODUCTION first)
client = PragmaClient(context="production")

# Require authentication (fail if no token)
client = PragmaClient(require_auth=True)
```

## API Reference

### HTTP Client Methods

Both `PragmaClient` (sync) and `AsyncPragmaClient` (async) provide the same methods, except where noted below.

#### Resources

Resource operations live on a project handle: `client.project("my-project").list_resources()`.

| Method | Description |
|--------|-------------|
| `list_resources(provider, resource, tags)` | List resources with optional filters |
| `get_resource(provider, resource, name)` | Get a specific resource |
| `apply_resource(resource)` | Create or update a resource |
| `deactivate_resource(provider, resource, name, dry_run=False)` | Deactivate a resource; returns the teardown's impact list |
| `delete_resource(provider, resource, name, dry_run=False)` | Delete a resource; returns the teardown's impact list |
| `wait_ready(provider, resource, name, timeout=300)` | Poll until READY; `timeout=0` or `None` waits forever |
| `wait_deactivated(provider, resource, name, timeout=300)` | Poll until DRAFT; `timeout=0` or `None` waits forever |
| `wait_deleted(provider, resource, name, timeout=300)` | Poll until DELETED; an unknown resource raises instead of returning; `timeout=0` or `None` waits forever |

#### Lifecycle Events

On the client itself, `AsyncPragmaClient` only.

| Method | Description |
|--------|-------------|
| `stream_lifecycle_events(timeout=None)` | Yield lifecycle event frames from the API's SSE stream; `timeout=0` or `None` streams forever |

#### Providers

| Method | Description |
|--------|-------------|
| `list_providers(query, scope, tags, limit, offset)` | Browse and search the provider catalog |
| `get_provider(provider_name)` | Get a provider from the catalog |
| `list_provider_versions(provider_name)` | List a provider's versions |
| `get_provider_version(provider_name, version)` | Get one version of a provider, in any status |
| `publish_provider_version(wheel_path, changelog=None)` | Rename, declare and upload a provider wheel |
| `delete_provider(provider_name)` | Delete a provider from the catalog |

#### Installed Providers

| Method | Description |
|--------|-------------|
| `install_provider(provider_name, version=None, config=None)` | Install a provider; installing a failed installation again reuses its stored version and config |
| `list_installations()` | List installed providers with their state, latest operation and its target version, and failure message |
| `upgrade_provider(provider_name, target_version=None)` | Upgrade an installed provider (latest version if none) |
| `downgrade_provider(provider_name, target_version)` | Downgrade an installed provider |
| `uninstall_provider(provider_name, cascade=False)` | Uninstall a provider |
| `list_provider_history(provider_name)` | List an installation's history, newest first; kept 30 days, also after an uninstall |

#### Dead-Letter Queue

| Method | Description |
|--------|-------------|
| `list_dead_letter_events(provider)` | List dead letter events with optional provider filter |
| `get_dead_letter_event(event_id)` | Get a dead letter event by ID |
| `delete_dead_letter_event(event_id)` | Delete a single dead letter event |
| `delete_dead_letter_events(provider, all)` | Delete multiple dead letter events |

#### User & Health

| Method | Description |
|--------|-------------|
| `is_healthy()` | Check API health |
| `get_me()` | Get current authenticated user information |

### Provider Classes

| Class | Description |
|-------|-------------|
| `Provider()` | Resource grouping with `@provider.resource()` decorator |
| `Resource[ConfigT, OutputsT]` | Base class with `on_create`, `on_observe`, `on_update`, `on_delete` |
| `computed = True` | Class attribute for a resource with no external object; exempts it from `on_observe` |
| `Config` | Base class for resource configuration (Pydantic model) |
| `Outputs` | Base class for resource outputs (Pydantic model) |
| `Field[T]` | Type alias for `T | FieldReference` |
| `ProviderHarness` | Local testing harness |

## Development

```bash
# Install the toolchain, dependencies and git hooks
mise run install

# Format code
mise run format

# Type check and lint
mise run check
```

## License

MIT
