✦TL;DR

Learn how to crawl Cloud SQL for PostgreSQL assets and publish them to Atlan for discovery.

||

Your AI can read this via Docs MCPcurl -fsSL "https://docs.atlan.com/install-docs-mcp" | bashConnect

Cloud SQL for PostgreSQL assets app

The Cloud SQL for PostgreSQL assets app crawls Cloud SQL for PostgreSQL databases, schemas, tables, views, and columns and publishes them to Atlan. Build it with the AtlanCloudsqlPostgresCrawler builder, which mirrors the "new app" wizard: Credential → Connection → Metadata.

Creating an app creates a new connection

Each create mints a new connection and new assets. To re-crawl, re-run the existing workflow (see Re-run an existing app).

Cloud SQL for PostgreSQL supports four authentication methods: basic (username/password), IAM user, IAM service account, and Workload Identity Federation. The port is optional and defaults to 5432.

Basic authentication

Python

Cloud SQL for PostgreSQL crawling with basic auth

from pyatlan.client.atlan import AtlanClient
from pyatlan.model.apps import AtlanCloudsqlPostgresCrawler

client = AtlanClient()

response = (
    AtlanCloudsqlPostgresCrawler(client)
    .basic( # (1)
        username="atlan_user", # (2)
        password="••••••", # (3)
        host="203.0.113.10", # (4)
        database="postgres", # (5)
    )
    .connection( # (6)
        name="production-cloudsql",
        admin_roles=[client.role_cache.get_id_for_name("$admin")],
    )
    .include_metadata({"postgres": ["analytics", "sales"]}) # (7)
    .run(name="cloudsql-prod") # (8)
)
print(response.slug, response.run_id)

1. Step 1—Credential. Username/password auth; the secret is vaulted.
2. Required. Database username.
3. Required. Password.
4. Required. The public IP address of the Cloud SQL instance. The port (port=) is optional and defaults to 5432.
5. Required. Database to connect to.
6. Step 2—Connection. Display name + at least one admin.
7. Step 3—Metadata. Schemas to crawl, as {database: [schema, ...]}. Omit to crawl everything. Exclude takes priority over include.
8. .run(name=...) creates and submits a run; use .create(name=...) to create without running.

IAM authentication

For the IAM methods, Atlan connects through the Cloud SQL Language Connector, so name is the instance connection name from the instance's Overview page (<project>:<region>:<instance>). The IAM identity must also exist as a database user on the instance.

Python

Cloud SQL for PostgreSQL with IAM auth

# IAM user
AtlanCloudsqlPostgresCrawler(client).iam_user(
    username="analyst@example.com", # (1)
    password="ya29.••••••", # (2)
    name="my-project:us-central1:my-instance", # (3)
    database="postgres",
)

# IAM service account
AtlanCloudsqlPostgresCrawler(client).iam_service_account(
    username="atlan-crawler@my-project.iam", # (4)
    password="••••••", # (5)
    name="my-project:us-central1:my-instance",
    gcp_service_account='{"type": "service_account", ...}', # (6)
    database="postgres",
)

# Workload Identity Federation
AtlanCloudsqlPostgresCrawler(client).workload_identity_federation(
    username="atlan-crawler@my-project.iam", # (7)
    name="my-project:us-central1:my-instance",
    database="postgres",
)

1. Required. The IAM user's email address, without a trailing .gserviceaccount.com.
2. Required. A short-lived access token for that user (gcloud auth print-access-token --account=<iam-user>). It expires in about an hour, so prefer a service account for scheduled workflows.
3. Required. The instance connection name.
4. Required. The service account's email address, without a trailing .gserviceaccount.com.
5. Required. The password field of the IAM service account form.
6. Optional. The service account's JSON key, as a string. Omit it for keyless setups where the environment can obtain access tokens.
7. Required. The service account the Self-Deployed Runtime's federation is configured for. Available only on a Self-Deployed Runtime on Google Cloud.

Configuration options

All metadata options are optional—set only what you need:

Python

Cloud SQL for PostgreSQL metadata configuration

(
    AtlanCloudsqlPostgresCrawler(client)
    .basic(username="atlan_user", password="••••••", host="...", database="postgres")
    .connection(name="production-cloudsql", admin_roles=[...])
    .include_metadata({"postgres": ["analytics"]}) # (1)
    .exclude_metadata({"postgres": ["staging"]}) # (2)
    .exclude_regex_for_tables_views(".*_TMP|.*_TEMP|TMP.*|TEMP.*") # (3)
    .redact_sql_text(True) # (4)
    .cloud_provider("AWS") # (5)
    .run(name="cloudsql-prod")
)

1. Databases/schemas to include, as {database: [schema, ...]}.
2. Databases/schemas to exclude—exclude takes priority over include.
3. Regex of tables/views to ignore.
4. Replace literal values in extracted view, function, and procedure definitions with placeholders. Off by default. See Query redaction.
5. The cloud the Atlan tenant runs on, as the UI sets it for the same workflow.

Update existing workflows

Publish a new version on the same slug with the builder's load() / update(). It reads the workflow's current inputs, lets you change only the fields you pass, and preserves everything else, including the connection and its credential. No run starts; the next run, manual or scheduled, uses the new version.

Python

Turn on Redact SQL Text via the builder

(
    AtlanCloudsqlPostgresCrawler(client)
    .load("atlan-cloudsql-postgres-1700000000-AbC123") # (1)
    .redact_sql_text(True) # (2)
    .update()
)

1. Seed the builder from the workflow's current inputs.
2. Change only the fields you need; anything unset keeps its current value.

To apply one change across every Cloud SQL for PostgreSQL crawler and miner on the tenant, loop over the workflows:

Python

Turn on Redact SQL Text across all Cloud SQL for PostgreSQL workflows

from pyatlan.model.apps import AtlanCloudsqlPostgresCrawler, CloudsqlPostgresMiner

BUILDERS = {
    "atlan-cloudsql-postgres": AtlanCloudsqlPostgresCrawler, # (1)
    "cloudsql-postgres-miner": CloudsqlPostgresMiner, # (2)
}

cursor = None
while True:
    page = client.app.get_all(limit=100, cursor=cursor)
    for wf in page.workflows:
        app_id = (wf.dag.get("extract") or {}).get("inputs", {}).get("app_id")
        builder = BUILDERS.get(app_id)
        if builder is None:
            continue
        builder(client).load(wf.slug).redact_sql_text(True).update() # (3)
    if not page.has_more:
        break
    cursor = page.next_cursor

1. Crawler workflows.
2. Miner workflows. Redact SQL Text is a separate setting on each.
3. One new workflow version per workflow, with only the toggle changed. Running the loop again is harmless.

See Manage apps for the rest of the lifecycle—list, get, run status, schedules, and delete—shared by every app.