Python SDK

The OpenMetadata Python SDK provides a comprehensive interface for interacting with the OpenMetadata API. It offers type-safe operations for managing metadata entities and seamless integration with your Python applications.

Installation

Install the OpenMetadata Python SDK using pip:

pip install openmetadata-ingestion

Which Python SDK Should I Use?

The openmetadata-ingestion package contains two Python clients. Both talk to the same OpenMetadata API.

	Original client	SDK 2.0
Import	`metadata.ingestion.ometa.ometa_api.OpenMetadata`	`metadata.sdk`
Setup	Build an `OpenMetadataConnection` and pass it to `OpenMetadata(...)`	Call `configure(host=..., jwt_token=...)` once
Style	Generic methods that take an entity type, such as `metadata.get_by_name(entity=Table, fqn=...)`	One class per entity, such as `Tables.retrieve_by_name(fqn)`
Available since	Every release	OpenMetadata 1.10.0

For new scripts and jobs, use SDK 2.0. It needs less setup and reads more simply. Use the original client when you need an operation that SDK 2.0 doesn’t offer yet, or when you maintain existing code that already uses it.

Using Both Clients in One Script

Both ship in the same package, so one script can use both. Set up each one separately: call configure() for SDK 2.0 and build an OpenMetadata client for the original client.

Migrating an Existing Job

Change one call at a time:

Original client	SDK 2.0
`metadata.create_or_update(data=request)`	`Tables.create(request)`
`metadata.get_by_id(entity=Table, entity_id=id)`	`Tables.retrieve(id)`
`metadata.get_by_name(entity=Table, fqn=fqn)`	`Tables.retrieve_by_name(fqn)`
`metadata.list_all_entities(entity=Table)`	`Tables.list_all()`
`metadata.list_entities(entity=Table, params={...})`	`Tables.list(filters={...})`
`metadata.delete(entity=Table, entity_id=id)`	`Tables.delete(id)`

Tables.create(request) also updates an existing table, the same as create_or_update. Three details differ from what you might expect:

In configure(host=...), include /api at the end of the URL, the same as hostPort in the original client.
When you list entities, pass query filters as filters={"databaseSchema": "service.database.schema"} and read the results from .entities.
With the original client, set authProvider="openmetadata" when you authenticate with a JWT token. SDK 2.0 doesn’t need an authProvider.

Quick Start: SDK 2.0

Connect with configure(), then create, update, and delete a table:

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

# The host must end with /api
configure(host="http://localhost:8585/api", jwt_token="your-jwt-token")

# Create
table = Tables.create(
    CreateTableRequest(
        name="customers",
        databaseSchema="mysql-prod.analytics.public",
        columns=[Column(name="id", dataType=DataType.BIGINT)],
    )
)

print(table.fullyQualifiedName.root)

# Update
table.description = "Customer master table"
table = Tables.update(table)

# Delete
Tables.delete(table.id)

Fields such as name, id, and fullyQualifiedName are wrapper objects. Use .root to get the plain value, for example table.fullyQualifiedName.root. Call configure() with no arguments to read OPENMETADATA_HOST and OPENMETADATA_JWT_TOKEN from your environment.

Quick Start: Original Client

Connect with OpenMetadataConnection, then read and update entities:

Basic Connection

from metadata.generated.schema.entity.services.connections.metadata.openMetadataConnection import (
    OpenMetadataConnection,
)
from metadata.ingestion.ometa.ometa_api import OpenMetadata
from metadata.generated.schema.security.client.openMetadataJWTClientConfig import (
    OpenMetadataJWTClientConfig,
)

# Configure connection
server_config = OpenMetadataConnection(
    hostPort="http://localhost:8585/api",
    authProvider="openmetadata",
    securityConfig=OpenMetadataJWTClientConfig(
        jwtToken="your-jwt-token"
    ),
)

# Create OpenMetadata client
metadata = OpenMetadata(server_config)

# Health check
if metadata.health_check():
    print("Successfully connected to OpenMetadata!")

Working with Entities

from metadata.generated.schema.entity.data.table import Table
from metadata.generated.schema.entity.services.databaseService import DatabaseService

# Get all database services
services = list(metadata.list_all_entities(entity=DatabaseService))
print(f"Found {len(services)} database services")

# Get a specific table by name
table = metadata.get_by_name(
    entity=Table,
    fqn="your-service.your-database.your-schema.your-table"
)

if table:
    print(f"Table: {table.name.root}")
    print(f"Columns: {len(table.columns) if table.columns else 0}")

Core Functionality

Entity Management

The Python SDK provides full CRUD operations for all OpenMetadata entities:

Create or Update Entities

from metadata.generated.schema.api.data.createTable import CreateTableRequest

# Create table request
create_table = CreateTableRequest(
    name="sample_table",
    databaseSchema="your_database.your_schema",
    columns=[
        # Define your columns here
    ],
    description="Sample table created via Python SDK"
)

# Create or update the table
table = metadata.create_or_update(data=create_table)

Retrieve Entities

# Get by ID
table = metadata.get_by_id(entity=Table, entity_id="uuid-here")

# Get by fully qualified name
table = metadata.get_by_name(
    entity=Table,
    fqn="service.database.schema.table"
)

# List entities with pagination
tables = metadata.list_entities(
    entity=Table,
    fields=["owners", "tags"],
    limit=50
)

List All Entities

# Generator for large datasets
all_tables = metadata.list_all_entities(
    entity=Table,
    fields=["owners", "columns"],
    params={"databaseSchema": "your-service.your-database.your-schema"}
)

for table in all_tables:
    print(f"Processing table: {table.name.root}")

Entity References

# Get entity reference for relationships
table_ref = metadata.get_entity_reference(
    entity=Table,
    fqn="service.database.schema.table"
)

# Use in other entity creation
if table_ref:
    print(f"Table reference: {table_ref.id.root}")

Advanced Features

Validating a Custom Certificate Authority (CA) Certificate

To trust a private CA, pass the path to its certificate in sslConfig:

from metadata.generated.schema.security.ssl.validateSSLClientConfig import ValidateSslClientConfig

server_config = OpenMetadataConnection(
    hostPort="https://your-openmetadata-instance.com/api",
    authProvider="openmetadata",
    securityConfig=OpenMetadataJWTClientConfig(
        jwtToken="your-jwt-token"
    ),
    verifySSL="validate",
    sslConfig=ValidateSslClientConfig(caCertificate="/path/to/rootCert"),
)

metadata = OpenMetadata(server_config)

Error Handling

# get_by_name returns None when the entity does not exist
table = metadata.get_by_name(entity=Table, fqn="non-existent-table")
if table is None:
    print("Table not found")

Working with Versions

# Get entity version history
versions = metadata.list_versions(
    entity_id="your-entity-id",
    entity=Table
)

print(f"Found {len(versions.versions)} versions")

Common Use Cases

Data Discovery

# Search for tables containing specific keywords
all_tables = metadata.list_all_entities(entity=Table)

matching_tables = [
    table for table in all_tables
    if "customer" in table.name.root.lower()
]

for table in matching_tables:
    print(f"Found customer table: {table.fullyQualifiedName.root}")

Metadata Automation

# Bulk update table descriptions
tables_to_update = metadata.list_all_entities(
    entity=Table,
    params={"databaseSchema": "production-service.production-db.public"}
)

for table in tables_to_update:
    if not table.description:
        # Update with a default description
        table.description = f"Production table: {table.name.root}"
        metadata.create_or_update(data=table)

Lineage Management

# Get table lineage (returns a dict)
lineage = metadata.get_lineage_by_name(
    entity=Table,
    fqn="service.database.schema.table"
)

if lineage:
    print(f"Upstream edges: {len(lineage['upstreamEdges'])}")
    print(f"Downstream edges: {len(lineage['downstreamEdges'])}")

API Reference

The Python SDK provides a comprehensive API based on the OpenMetadata data model:

Core Classes

OpenMetadata: Main client class for API interactions
OpenMetadataConnection: Connection configuration
Entity Classes: Type-safe representations of all OpenMetadata entities

Key Methods

create_or_update(data): Create or update an entity
get_by_name(entity, fqn): Retrieve entity by fully qualified name
get_by_id(entity, entity_id): Retrieve entity by ID
list_entities(entity, **kwargs): List entities with pagination
list_all_entities(entity, **kwargs): Generator for all entities
delete(entity, entity_id): Delete an entity
health_check(): Check API connectivity

Type Safety

The Python SDK is built on generated Pydantic models, providing:

Type hints for better IDE support
Runtime validation of data structures
Auto-completion for entity properties
Error prevention through static typing

# Type-safe entity creation
from metadata.generated.schema.entity.data.table import Table

# IDE will provide auto-completion and type checking
table: Table = metadata.get_by_name(entity=Table, fqn="my.table")
if table:
    columns_count: int = len(table.columns) if table.columns else 0

Best Practices

Connection Management: Reuse OpenMetadata client instances
Error Handling: Always handle potential exceptions
Pagination: Use list_all_entities() for large datasets
Performance: Specify only required fields when fetching entities
Resource Cleanup: Call metadata.close() when done

​Python SDK

​Installation

​Which Python SDK Should I Use?

​Using Both Clients in One Script

​Migrating an Existing Job

​Quick Start: SDK 2.0

​Quick Start: Original Client

​Basic Connection

​Working with Entities

​Core Functionality

​Entity Management

​Create or Update Entities

​Retrieve Entities

​List All Entities

​Entity References

​Advanced Features

​Validating a Custom Certificate Authority (CA) Certificate

​Error Handling

​Working with Versions

​Common Use Cases

​Data Discovery

​Metadata Automation

​Lineage Management

​API Reference

​Core Classes

​Key Methods

​Type Safety

​Best Practices