### v2 Worker Setup

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/upgrade-guide-v3.md

In v2, worker setup required explicit registration of workflow and activity classes. This example shows the manual client and worker instantiation.

```python
from application_sdk.worker import Worker
from application_sdk.clients.temporal import TemporalWorkflowClient

client = TemporalWorkflowClient(host=..., namespace=...)
await client.load()

worker = Worker(
    workflow_client=client,
    workflow_classes=[MyWorkflow],
    workflow_activities=[MyActivities()],
    passthrough_modules=["my_connector"],
)
await worker.run()
```

--------------------------------

### Install Atlan Application SDK with uv

Source: https://github.com/atlanhq/application-sdk/blob/main/README.md

Install the SDK using uv, the recommended package manager. Ensure uv is installed in your environment.

```bash
uv add atlan-application-sdk
```

--------------------------------

### Start Connector Application Server

Source: https://github.com/atlanhq/application-sdk/blob/main/tests/integration/_example/README.md

Ensure your connector application is running before executing tests. This command provides an example of how to start a Python-based connector.

```bash
# Example: Start your connector
python your_connector_app.py
```

--------------------------------

### Install Atlan Application SDK with pip

Source: https://github.com/atlanhq/application-sdk/blob/main/README.md

Install the SDK using pip, a standard Python package installer. This is an alternative to using uv.

```bash
pip install atlan-application-sdk
```

--------------------------------

### Generate Single Example

Source: https://github.com/atlanhq/application-sdk/blob/main/contract-toolkit/AGENTS.md

Generate a single example by evaluating a specific .pkl file.

```bash
pkl eval -m examples/trino/generated examples/trino/app.pkl
```

--------------------------------

### Start Workflow with SDK API

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/guides/atlantis.md

Demonstrates how to initiate a workflow using the SDK handler API. This requires a credential GUID and connection details.

```javascript
// Start a workflow
await fetch('/workflows/v1/start', {
  method: 'POST',
  headers: { 'Content-Type': 'application/json' },
  body: JSON.stringify({
    credential_guid: '<guid>',
    connection: {
      connection_name: 'my-db',
      connection_qualified_name: 'default/postgres/1234',
    },
  }),
});
```

--------------------------------

### Workflow Phase 1: Setup

Source: https://github.com/atlanhq/application-sdk/blob/main/application_sdk/common/incremental/skills/implement-incremental-extraction/SKILL.md

The setup phase involves retrieving workflow arguments, fetching the incremental marker, reading the current state, and saving the initial state.

```text
Phase 1: Setup
  get_workflow_args → fetch_incremental_marker → read_current_state → save_state
```

--------------------------------

### Install Conformance Package

Source: https://github.com/atlanhq/application-sdk/blob/main/conformance/README.md

Install the `atlan-application-sdk-conformance` package as a development dependency using `uv`.

```sh
uv add --dev atlan-application-sdk-conformance
```

--------------------------------

### Install Dependencies

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/agents/dev-commands.md

Synchronize and install all project dependencies using uv. This is a crucial step for setting up the development environment.

```bash
uv sync
```

--------------------------------

### Start Workflow via Curl

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/agents/dev-commands.md

Initiate a workflow by sending a POST request to the application's start endpoint, providing the credential GUID obtained from provisioning.

```bash
curl -s -X POST http://localhost:8000/workflows/v1/start \
  -H "Content-Type: application/json" \
  -d '{"credential_guid": "<GUID_FROM_ABOVE>",
       "connection": {"connection_name": "test", "connection_qualified_name": "default/app/1234"}}'
```

--------------------------------

### Example App with Task Abstraction

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/adr/0005-infrastructure-abstraction.md

Demonstrates how to define an application using the `App` and `@task` abstractions. Framework provides replay-safe alternatives for common operations like getting the current time and generating UUIDs.

```python
from application_sdk.app import App, Input, Output, task

class MyPipeline(App):

    @task
    async def fetch_data(self, input: FetchInput) -> FetchOutput:
        # Can do any I/O here - no sandbox restrictions
        return FetchOutput(data=await response.json())

    async def run(self, input: MyInput) -> MyOutput:
        # Use self.now() instead of datetime.now()
        started_at = self.now()
        # Use self.uuid() instead of uuid.uuid4()
        request_id = self.uuid()
        data = await self.fetch_data(FetchInput(url=input.url))
        return MyOutput(data=data.data, started_at=started_at)
```

--------------------------------

### Install and Run App Playground

Source: https://github.com/atlanhq/application-sdk/blob/main/contract-toolkit/CLAUDE.md

Installs the app playground to a specified directory and runs the application. Ensure to run this only after generation and static checks pass.

```bash
git check-ignore app/generated/frontend/static
npx @atlanhq/app-playground@3.1.0 install-to app/generated/frontend/static
atlan app run -p .
```

--------------------------------

### Generate Example Contract

Source: https://github.com/atlanhq/application-sdk/blob/main/contract-toolkit/CLAUDE.md

This command generates a specific example contract using the Pkl evaluation tool. The example directory serves as the root for the consuming app's repository.

```bash
pkl eval -m examples/full examples/full/app.pkl
```

--------------------------------

### Install All Development Dependencies

Source: https://github.com/atlanhq/application-sdk/blob/main/AGENTS.md

Use this command to install all development and testing dependencies, including optional extras. This ensures your environment is fully set up for development and testing.

```bash
uv sync --all-extras --all-groups
```

--------------------------------

### Install PKL

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/guides/pkl-contracts.md

Install the PKL language using Homebrew. Ensure version 0.27 or higher is recommended.

```bash
brew install pkl
pkl --version   # 0.27+ recommended
```

--------------------------------

### v2 Application Entry Point

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/upgrade-guide-v3.md

Example of how an application entry point was configured in v2.

```python
from application_sdk.application.metadata_extraction.sql import (
    BaseSQLMetadataExtractionApplication,
)

app = BaseSQLMetadataExtractionApplication(
    name="my-connector",
    client_class=MyClient,
    handler_class=MyHandler,
)
await app.start()
```

--------------------------------

### Minimal AdvancedJDBCUrlGroup Skeleton Example

Source: https://github.com/atlanhq/application-sdk/blob/main/contract-toolkit/docs/reference.md

Demonstrates the basic setup for an AdvancedJDBCUrlGroup for MSSQL, including URL prefix, host/port fields, and common extra fields like database and SSL enablement. Also shows a 'basic' authentication option with its URL mapping.

```pkl
credentialUrlGroup = new AdvancedJDBCUrlGroup {
  urlAddonBefore = "mssql+pyodbc://"
  hostField = new FieldSpec { name = "host"; displayName = "Host"; width = 6 }
  portField = new FieldSpec { name = "port"; fieldType = "number"; defaultValue = "1433"; width = 2 }
  extraFields {
    new FieldSpec { name = "database" }
    new FieldSpec { name = "__enable_ssl"; fieldType = "select"; options { "yes"; "no" } }
  }
}

credentialAuthOptions {
  ["basic"] = new JDBCUrlAuthOption {
    label = "Basic"
    nestedLabel = "Basic Authentication"
    urlPartMapping = new UrlPartMapping {
      hostname = "host"; port = "port"; pathname = "extra.database"
      searchParams {
        ["Encrypt"] = new UrlRef { field = "extra.__enable_ssl" }
        ["TrustServerCertificate"] = new UrlLiteral { value = "yes" }
      }
    }
    fields {
      new FieldSpec { name = "username" }
      new FieldSpec { name = "password"; sensitive = true }
    }
  }
}
```

--------------------------------

### Install App Playground

Source: https://github.com/atlanhq/application-sdk/blob/main/contract-toolkit/AGENTS.md

Install the app playground to the specified static directory. Ensure this directory is ignored by git.

```bash
git check-ignore app/generated/frontend/static
npx @atlanhq/app-playground@3.1.0 install-to app/generated/frontend/static
```

--------------------------------

### Copy Example Integration Test Structure

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/guides/integration-testing.md

Use this command to copy the example integration test structure to your connector's directory. This sets up the basic files needed for integration testing.

```bash
cp -r tests/integration/_example tests/integration/my_connector
```

--------------------------------

### Athena-style View Lineage Example

Source: https://github.com/atlanhq/application-sdk/blob/main/contract-toolkit/docs/reference.md

Demonstrates how to configure QueryIntelligenceNode and LineageNode for Athena-style view lineage. This example shows setting up dependencies and enabling connection caching.

```pkl
extraNodes {
  ["qi"] = new QueryIntelligenceNode {
    vendorName = "athena"
    sqlKey = "attributes.definition"
    catalogKey = "attributes.databaseName"
    schemaKey = "attributes.schemaName"
    inputPrefix = "$.extract.outputs.transformed_data_prefix"
    outputPrefix = "$.extract.outputs.view_lineage_output_prefix"
  }
  ["publish"] = new PublishNode {
    args {
      ...super.args
      ["connection_cache_enabled"] = true
      ["connection_cache_via_app_enabled"] = true
      ["current_state_via_app_enabled"] = true
    }
  }
  ["lineage-app"] = new LineageNode {
    connectorName = "athena"
    sqlUnquotedCase = "lower"
    ignoreAllCase = false
    dependsOn = null
    dependsOnCondition = new DependencyCondition {
      andConditions {
        new DependencyCondition { nodeId = "qi"; tag = "success" }
        new DependencyCondition { nodeId = "publish"; tag = "success" }
      }
    }
  }
  ["lineage-publish"] = new LineagePublishNode {}
}
```

--------------------------------

### v3 Worker Setup with Direct Handle

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/upgrade-guide-v3.md

In v3, worker setup is automatic via CLI or run_dev_combined(). For direct access, use create_temporal_client and create_worker. Explicit client host/namespace can be passed.

```python
from application_sdk.execution import create_temporal_client, create_worker

client = await create_temporal_client(host="localhost:7233")  # pass host/namespace explicitly

# All registered App subclasses auto-discovered — no explicit list
worker = create_worker(client)
await worker.run()
```

--------------------------------

### Start Temporal Development Server

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/guides/getting-started.md

Starts a local Temporal development server with a SQLite database. This is required for running applications in combined mode.

```bash
temporal server start-dev --db-filename temporal.db
```

--------------------------------

### Start Temporal Development Server

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/guides/build-your-first-app.md

Starts the Temporal development server in a dedicated terminal. This command initializes a local database file for Temporal.

```bash
# Terminal 1
temporal server start-dev --db-filename temporal.db
```

--------------------------------

### Install uv and Python

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/setup/MAC.md

Installs uv, a Python package and environment manager, and a specific Python version. Activates the virtual environment and verifies the Python installation.

```bash
# Install UV
curl -LsSf https://astral.sh/uv/0.7.3/install.sh | sh

# Install Python 3.11.10
uv venv --python 3.11.10

# activate the venv
source .venv/bin/activate

# Verify installation
python --version
```

--------------------------------

### Starting Application with MCP Enabled

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/concepts/mcp.md

Run your main application script to start the MCP server. Ensure the ENABLE_MCP environment variable is set to true.

```bash
python main.py
```

--------------------------------

### Dockerfile for Incremental SQL Connector

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/upgrade-guide-v3.md

Example Dockerfile for setting up an incremental SQL connector.

```dockerfile
FROM registry.atlan.com/public/app-runtime-base:3
RUN uv pip install "atlan-application-sdk[incremental]"
ENV ATLAN_APP_MODULE=app.app:MyIncrementalConnector
```

--------------------------------

### Example Emission with All Labels

Source: https://github.com/atlanhq/application-sdk/blob/main/application_sdk/common/incremental/skills/incremental-migrate/references/metrics-template.md

This example demonstrates emitting a counter metric with all common labels, including connection qualified name, workflow ID, and workflow run ID. Ensure all required labels are included for proper routing and correlation.

```python
emit(
    "incremental_updated_count",
    len(changed_ids),
    metric_type=MetricType.COUNTER,
    connection_qualified_name=config.connection_qualified_name,
    workflow_id=workflow_info.workflow_id,
    workflow_run_id=workflow_info.run_id,
)
```

--------------------------------

### Example Integration Test Directory Structure

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/guides/integration-testing.md

Illustrates a typical directory layout for integration tests, including shared fixtures, reference examples, and connector-specific test directories.

```text
tests/integration/
├── __init__.py
├── conftest.py              # Shared fixtures
├── README.md
├── _example/                # Reference example
│   ├── __init__.py
│   ├── conftest.py
│   ├── scenarios.py
│   ├── test_integration.py
│   └── README.md
└── my_connector/            # Your connector tests
    ├── __init__.py
    ├── conftest.py
    ├── scenarios.py
    └── test_integration.py
```

--------------------------------

### Install and Run Reactor Components

Source: https://github.com/atlanhq/application-sdk/blob/main/remediation/README.md

Install necessary Reactor packages, scaffold a new project, compile the Directed Acyclic Graph (DAG), and run the conformance remediation process using the Reactor framework.

```bash
# Install (dev-only):
npm i -D @openprose/reactor @openprose/reactor-cli @openprose/reactor-devtools

# First-time scaffold:
npx reactor init remediation

# Compile DAG:
npx reactor compile

# Run:
npx reactor run conformance-remediation scope=. mode=default
```

--------------------------------

### Dockerfile for API-based Connector

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/upgrade-guide-v3.md

Example Dockerfile for setting up an API-based connector without SQL dependencies.

```dockerfile
FROM registry.atlan.com/public/app-runtime-base:3
# ... (see Dockerfile for full setup)
RUN uv pip install atlan-application-sdk
ENV ATLAN_APP_MODULE=app.app:MyOpenApiApp
```

--------------------------------

### Initialize New Project with uv

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/guides/getting-started.md

Use `uv init` to create a new project directory, `cd` into it, and then add the `atlan-application-sdk` dependency.

```bash
uv init my-app
cd my-app
uv add atlan-application-sdk
```

--------------------------------

### Initialize Project and Add Dependencies

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/guides/rest-api-application-guide.md

Initialize a new project using uv and add the necessary SDK and HTTP client dependencies.

```bash
uv init my-connector && cd my-connector
uv add "atlan-application-sdk"
uv add httpx  # or aiohttp, requests — your HTTP client of choice
mkdir -p app
```

--------------------------------

### Initialize Project and Add Dependencies

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/guides/build-your-first-app.md

Use `uv` to initialize a new Python project and add the necessary SDK and HTTP client libraries. Then, create the application directory structure.

```bash
uv init github-connector
cd github-connector
uv add "atlan-application-sdk" httpx
mkdir app
touch app/__init__.py
```

--------------------------------

### Run Local Development Combined Mode

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/guides/sql-application-guide.md

Starts both the Temporal worker and the HTTP handler in a single process for local development. Ensure you have the `application_sdk` installed.

```python
# main.py
import asyncio
from application_sdk.main import run_dev_combined
from app.connector import PostgresApp

asyncio.run(run_dev_combined(PostgresApp))
```

--------------------------------

### v3 Local Development Entry Point

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/upgrade-guide-v3.md

Example of how to run the application in combined mode for local development in v3.

```python
from application_sdk.main import run_dev_combined

asyncio.run(run_dev_combined(MyMetadataExtractor))
```

--------------------------------

### Implement HTTP API Client

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/guides/rest-api-application-guide.md

Create an asynchronous HTTP client to interact with a REST API. This example uses httpx for making GET requests and handling authentication.

```python
from __future__ import annotations

import httpx

class MyApiClient:
    def __init__(self, api_token: str, base_url: str = "https://api.example.com") -> None:
        self._client = httpx.AsyncClient(
            base_url=base_url,
            headers={"Authorization": f"Bearer {api_token}"},
            timeout=30,
        )

    async def get_entities(self, page: int = 1) -> list[dict]:
        response = await self._client.get("/v1/entities", params={"page": page})
        response.raise_for_status()
        return response.json()["data"]

    async def close(self) -> None:
        await self._client.aclose()
```

--------------------------------

### Trigger Workflow via HTTP POST

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/guides/rest-api-application-guide.md

Send an HTTP POST request to start a workflow. This example demonstrates how to structure the request payload, including connection details and a credential identifier.

```bash
curl -X POST http://localhost:8000/workflows/v1/start \
  -H "Content-Type: application/json" \
  -d '{
    "credential_guid": "test-local-cred",
    "connection": {
      "qualifiedName": "default/my-connector/1234567890",
      "name": "my-test-connection"
    }
  }'

```

--------------------------------

### Basic Logging Example

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/standards/logging.md

Demonstrates how to obtain a logger and perform basic informational logging. Ensure `get_logger` is imported from `application_sdk.observability`.

```python
from application_sdk.observability import get_logger

logger = get_logger(__name__)

# Basic logging
logger.info("Operation completed successfully")
```

--------------------------------

### Implement Postgres Handler

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/guides/sql-application-guide.md

Implement the Handler ABC to define endpoints for connector setup and operation. This example shows how to handle authentication, preflight checks, and metadata fetching for a PostgreSQL database.

```python
from application_sdk.handler import Handler
from application_sdk.handler.contracts import (
    AuthInput,
    AuthOutput,
    AuthStatus,
    PreflightInput,
    PreflightOutput,
    PreflightStatus,
    MetadataInput,
    MetadataOutput,
    SqlMetadataObject,
    SqlMetadataOutput,
)
from app.clients import PostgresClient
from app.contracts import MyCredential


class PostgresHandler(Handler):
    async def test_auth(self, input: AuthInput) -> AuthOutput:
        """Verify that the provided credentials can connect to the database."""
        try:
            cred = MyCredential.from_list(input.credentials)
            client = PostgresClient()
            await client.load(credentials=cred.to_dict())
            return AuthOutput(status=AuthStatus.SUCCESS, message="Connected")
        except Exception as e:
            return AuthOutput(status=AuthStatus.FAILED, message=str(e))

    async def preflight_check(self, input: PreflightInput) -> PreflightOutput:
        """Run pre-extraction checks (connectivity, permissions, table counts)."""
        return PreflightOutput(status=PreflightStatus.READY, message="All checks passed")

    async def fetch_metadata(self, input: MetadataInput) -> SqlMetadataOutput:
        """Return catalog/schema pairs for the Atlan UI filter tree."""
        cred = MyCredential.from_list(input.credentials)
        client = PostgresClient()
        await client.load(credentials=cred.to_dict())

        rows = []
        async for batch in client.run_query(
            "SELECT catalog_name AS table_catalog, schema_name AS table_schema "
            "FROM information_schema.schemata "
            "WHERE schema_name NOT LIKE 'pg_%' "
            "AND schema_name != 'information_schema'"
        ):
            rows.extend(batch)
        # BaseSQLClient.run_query lowercases column names, so dict keys are lowercase.
        # SqlMetadataObject keeps the uppercase field names from the wire schema.
        objects = [
            SqlMetadataObject(TABLE_CATALOG=r["table_catalog"], TABLE_SCHEMA=r["table_schema"])
            for r in rows
        ]
        return SqlMetadataOutput(objects=objects)
```

--------------------------------

### Bundle Output Layout Example

Source: https://github.com/atlanhq/application-sdk/blob/main/contract-toolkit/docs/reference.md

Illustrates the directory structure and file organization for a multi-entrypoint bundle, including atlan.yaml, app.yaml, and generated artifacts.

```text
atlan.yaml                              # bundle-level (entrypoints listed)
app.yaml
app/generated/
├── atlan-connectors-{shared}.json      # hoisted from entrypoints, deduped by filename
├── {entrypoint1}/
│   ├── __init__.py
│   ├── {name}.json
│   ├── manifest.json
│   └── _input.py
└── {entrypoint2}/
    └── ...

```

--------------------------------

### Typical Query-History DAG Shape with PopularityNode

Source: https://github.com/atlanhq/application-sdk/blob/main/contract-toolkit/docs/reference.md

Illustrates a common DAG structure for query history processing, including QueryIntelligenceNode, PublishNode, and PopularityNode. This example shows the typical setup and interconnections between these nodes.

```pkl
extraNodes {
  ["qi"] = new QueryIntelligenceNode {
    vendorName = "snowflake"
    sqlKey = "QUERY_TEXT"
    inputPrefix = "$.extract.outputs.query_history_prefix"
    outputPrefix = "$.extract.outputs.qi_output_prefix"
  }
  ["publish"] = new PublishNode {
    args {
      ["connection_cache_enabled"] = true
      ["connection_cache_via_app_enabled"] = true
    }
  }
  ["popularity"] = new PopularityNode {
    tenantId = "$.extract.outputs.tenant_id"
    parsedDataPrefix = "$.extract.outputs.qi_output_prefix"
    minedDataPrefix = "$.extract.outputs.query_history_prefix"
    connectionCachePath = "$.extract.outputs.connection_cache_path"
    outputPrefix = "$.extract.outputs.popularity_output_prefix"
  }
}
```

--------------------------------

### v2 BaseApplication + Explicit Worker Setup

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/whats-new-v3.md

Demonstrates the older method of setting up an application and worker explicitly using BaseSQLMetadataExtractionApplication and Worker classes.

```python
from application_sdk.application.metadata_extraction.sql import (
    BaseSQLMetadataExtractionApplication,
)
from application_sdk.worker import Worker
from application_sdk.clients.temporal import TemporalWorkflowClient

# Wiring the application
app = BaseSQLMetadataExtractionApplication(
    name="my-connector",
    client_class=MyClient,
    handler_class=MyHandler,
)
await app.setup_workflow(
    workflow_and_activities_classes=[(MyWorkflow, MyActivities)]
)
await app.start()

# Or manually:
client = TemporalWorkflowClient(host=host, namespace=namespace)
await client.load()
worker = Worker(
    workflow_client=client,
    workflow_classes=[MyWorkflow],
    workflow_activities=[MyActivities()],
    passthrough_modules=["my_connector"],
)
await worker.run()
```

--------------------------------

### Test Secret Store - Get Secret

Source: https://github.com/atlanhq/application-sdk/blob/main/components/README.md

Retrieve a secret from the Dapr secret store component. This example fetches the secret associated with the key 'HOMEBREW_CELLAR', assuming it's configured to use system environment variables.

```bash
curl http://localhost:3500/v1.0/secret/secretstore/HOMEBREW_CELLAR
```

--------------------------------

### Example BaseClient Subclass for API Integration

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/concepts/clients.md

Demonstrates subclassing BaseClient to create a custom client for a specific API. Includes methods for fetching data via GET and creating resources via POST, with authentication and content type headers set in the load method.

```python
from typing import Dict, Any
from application_sdk.clients import BaseClient

class MyApiClient(BaseClient):
    async def load(self, **kwargs: Any) -> None:
        """Initialize the client with credentials and set up HTTP headers."""
        credentials = kwargs.get("credentials", {})

        # Set up authentication headers
        self.http_headers = {
            "Authorization": f"Bearer {credentials.get('api_token')}",
            "User-Agent": "MyApp/1.0",
            "Content-Type": "application/json"
        }

        # Optionally set up custom retry transport for advanced retry logic
        # from httpx_retries import Retry, RetryTransport
        # retry = Retry(total=5, backoff_factor=10, status_forcelist=[429, 500, 502, 503, 504])
        # self.http_retry_transport = RetryTransport(retry=retry)

    async def fetch_data(self, endpoint: str, params: Dict[str, Any] = None) -> Dict[str, Any]:
        """Custom method to fetch data from the API."""
        response = await self.execute_http_get_request(
            url=f"https://api.example.com/{endpoint}",
            params=params
        )
        if response and response.status_code == 200:
            return response.json()
        return {}

    async def create_resource(self, endpoint: str, data: Dict[str, Any]) -> Dict[str, Any]:
        """Custom method to create a resource via POST."""
        response = await self.execute_http_post_request(
            url=f"https://api.example.com/{endpoint}",
            json_data=data
        )
        if response and response.status_code == 201:
            return response.json()
        return {}

```

--------------------------------

### Dockerfile for Application SDK v3

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/guides/sql-application-guide.md

Example Dockerfile using the application-sdk v3 base image. It sets up dependencies, copies application code, and configures essential environment variables for runtime mode selection. No custom entrypoint or command is needed.

```dockerfile
# Application-sdk v3 base image (Chainguard-based)
FROM registry.atlan.com/public/app-runtime-base:3

WORKDIR /app

# Install dependencies first (better caching)
COPY --chown=appuser:appuser pyproject.toml uv.lock README.md ./
RUN --mount=type=cache,target=/home/appuser/.cache/uv,uid=1000,gid=1000 \
    uv venv .venv && \
    uv sync --locked --no-install-project

# Copy application code
COPY --chown=appuser:appuser . .

# App-specific environment variables
ENV ATLAN_HANDLER_PORT=8000
ENV ATLAN_APP_MODULE=app.connector:PostgresApp
ENV ATLAN_CONTRACT_GENERATED_DIR=app/generated
```

--------------------------------

### Install uv with sudo for permission issues

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/setup/troubleshooting.md

Run this command with sudo if you encounter permission errors during uv installation. This grants necessary privileges for the installation.

```bash
curl -LsSf https://astral.sh/uv/0.7.3/install.sh | sudo sh
```

--------------------------------

### Client Construction with Credentials

Source: https://github.com/atlanhq/application-sdk/blob/main/tools/migrate_v3/MIGRATION_PROMPT.md

Demonstrates how to construct an HTTP client by handling both `credential_guid` from production and `credentials` dictionary from local development environments.

```python
async def _get_or_create_client(self, input) -> MyClient:
    if hasattr(input, "credential_guid") and input.credential_guid:
        creds = await self.task_context.get_secret(input.credential_guid)
    else:
        creds = input.credentials
    client = MyClient()
    await client.load(credentials=creds)
    return client
```

--------------------------------

### Install uv and Python on Windows

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/setup/WINDOWS.md

Use this PowerShell command to install uv version 0.7.3 and Python 3.11.10. The SDK auto-manages Dapr and Temporal, so no separate installation is needed for those.

```powershell
# Install UV
powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/0.7.3/install.ps1 | iex"

# Install Python 3.11.10
uv venv --python 3.11.10

# Verify installation
uv run python --version
```

--------------------------------

### Dockerfile for SQL Connector

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/upgrade-guide-v3.md

Example Dockerfile for setting up a SQL connector, including necessary SQL dependencies.

```dockerfile
FROM registry.atlan.com/public/app-runtime-base:3
RUN uv pip install "atlan-application-sdk[sql]"
ENV ATLAN_APP_MODULE=app.app:MyDatabaseConnector
```

--------------------------------

### Install uv and Python 3.11.10

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/setup/LINUX.md

Installs the 'uv' package manager and a specific Python version (3.11.10) using 'uv'. Activates the virtual environment and verifies the Python installation.

```bash
# Install UV
curl -LsSf https://astral.sh/uv/0.7.3/install.sh | sh

# Install Python 3.11.10
uv venv --python 3.11.10

# activate the venv
source .venv/bin/activate

# Verify installation
python --version # Should show Python 3.11.10
```

--------------------------------

### Setup SDK Logger

Source: https://github.com/atlanhq/application-sdk/blob/main/tools/migrate_v3/MIGRATION_PROMPT.md

Import and initialize the SDK's logger in every module that needs to log. This ensures Temporal/tenant context is injected and logs are routed correctly.

```python
from application_sdk.observability.logger_adaptor import get_logger

logger = get_logger(__name__)
```

--------------------------------

### Install Development Dependencies

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/guides/build-your-first-app.md

Install necessary development dependencies for testing, including pytest, pytest-asyncio, and pytest-httpx.

```bash
uv add --dev pytest pytest-asyncio pytest-httpx
```

--------------------------------

### Define a Simple Application with Tasks

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/guides/getting-started.md

Create a Python application defining input, output, and a task. This example shows how to structure an app with a `greet` task and a `run` method.

```python
from application_sdk.app import App, task
from application_sdk.contracts import Input, Output


class HelloInput(Input):
    name: str = "world"


class HelloOutput(Output):
    message: str = ""


class HelloApp(App):
    @task(timeout_seconds=60)
    async def greet(self, input: HelloInput) -> HelloOutput:
        msg = f"Hello, {input.name}!"
        self.logger.info("greeting name=%s", input.name)
        return HelloOutput(message=msg)

    async def run(self, input: HelloInput) -> HelloOutput:
        return await self.greet(input)
```

--------------------------------

### Direct Dependency Condition Example

Source: https://github.com/atlanhq/application-sdk/blob/main/contract-toolkit/docs/reference.md

Example of setting a direct dependency condition on a specific node and tag.

```pkl
dependsOnCondition = new DependencyCondition {
  nodeId = "extract"
  tag = "success"
}
```

--------------------------------

### Install Homebrew

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/setup/MAC.md

Use this command to install Homebrew, the package manager for macOS. Follow any post-installation instructions provided in your terminal.

```bash
/bin/bash -c "$(curl -fsSL https://raw.githubusercontent.com/Homebrew/install/HEAD/install.sh)"
```

--------------------------------

### Install Pre-commit Hooks

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/agents/dev-commands.md

Install the Git pre-commit hooks. These hooks will automatically run on each commit to enforce code standards.

```bash
uv run pre-commit install
```

--------------------------------

### Proper Configuration Management

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/standards/review-checklist.md

Demonstrates correct environment variable handling with validation and default values. Use this pattern to ensure configuration is robust and predictable.

```Python
# REJECT: Poor configuration management
max_locks = os.getenv("MAX_LOCKS")  # No validation, no defaults
REDIS_ERROR = "true"  # Hardcoded, not configurable

# REQUIRE: Proper configuration management
FAIL_WORKFLOW_ON_REDIS_UNAVAILABLE = os.getenv("FAIL_WORKFLOW_ON_REDIS_UNAVAILABLE", "false").lower() == "true"
max_locks = int(os.getenv("MAX_LOCKS", "10"))  # Default with validation
```

--------------------------------

### OR Condition with Failure Tags Example

Source: https://github.com/atlanhq/application-sdk/blob/main/contract-toolkit/docs/reference.md

Example demonstrating an OR condition where the dependency is met if either the 'left' or 'right' node fails.

```pkl
dependsOnCondition = new DependencyCondition {
  orConditions {
    new DependencyCondition { nodeId = "left"; tag = "failure" }
    new DependencyCondition { nodeId = "right"; tag = "failure" }
  }
}
```

--------------------------------

### Custom Handler Implementation Example

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/concepts/handlers.md

Example of defining a custom handler by subclassing Handler and implementing specific methods like test_auth.

```python
from application_sdk.handler import Handler, AuthInput, AuthOutput

class MyAppHandler(Handler):   # name must be {AppClassName}Handler
    async def test_auth(self, input: AuthInput) -> AuthOutput:
        ... 
```

--------------------------------

### Programmatic Entry Point (Local Dev)

Source: https://github.com/atlanhq/application-sdk/blob/main/tools/migrate_v3/MIGRATION_PROMPT.md

Replace the v2 `BaseXxxApplication` instantiation with `run_dev_combined` for local development. This function handles the application startup and execution.

```python
from application_sdk.main import run_dev_combined
import asyncio

asyncio.run(run_dev_combined(MyExtractor, handler_class=MyHandler))
```

--------------------------------

### Install System Dependencies

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/setup/LINUX.md

Installs essential build dependencies required for the Application SDK on Debian/Ubuntu systems. Ensure you have sudo privileges before running.

```bash
sudo apt-get update
sudo apt-get install -y make build-essential libssl-dev zlib1g-dev \
libbz2-dev libreadline-dev libsqlite3-dev wget curl llvm \
libncurses5-dev libncursesw5-dev xz-utils tk-dev libffi-dev \
liblzma-dev
```

--------------------------------

### Nested AND/OR Dependency Condition Example

Source: https://github.com/atlanhq/application-sdk/blob/main/contract-toolkit/docs/reference.md

Example of a complex, nested dependency condition combining AND and OR logic for multiple nodes and success tags.

```pkl
dependsOnCondition = new DependencyCondition {
  andConditions {
    new DependencyCondition { nodeId = "prepare"; tag = "success" }
    new DependencyCondition {
      orConditions {
        new DependencyCondition { nodeId = "primary"; tag = "success" }
        new DependencyCondition { nodeId = "fallback"; tag = "success" }
      }
    }
  }
}
```

--------------------------------

### Run Full Conformance Suite

Source: https://github.com/atlanhq/application-sdk/blob/main/conformance/README.md

Execute the full conformance suite, detecting issues in the repository and outputting a SARIF report.

```sh
uv run atlan-application-sdk-conformance detect --repo . --output report.sarif
```

--------------------------------

### Setup and Cleanup Test Environment

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/guides/integration-testing.md

Use class methods to set up the test environment before tests run and clean up any created test data afterward.

```python
@classmethod
def setup_test_environment(cls):
    cls.test_data = create_test_data()

@classmethod
def cleanup_test_environment(cls):
    delete_test_data(cls.test_data)
```

--------------------------------

### v2 Credential Handling with GUID

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/whats-new-v3.md

Shows the v2 approach to handling credentials using a credential GUID and a raw dictionary, typically within an activity.

```python
class ExtractionInput(Input, allow_unbounded_fields=True):
    credential_guid: str = ""

# In an activity:
from application_sdk.services.secretstore import SecretStore
credentials = await SecretStore.get_credentials({"credential_guid": input.credential_guid})
await client.load(credentials)  # dict[str, Any]
```

--------------------------------

### Backward Compatibility: Credential Input Examples

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/upgrade-guide-v3.md

Shows how both legacy `credential_guid` and the new `credential_ref` are supported for backward compatibility.

```python
# These are both valid:
ExtractionInput(credential_guid="abc-123")       # legacy — still works
ExtractionInput(credential_ref=api_key_ref("prod-key"))  # new typed path
```

--------------------------------

### SQL Query Optimization Examples

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/standards/performance.md

Illustrates best practices for writing efficient SQL queries, including selecting specific columns, using LIMIT, employing parameterized queries for security and performance, and leveraging WHERE clauses on indexed columns.

```sql
# Use specific columns instead of SELECT *
query = "SELECT id, name, created_at FROM users WHERE status = 'active' LIMIT 1000"

```

```sql
# Use parameterized queries for security and performance
query = "SELECT * FROM users WHERE age > %s AND city = %s LIMIT %s"
params = (18, 'New York', 100)

```

```sql
# Use appropriate indexes in WHERE clauses
query = "SELECT * FROM orders WHERE user_id = %s AND created_at > %s"

```

--------------------------------

### CLI Modes for Application SDK

Source: https://github.com/atlanhq/application-sdk/blob/main/docs/guides/sql-application-guide.md

Demonstrates how to use the application-sdk CLI in different modes: combined for local development, and separate worker/handler modes for production. The `--app` flag is optional in production.

```bash
# Local development
application-sdk --mode combined --app app.connector:PostgresApp

# Production (separate pods)
application-sdk --mode worker
application-sdk --mode handler
```

--------------------------------

### Run-Level Completion Finalizer Condition Example

Source: https://github.com/atlanhq/application-sdk/blob/main/contract-toolkit/docs/reference.md

Example of setting a run-level completion condition that triggers on any terminal state (success or failure). This condition does not require a specific nodeId.

```pkl
dependsOnCondition = new DependencyCondition {
  tag = "workflow_complete"
}
```