Tables

The Tables class provides CRUD operations for database table metadata, plus table-specific helpers for tagging, column descriptions, custom metrics, and sample data. FQN format: service_name.database_name.schema_name.table_name

Import

from metadata.sdk import configure, Tables
from metadata.generated.schema.api.data.createTable import CreateTableRequest
from metadata.generated.schema.entity.data.table import Column, DataType

configure(host="https://your-instance.getcollate.io/api", jwt_token="your-token")

Create

table = Tables.create(
    CreateTableRequest(
        name="orders",
        databaseSchema="my_postgres.analytics.public",
        columns=[
            Column(name="id", dataType=DataType.BIGINT),
            Column(name="customer_id", dataType=DataType.BIGINT),
            Column(name="total", dataType=DataType.DOUBLE),
        ],
        description="Customer order records",
    )
)
print(table.fullyQualifiedName)  # "my_service.my_db.public.orders"

Tip: create() is idempotent — it calls createOrUpdate under the hood. If a table with the same FQN already exists, it is updated.

Retrieve

By ID

table = Tables.retrieve("550e8400-e29b-41d4-a716-446655440000")

By fully-qualified name

table = Tables.retrieve_by_name("my_service.my_db.public.orders")

With specific fields

table = Tables.retrieve_by_name(
    "my_service.my_db.public.orders",
    fields=["columns", "tags", "owners", "followers"],
)

Available fields include: columns, tags, owners, followers, domain, description, sampleData, tableConstraints, customMetrics.

List

Single page

page = Tables.list(limit=25)
for table in page.entities:
    print(table.name)

# Fetch next page
if page.after:
    next_page = Tables.list(limit=25, after=page.after)

All tables (auto-paginated)

all_tables = Tables.list_all(batch_size=100)
print(f"Total tables: {len(all_tables)}")

With filters

tables = Tables.list_all(
    fields=["tags", "owners"],
    filters={"database": "my_service.my_db"},
)

filters are sent to the Tables list endpoint as query parameters. The endpoint accepts only the following filters:

service
database
databaseSchema
databaseSchemaRegex
tableRegex
include
includeEmptyTestSuite
regexFilterByFqn
regexMode

Filtering by tag isn’t supported. To find tables by tag, see Filter by Tag.

Update

table = Tables.retrieve_by_name("my_service.my_db.public.orders")
table.description = "Updated description"
Tables.update(table)

update() computes a JSON Patch diff between the fetched entity and your modified version.

Delete

# Soft delete (recoverable)
Tables.delete("table-uuid")

# Soft delete with child entities
Tables.delete("table-uuid", recursive=True)

# Permanent hard delete
Tables.delete("table-uuid", hard_delete=True)

Search

results = Tables.search("orders", size=10)
for table in results:
    print(table.fullyQualifiedName)

Tables.search() matches only the table name or fully qualified name. It doesn’t search tags or other fields.

Filter by Tag

To find tables with a specific tag or glossary term, use the Search API with a tags.tagFQN filter against the table search index. The following example collects every table tagged PII.Sensitive and pages through the results with from_ and size:

from metadata.sdk.api import Search

results = []
offset = 0
page_size = 100

while True:
    page = Search.search(
        query="*",
        index="table_search_index",
        filters={"tags.tagFQN": "PII.Sensitive"},
        from_=offset,
        size=page_size,
    )
    hits = page.get("hits", {}).get("hits", [])
    results.extend(hit["_source"] for hit in hits)
    if len(hits) < page_size:
        break
    offset += page_size

for table in results:
    print(table["fullyQualifiedName"])

The search returns JSON documents, not Table objects. Each _source holds the table’s indexed fields. To search other asset types, use the matching search index, such as dashboard_search_index.

Table-Specific Operations

Add a tag

Attach a classification or glossary tag by its FQN:

Tables.add_tag("table-uuid", tag_fqn="PII.Sensitive")
Tables.add_tag("table-uuid", tag_fqn="Glossary.Finance.Revenue")

Update a column description

Tables.update_column_description(
    "table-uuid",
    column_name="customer_id",
    description="Foreign key referencing the customers table",
)

Add a custom metric

from metadata.generated.schema.api.tests.createCustomMetric import CreateCustomMetricRequest

Tables.add_custom_metric(
    "table-uuid",
    custom_metric=CreateCustomMetricRequest(
        name="revenue_sum",
        expression="SUM(total)",
        description="Total revenue across all orders",
        columnName="total",
    ),
)

Add sample data

from metadata.generated.schema.entity.data.table import TableData

Tables.add_sample_data(
    "table-uuid",
    sample_data=TableData(
        columns=["id", "customer_id", "total"],
        rows=[[1, 42, 99.99], [2, 17, 249.00]],
    ),
)

Get sample data

table = Tables.get_sample_data("table-uuid")
print(table.sampleData)

CSV Export / Import

# Export table definitions to CSV
csv_content = Tables.export_csv("my_service.my_db.public").execute()

# Import table definitions from CSV
Tables.import_csv("my_service.my_db.public").with_data(csv_content).execute()

# Dry run to preview import without committing
Tables.import_csv("my_service.my_db.public").with_data(csv_content).set_dry_run(True).execute()

Version History

# All versions
versions = Tables.get_versions("table-uuid")

# Specific version
v = Tables.get_specific_version("table-uuid", "0.2")

Restore

# Restore a soft-deleted table
Tables.restore("table-uuid")