Python SDK

Overview

We are now going to present a high-level Python API as a type-safe and gentle wrapper for the OpenMetadata backend.

The Python SDK is part of the openmetadata-ingestion base package. You can install it from PyPI.
Make sure to use the same openmetadata-ingestion version as your server version. For example, if you have the OpenMetadata server at version 2.0.5, you need to install:

pip install "openmetadata-ingestion~=2.0.5.0"

In the OpenMetadata Design, we have been dissecting the internals of OpenMetadata. The main conclusion here is twofold:

Everything is handled via the API, and
Data structures (Entity definitions) are at the heart of the solution.

This means that whenever we need to interact with the metadata system or develop a new connector or logic, we have to make sure that we pass the proper inputs and handle the types of outputs.

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.

SDK 2.0 Quick Start

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.

Introducing the Python API (Original Client)

Let’s suppose that we have our local OpenMetadata server running at http:localhost:8585. We can play with it with simple cURL or httpie commands, and if we just want to take a look at the Entity instances we have lying around, that might probably be enough. However, let’s imagine that we want to create or update an ML Model Entity with a PUT. To do so, we need to make sure that we are providing a proper JSON, covering all the attributes and types required by the Entity definition. By reviewing the JSON Schema for the create operation and the fields definitions of the Entity, we could come up with a rather simple description of a toy ML Model:

{
    "name": "my-model",
    "description": "sample ML Model",
    "algorithm": "regression",
    "mlFeatures": [
        {
            "name": "age",
            "dataType": "numerical",
            "featureSources": [
                {
                    "name": "age",
                    "dataType": "integer"
                }
            ]
        },
        {
            "name": "persona",
            "dataType": "categorical",
            "featureSources": [
                {
                    "name": "age",
                    "dataType": "integer"
                },
                {
                    "name": "education",
                    "dataType": "string"
                }
            ],
            "featureAlgorithm": "PCA"
        }
    ],
    "mlHyperParameters": [
        {
            "name": "regularisation",
            "value": "0.5"
        }
    ]
}

If we needed to repeat this process with a full-fledged model that is built ad-hoc and updated during the CICD process, we would just be adding a hardly maintainable, error-prone requirement to our production deployment pipelines. The same would happen if, inside the actual OpenMetadata code, there was not a way to easily interact with the API and make sure that we send proper data and can safely process the outputs.

Using Generated Sources

As OpenMetadata is a data-centric solution, we need to make sure we have the right ingredients at all times. That is why we have developed a high-level Python API, using pydantic models automatically generated from the JSON Schemas.

OBS: If you are using a published version of the Ingestion Framework, you are already good to go, as we package the code with the metadata.generated module. If you are developing a new feature, you can get more information here.

This API wrapper helps developers and consumers in:

Validating data during development and with specific error messages at runtime,
Receiving typed responses to ease further processing.

Thanks to the recursive model setting of pydantic the example above can be rewritten using only Python classes, and thus being able to get help from IDEs and the Python interpreter. We can rewrite the previous JSON as:

from metadata.generated.schema.api.data.createMlModel import CreateMlModelRequest

from metadata.generated.schema.entity.data.mlmodel import (
    FeatureSource,
    FeatureSourceDataType,
    FeatureType,
    MlFeature,
    MlHyperParameter,
    MlModel,
)

model = CreateMlModelRequest(
    name="test-model-properties",
    algorithm="algo",
    service="demo-ml-service",
    mlFeatures=[
        MlFeature(
            name="age",
            dataType=FeatureType.numerical,
            featureSources=[
                FeatureSource(
                    name="age",
                    dataType=FeatureSourceDataType.integer,
                )
            ],
        ),
        MlFeature(
            name="persona",
            dataType=FeatureType.categorical,
            featureSources=[
                FeatureSource(
                    name="age",
                    dataType=FeatureSourceDataType.integer,
                ),
                FeatureSource(
                    name="education",
                    dataType=FeatureSourceDataType.string,
                ),
            ],
            featureAlgorithm="PCA",
        ),
    ],
    mlHyperParameters=[
        MlHyperParameter(name="regularisation", value="0.5"),
    ],
)

One syntax to rule them all

Now that we know how to directly use the pydantic models, we can start showcasing the solution. This module has been built with two main principles in mind:

Reusability: We should be able to support existing and new entities with minimum effort,
Extensibility: However, we are aware that not all Entities are the same. Some of them may require specific functionalities or slight variations (such as Lineage or Location), so it should be easy to identify those special methods and create new ones when needed.

To this end, we have the main class OpenMetadata (source) based on Python’s TypeVar. Thanks to this we can exploit the complete power of the pydantic models, having methods with Type Parameters that know how to respond to each Entity. At the same time, we have the Mixins (source) module, with special extensions to some Entities.

Walkthrough

Let’s use Python’s API to create, update and delete a Table Entity. Choosing the Table is a nice starter, as its attributes define the following hierarchy:

DatabaseService -> Database -> Schema -> Table

This will help us showcase how we can reuse the same syntax with the three different Entities.

1. Initialize OpenMetadata

OpenMetadata is the class holding the connection to the API and handling the requests. We can instantiate this by passing the proper configuration to reach the server API:

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

server_config = OpenMetadataConnection(
    hostPort="https://your-company.open-metadata.org/api",
    authProvider=AuthProvider.openmetadata,
    securityConfig=OpenMetadataJWTClientConfig(
        jwtToken="<YOUR-INGESTION-BOT-JWT-TOKEN>",
    ),
)
metadata = OpenMetadata(server_config)

For local development, we can get a JWT token for the ingestion bot as described here and use that when we specify the jwtToken. For a real-world deployment, we can also use different authentication methods and specify other settings of the connection (such as sslConfig). For the OpenMetadataConnection client, use verifySSL="validate" with sslConfig.caCertificate to validate an HTTPS server using your CA certificate. The fluent SDK (metadata.sdk.configure) uses separate options: verify_ssl=True verifies the server certificate against the system trust store, or against ca_bundle when supplied. OpenMetadata 2.0.3 fixes a bug that disabled verification when verify_ssl=True. Install the matching 2.0.3 ingestion package to receive this fix. The following example configures certificate validation with OpenMetadataConnection:

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


server_config = OpenMetadataConnection(
    hostPort="https://your-company.open-metadata.org/api",
    authProvider=AuthProvider.openmetadata,
    securityConfig=OpenMetadataJWTClientConfig(
        jwtToken="<YOUR-INGESTION-BOT-JWT-TOKEN>",
    ),
    verifySSL="validate",  # Validate using the CA certificate below
    sslConfig=ValidateSslClientConfig(caCertificate="/path/to/rootCert"),
)

metadata = OpenMetadata(server_config)

The OpenMetadataConnection is defined as a JSON Schema as well. You can check the definition here

From this point onwards, we will interact with the API by using OpenMetadata methods. An interesting validation we can already make at this point is verifying that the service is reachable and healthy. To do so, we can validate the Bool output from:

metadata.health_check()  # `True` means we are alright :)

2. Create the DatabaseService

Following the hierarchy, we need to start by defining a DatabaseService. This will be system hosting our Database, which will contain the Table. Recall how we have mainly two types of models:

Entity definitions, such as Table, MlModel or Topic.
API definitions, useful when running a PUT, POST or PATCH request: CreateTable, CreateMlModel or CreateTopic.

As we are just creating Entities right now, we’ll stick to the pydantic models with the API definitions. Let’s imagine that we are defining a MySQL:

from metadata.generated.schema.api.services.createDatabaseService import (
    CreateDatabaseServiceRequest,
)
from metadata.generated.schema.entity.services.databaseService import (
    DatabaseService,
    DatabaseServiceType,
    DatabaseConnection,
)
from metadata.generated.schema.entity.services.connections.database.common.basicAuth import (
    BasicAuth,
)
from metadata.generated.schema.entity.services.connections.database.mysqlConnection import (
    MysqlConnection,
)

create_service = CreateDatabaseServiceRequest(
    name="test-service-table",
    serviceType=DatabaseServiceType.Mysql,
    connection=DatabaseConnection(
        config=MysqlConnection(
            username="username",
            authType=BasicAuth(password="password"),
            hostPort="http://localhost:1234",
        )
    ),
)

Note how we can use both String definitions for the attributes, as well as specific types when possible, such as serviceType=DatabaseServiceType.Mysql. The less information we need to hardcode, the better. Another important point here is that the connection definitions are centralized as JSON Schemas. Here you can find the root of all of them. We can review the information that will be passed to the API by visiting the JSON definition of the class we just instantiated. As all these models are powered by pydantic, this conversion is transparent to us:

create_service.json()
# '{"name": "test-service-table", "description": null, "serviceType": "Mysql", "connection": {"config": {"type": "Mysql", "scheme": "mysql+pymysql", "username": "username", "password": "**********", "hostPort": "http://localhost:1234", "database": null, "connectionOptions": null, "connectionArguments": null, "supportsMetadataExtraction": null, "supportsProfiler": null}}, "owner": null}'

Executing the actual creation is easy! As our create_service variable already holds the proper datatype, there is a single line to execute:

service_entity = metadata.create_or_update(data=create_service)

Important: The create_or_update() method uses PUT operations with server-side business rules that may prevent certain field overwrites for data integrity reasons. This applies to various entities where certain fields are protected for business or data integrity purposes.If you need to override existing metadata fields, consider using patch methods instead:

For descriptions: Use patch_description(force=True)
For general metadata: Use patch(override_metadata=True)

Moreover, running a create_or_update will return us the Entity type, so we can explore its attributes easily:

type(service_entity)
# metadata.generated.schema.entity.services.databaseService.DatabaseService

service_entity.json()
# '{"id": "a823952a-1fc1-46d4-bd0e-27f9812871f4", "name": "test-service-table", "displayName": null, "serviceType": "Mysql", "description": null, "connection": {"config": {"type": "Mysql", "scheme": "mysql+pymysql", "username": "username", "password": "**********", "hostPort": "http://localhost:1234", "database": null, "connectionOptions": null, "connectionArguments": null, "supportsMetadataExtraction": null, "supportsProfiler": null}}, "pipelines": null, "version": 0.1, "updatedAt": 1651237632058, "updatedBy": "anonymous", "owner": null, "href": "http://localhost:8585/api/v1/services/databaseServices/a823952a-1fc1-46d4-bd0e-27f9812871f4", "changeDescription": null, "deleted": false}'

3. Create the Database

We can now repeat the process to create a Database Entity. However, if we review the definition of the CreateDatabaseEntityRequest model…

class CreateDatabaseRequest(BaseModel):
    class Config:
        extra = Extra.forbid

    name: basic.EntityName = Field(
        ..., description='Name that identifies this database instance uniquely.'
    )
    description: Optional[str] = Field(
        None,
        description='Description of the database instance. What it has and how to use it.',
    )
    owners: Optional[entityReferenceList.EntityReferenceList] = Field(
        None, description='Owners of this database'
    )
    service: basic.FullyQualifiedEntityName = Field(
        ...,
        description='Link to the database service fully qualified name where this database is hosted in',
    )
    default: Optional[bool] = Field(
        False,
        description="Some databases don't support a database/catalog in the hierarchy and use default database. For example, `MySql`. For such databases, set this flag to true to indicate that this is a default database.",
    )

Note how the only non-optional fields are name and service. The type of service, however, is FullyQualifiedEntityName. This is expected, as there we need to pass the information of an existing Entity. In our case, the fullyQualifiedName of the DatabaseService we just created. The owners field takes a list of EntityReference. To instantiate an EntityReference, pass an id: uuid.UUID and type: str. Specify the id and type of a User. Querying by name The id we actually saw it by printing the service_entity JSON. However, let’s imagine that it did not happen, and the only information we have from the DatabaseService is its name. To retrieve the id, we should then ask the metadata to find our Entity by its FQN:

service_query = metadata.get_by_name(entity=DatabaseService, fqn="test-service-table")

We have just used the get_by_name method. This method is the same that we will use for any Entity. This is why as an argument, we need to provide the entity field. Again, instead of relying on error-prone handwritten parameters, we can just pass the pydantic model we expect to get back. In our case, a DatabaseService.

from metadata.generated.schema.api.data.createDatabase import (
    CreateDatabaseRequest,
)

create_db = CreateDatabaseRequest(
    name="test-db",
    service=service_entity.fullyQualifiedName,
)

db_entity = metadata.create_or_update(create_db)

4. Create the Schema

With the addition of the Schema Entity in 0.10, we now also need to create a Schema, which will be the one containing the Tables. As this entity is a link between other entities, an Entity Reference will be required too.

from metadata.generated.schema.api.data.createDatabaseSchema import (
    CreateDatabaseSchemaRequest,
)

create_schema = CreateDatabaseSchemaRequest(
    name="test-schema",
    database=db_entity.fullyQualifiedName
)

schema_entity = metadata.create_or_update(data=create_schema)

5. Create the Table

Now that we have all the preparations ready, we can just reuse the same steps to create the Table:

from metadata.generated.schema.api.data.createTable import CreateTableRequest
from metadata.generated.schema.entity.data.table import (
    Column,
    DataType,
    Table,
)

create_table = CreateTableRequest(
    name="test",
    databaseSchema=schema_entity.fullyQualifiedName,
    columns=[Column(name="id", dataType=DataType.BIGINT)],
)

table_entity = metadata.create_or_update(create_table)

6. Update the Table

Let’s now update the Table by adding an owner. This will require us to create a User, and then update the Table with it. Afterwards, we will validate that the information has been properly stored. First, make sure that no owner has been set during the creation:

print(table_entity.owners)
# root=[]

Now, create a User:

from metadata.generated.schema.api.teams.createUser import CreateUserRequest

user = metadata.create_or_update(
    data=CreateUserRequest(name="random-user", email="random@user.com"),
)

Update your create_table instance to add the owners field. Use the Create class, because this runs a PUT. Then update the Entity:

from metadata.generated.schema.type.entityReference import EntityReference

create_table.owners = [EntityReference(id=user.id, type="user")]
updated_table_entity = metadata.create_or_update(create_table)

print(updated_table_entity.owners.root[0].name)
# random-user

If you didn’t save the updated_table_entity variable, query it to review the owners field. Run get_by_name with the FQN of the table:

my_table = metadata.get_by_name(entity=Table, fqn="test-service-table.test-db.test-schema.test")

When querying an Entity we might not find it! The Entity could not exist, or there might be an error in the id or fullyQualifiedName.In those cases, the get method won’t fail, but instead will return None. Note that the signature of the get methods is Optional[T], so make sure to validate that there is data coming back!

7. Cleanup

Finally, we can clean up the Table by running the delete method:

metadata.delete(entity=Table, entity_id=my_table.id)

We could directly clean up the service itself with a Hard and Recursive delete. Note that this is ok for this test, but beware when working with production data!

service_id = metadata.get_by_name(
    entity=DatabaseService, fqn="test-service-table"
).id

metadata.delete(
    entity=DatabaseService,
    entity_id=service_id,
    recursive=True,
    hard_delete=True,
)

API Reference

Data Assets

AI Assets

AI SDK

SDK

Discovery

Lineage

Data Contracts

Teams And Users

Data Governance

Data Quality

Alerts

Incidents

Main Concepts

Python SDK | OpenMetadata Python SDK Documentation

Python SDK

Overview

Which Python SDK Should I Use?

Using Both Clients in One Script

Migrating an Existing Job

SDK 2.0 Quick Start

Introducing the Python API (Original Client)

Using Generated Sources

One syntax to rule them all

Walkthrough

1. Initialize OpenMetadata

2. Create the DatabaseService

3. Create the Database

4. Create the Schema

5. Create the Table

6. Update the Table

7. Cleanup

​Python SDK

​Overview

​Which Python SDK Should I Use?

​Using Both Clients in One Script

​Migrating an Existing Job

​SDK 2.0 Quick Start

​Introducing the Python API (Original Client)

​Using Generated Sources

​One syntax to rule them all

​Walkthrough

​1. Initialize OpenMetadata

​2. Create the DatabaseService

​3. Create the Database

​4. Create the Schema

​5. Create the Table

​6. Update the Table

​7. Cleanup

Python SDK

Overview

Which Python SDK Should I Use?

Using Both Clients in One Script

Migrating an Existing Job

SDK 2.0 Quick Start

Introducing the Python API (Original Client)

Using Generated Sources

One syntax to rule them all

Walkthrough

1. Initialize OpenMetadata

2. Create the DatabaseService

3. Create the Database

4. Create the Schema

5. Create the Table

6. Update the Table

7. Cleanup