Python SDK for Tags

Python SDK for Tags

In this guide, we will use the Python SDK to create Tags.

A prerequisite for this section is to have previously gone through the following docs.

In the following sections we will:

• Create a Database Service, a Database, a Schema and one Table,
• Create a Classification,
• Create a Tag for the Classification and add it to the Table.

Creating the Entities

To prepare the necessary ingredients, execute the following steps.

1. Preparing the Client

from metadata.sdk import configure

configure(
    host="http://localhost:8585/api",
    jwt_token="<token>",
)

2. Creating the Database Service

We are mocking a MySQL instance. Note how we need to pass the right configuration class MysqlConnection, as a parameter for the generic DatabaseConnection type.

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

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

db_service_entity = DatabaseServices.create(db_service)

3. Creating the Database

Any Entity that is created and linked to another Entity, has to hold the fullyQualifiedName to the Entity it relates to. In this case, a Database is bound to a specific service.

from metadata.sdk import Databases
from metadata.generated.schema.api.data.createDatabase import CreateDatabaseRequest

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

create_db_entity = Databases.create(create_db)

4. Creating the Schema

The same happens with the Schemas. They are related to a Database.

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

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

create_schema_entity = DatabaseSchemas.create(create_schema)

5. Creating the Table

We are doing a simple example with a single column.

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

table = CreateTableRequest(
    name="my_table",
    databaseSchema=create_schema_entity.fullyQualifiedName,
    columns=[Column(name="id", dataType=DataType.BIGINT)],
)

table_entity = Tables.create(table)

6. Creating the Classification

Classifications are the entities we use to group tag definitions. Let’s create a sample one:

from metadata.sdk import Classifications
from metadata.generated.schema.api.classification.createClassification import (
    CreateClassificationRequest,
)

classification_request = CreateClassificationRequest(
    name="TestClassification",
    description="Sample classification.",
)

classification_entity = Classifications.create(classification_request)

7. Creating the Tag

Once there is an existing classification, we create tags within that given classification:

from metadata.sdk import Tags
from metadata.generated.schema.api.classification.createTag import CreateTagRequest

tag_request = CreateTagRequest(
    classification=classification_entity.fullyQualifiedName,
    name="TestTag",
    description="Sample Tag.",
)

tag_entity = Tags.create(tag_request)

8. Tagging a Table

Now that we have the Tag, we can proceed and tag a specific asset. In this example, we’ll tag the Table we created above. The Python SDK exposes a helper method that handles the update:

tag_fqn = str(tag_entity.fullyQualifiedName.root)

table_entity = Tables.add_tag(
    table_id=str(table_entity.id.root),
    tag_fqn=tag_fqn,
)

If we now check the tags for the Table, the new tag will be appearing in there.

If we can control the Tag creation process when the Table metadata is generated, we can also assign the tags in the tags property of CreateTableRequest.

9. Tagging a Column

The same idea can be replicated to tag a column:

from metadata.generated.schema.type.tagLabel import LabelType, State, TagLabel, TagSource

table_with_columns = Tables.retrieve(
    str(table_entity.id.root),
    fields=["columns", "tags"],
)

for column in table_with_columns.columns or []:
    if column.name.root == "id":
        column.tags = [
            TagLabel(
                tagFQN=tag_fqn,
                source=TagSource.Classification,
                labelType=LabelType.Manual,
                state=State.Confirmed,
            )
        ]
        break

table_entity = Tables.update(table_with_columns)