Python SDK Overview

Full-featured Python client library for OpenMetadata integration

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

Quick Start

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="collate",
    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}")
    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=["owner", "tags"],
    limit=50
)

List All Entities

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

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

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}")

Advanced Features

Custom Configuration

# Using custom authentication
from metadata.generated.schema.security.client.basicAuthenticationProvider import (
    BasicAuthenticationProvider,
)

server_config = OpenMetadataConnection(
    hostPort="https://your-openmetadata-instance.com/api",
    authProvider="basic",
    securityConfig=BasicAuthenticationProvider(
        username="your-username",
        password="your-password"
    ),
)

metadata = OpenMetadata(server_config)

Error Handling

from metadata.ingestion.ometa.ometa_api import (
    OpenMetadata,
    InvalidEntityException,
    EmptyPayloadException
)

try:
    table = metadata.get_by_name(entity=Table, fqn="non-existent-table")
except InvalidEntityException as e:
    print(f"Invalid entity: {e}")
except EmptyPayloadException as e:
    print(f"No data received: {e}")

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.lower()
]

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

Metadata Automation

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

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

Lineage Management

from metadata.generated.schema.type.entityLineage import EntityLineage

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

if lineage:
    print(f"Upstream entities: {len(lineage.upstreamEdges)}")
    print(f"Downstream entities: {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

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