Skip to main content
Version: Next

Form

The form entity is a core entity in DataHub's metadata model that enables structured metadata collection and compliance initiatives. Forms provide a centralized, template-based approach for capturing essential metadata across data assets through a collaborative, crowdsourced workflow.

Identity​

Forms are identified by a unique string identifier that is chosen by the creator.

A form URN is structured as: urn:li:form:<form-id>

For example: urn:li:form:metadataInitiative2024

The form identifier should be meaningful and descriptive, typically representing the initiative or purpose of the form (e.g., piiClassification2024, dataQualityCompliance, retentionPolicyForm).

Important Capabilities​

Form Metadata and Configuration​

Forms are defined using the formInfo aspect, which contains all the core information about a form:

  • Name and Description: Display name and detailed description of the form's purpose
  • Type: Two types are supported:
    • COMPLETION: Forms designed purely for collecting metadata fields on entities
    • VERIFICATION: Forms used to verify that entities comply with a policy via presence of specific metadata fields
  • Prompts: A list of prompts (questions) that users need to respond to. Each prompt represents a requirement to fill out specific metadata. Prompts currently support:
    • STRUCTURED_PROPERTY: Prompts to apply structured properties to entities
    • FIELDS_STRUCTURED_PROPERTY: Prompts to apply structured properties to schema fields (columns) of datasets
  • Actor Assignment: Defines who should complete the form. Forms can be assigned to:
    • Asset owners (default)
    • Specific users
    • Specific groups

Here's an example of form metadata:

Python SDK: Create a form
import logging
import os

from datahub.emitter.mcp import MetadataChangeProposalWrapper
from datahub.emitter.rest_emitter import DatahubRestEmitter

# Imports for metadata model classes
from datahub.metadata.schema_classes import (
    FormActorAssignmentClass,
    FormInfoClass,
    FormPromptClass,
    FormPromptTypeClass,
    FormTypeClass,
    StructuredPropertyParamsClass,
)
from datahub.metadata.urns import FormUrn

log = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)

# Use the structured property created by structured_property_create_basic.py
retention_property_urn = "urn:li:structuredProperty:io.acryl.privacy.retentionTime"

# define the prompts for our form
prompt_1 = FormPromptClass(
    id="1",  # ensure IDs are globally unique
    title="Data Retention Policy",
    type=FormPromptTypeClass.STRUCTURED_PROPERTY,  # structured property type prompt
    structuredPropertyParams=StructuredPropertyParamsClass(urn=retention_property_urn),
    required=True,
)
prompt_2 = FormPromptClass(
    id="2",  # ensure IDs are globally unique
    title="Field-Level Retention",
    type=FormPromptTypeClass.FIELDS_STRUCTURED_PROPERTY,  # structured property prompt on dataset schema fields
    structuredPropertyParams=StructuredPropertyParamsClass(urn=retention_property_urn),
    required=False,  # dataset schema fields prompts should not be required
)

form_urn = FormUrn("metadata_initiative_1")
form_info_aspect = FormInfoClass(
    name="Metadata Initiative 2024",
    description="Please respond to this form for metadata compliance purposes",
    type=FormTypeClass.VERIFICATION,
    actors=FormActorAssignmentClass(owners=True),
    prompts=[prompt_1, prompt_2],
)

event: MetadataChangeProposalWrapper = MetadataChangeProposalWrapper(
    entityUrn=str(form_urn),
    aspect=form_info_aspect,
)

# Create rest emitter
rest_emitter = DatahubRestEmitter(
    gms_server=os.getenv("DATAHUB_GMS_URL", "http://localhost:8080"),
    token=os.getenv("DATAHUB_GMS_TOKEN"),
)
rest_emitter.emit_mcp(event)
print(f"Created form: {form_urn}")

Dynamic Form Assignment​

Forms can be assigned to entities in two ways:

  1. Explicit Assignment: Directly assign a form to specific entity URNs
  2. Dynamic Assignment: Use the dynamicFormAssignment aspect to automatically assign forms to entities matching certain criteria

The dynamicFormAssignment aspect enables rule-based form assignment by specifying filters. Entities matching the filter criteria will automatically have the form applied. The filter supports:

  • Entity type (e.g., datasets, dashboards, charts)
  • Platform (e.g., snowflake, bigquery, postgres)
  • Domain membership
  • Container membership

This dynamic approach is particularly powerful for large-scale compliance initiatives where you want to apply forms to all assets of a certain type or within a specific domain.

Python SDK: Assign forms dynamically with filters
import logging
import os

from datahub.emitter.mcp import MetadataChangeProposalWrapper
from datahub.emitter.rest_emitter import DatahubRestEmitter
from datahub.metadata.schema_classes import (
    CriterionClass,
    DynamicFormAssignmentClass,
    FilterClass,
    FormActorAssignmentClass,
    FormInfoClass,
    FormPromptClass,
    FormPromptTypeClass,
    FormTypeClass,
    StructuredPropertyParamsClass,
)
from datahub.metadata.urns import FormUrn

log = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)

# Create a form that will be dynamically assigned to all Snowflake datasets
# in the "Finance" domain

# Define the form metadata
form_urn = FormUrn("snowflake_finance_compliance")
form_info = FormInfoClass(
    name="Snowflake Finance Data Compliance",
    description="Compliance form for all Snowflake datasets in the Finance domain",
    type=FormTypeClass.VERIFICATION,
    actors=FormActorAssignmentClass(owners=True),
    prompts=[
        FormPromptClass(
            id="retention_time_prompt",
            title="Data Retention Period",
            description="Specify how long this data should be retained",
            type=FormPromptTypeClass.STRUCTURED_PROPERTY,
            structuredPropertyParams=StructuredPropertyParamsClass(
                urn="urn:li:structuredProperty:io.acryl.dataRetentionTime"
            ),
            required=True,
        ),
        FormPromptClass(
            id="pii_classification_prompt",
            title="PII Classification",
            description="Classify whether this dataset contains PII",
            type=FormPromptTypeClass.STRUCTURED_PROPERTY,
            structuredPropertyParams=StructuredPropertyParamsClass(
                urn="urn:li:structuredProperty:io.acryl.piiClassification"
            ),
            required=True,
        ),
    ],
)

# Define dynamic assignment filter
# This form will be assigned to all entities matching these criteria
dynamic_assignment = DynamicFormAssignmentClass(
    filter=FilterClass(
        criteria=[
            CriterionClass(
                field="platform",
                value="urn:li:dataPlatform:snowflake",
                condition="EQUAL",
            ),
            CriterionClass(
                field="domains",
                value="urn:li:domain:finance",
                condition="EQUAL",
            ),
            CriterionClass(
                field="_entityType",
                value="urn:li:entityType:dataset",
                condition="EQUAL",
            ),
        ]
    )
)

# Create rest emitter
rest_emitter = DatahubRestEmitter(
    gms_server=os.getenv("DATAHUB_GMS_URL", "http://localhost:8080"),
    token=os.getenv("DATAHUB_GMS_TOKEN"),
)

# Emit the form info
form_info_event = MetadataChangeProposalWrapper(
    entityUrn=str(form_urn),
    aspect=form_info,
)
rest_emitter.emit(form_info_event)

# Emit the dynamic assignment
dynamic_assignment_event = MetadataChangeProposalWrapper(
    entityUrn=str(form_urn),
    aspect=dynamic_assignment,
)
rest_emitter.emit(dynamic_assignment_event)

log.info(
    f"Created form {form_urn} with dynamic assignment to Snowflake datasets in Finance domain"
)

Form Assignment on Entities​

When forms are assigned to entities (either explicitly or dynamically), the assignments are tracked via the forms aspect on the target entity. This aspect maintains:

  • Incomplete Forms: Forms that have outstanding prompts to be completed
  • Complete Forms: Forms where all required prompts have been responded to
  • Verifications: For verification-type forms, tracks which forms have been successfully verified
  • Prompt Status: For each form, tracks which prompts are complete vs. incomplete, along with timestamps

This allows DataHub to provide progress tracking, notifications, and analytics about form completion across your data catalog.

Ownership​

Forms can have owners assigned through the standard ownership aspect. Owners of forms are typically the governance or compliance team members who are responsible for managing and maintaining the form.

Python SDK: Add owners to a form
import logging

from datahub.emitter.mcp import MetadataChangeProposalWrapper
from datahub.emitter.rest_emitter import DatahubRestEmitter
from datahub.metadata.schema_classes import (
    AuditStampClass,
    OwnerClass,
    OwnershipClass,
    OwnershipTypeClass,
)
from datahub.metadata.urns import CorpUserUrn, FormUrn

log = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)

# Form URN to add owner to
form_urn = FormUrn("metadata_initiative_2024")

# Create ownership aspect
ownership = OwnershipClass(
    owners=[
        OwnerClass(
            owner=str(CorpUserUrn("governance_team")),
            type=OwnershipTypeClass.TECHNICAL_OWNER,
        )
    ],
    lastModified=AuditStampClass(
        time=0, actor="urn:li:corpuser:datahub", impersonator=None
    ),
)

# Create and emit metadata change proposal
event = MetadataChangeProposalWrapper(
    entityUrn=str(form_urn),
    aspect=ownership,
)

rest_emitter = DatahubRestEmitter(gms_server="http://localhost:8080")
rest_emitter.emit(event)

log.info(f"Added owner to form {form_urn}")

Code Examples​

Creating and Managing Forms​

Python SDK: Create a verification form with multiple prompts
import logging
import os

from datahub.emitter.mcp import MetadataChangeProposalWrapper
from datahub.emitter.rest_emitter import DatahubRestEmitter

# Imports for metadata model classes
from datahub.metadata.schema_classes import (
    FormActorAssignmentClass,
    FormInfoClass,
    FormPromptClass,
    FormPromptTypeClass,
    FormTypeClass,
    StructuredPropertyParamsClass,
)
from datahub.metadata.urns import FormUrn

log = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)

# Use the structured property created by structured_property_create_basic.py
retention_property_urn = "urn:li:structuredProperty:io.acryl.privacy.retentionTime"

# define the prompts for our form
prompt_1 = FormPromptClass(
    id="1",  # ensure IDs are globally unique
    title="Data Retention Policy",
    type=FormPromptTypeClass.STRUCTURED_PROPERTY,  # structured property type prompt
    structuredPropertyParams=StructuredPropertyParamsClass(urn=retention_property_urn),
    required=True,
)
prompt_2 = FormPromptClass(
    id="2",  # ensure IDs are globally unique
    title="Field-Level Retention",
    type=FormPromptTypeClass.FIELDS_STRUCTURED_PROPERTY,  # structured property prompt on dataset schema fields
    structuredPropertyParams=StructuredPropertyParamsClass(urn=retention_property_urn),
    required=False,  # dataset schema fields prompts should not be required
)

form_urn = FormUrn("metadata_initiative_1")
form_info_aspect = FormInfoClass(
    name="Metadata Initiative 2024",
    description="Please respond to this form for metadata compliance purposes",
    type=FormTypeClass.VERIFICATION,
    actors=FormActorAssignmentClass(owners=True),
    prompts=[prompt_1, prompt_2],
)

event: MetadataChangeProposalWrapper = MetadataChangeProposalWrapper(
    entityUrn=str(form_urn),
    aspect=form_info_aspect,
)

# Create rest emitter
rest_emitter = DatahubRestEmitter(
    gms_server=os.getenv("DATAHUB_GMS_URL", "http://localhost:8080"),
    token=os.getenv("DATAHUB_GMS_TOKEN"),
)
rest_emitter.emit_mcp(event)
print(f"Created form: {form_urn}")

Python SDK: Create a form with dynamic assignment
import logging
import os

from datahub.emitter.mcp import MetadataChangeProposalWrapper
from datahub.emitter.rest_emitter import DatahubRestEmitter
from datahub.metadata.schema_classes import (
    CriterionClass,
    DynamicFormAssignmentClass,
    FilterClass,
    FormActorAssignmentClass,
    FormInfoClass,
    FormPromptClass,
    FormPromptTypeClass,
    FormTypeClass,
    StructuredPropertyParamsClass,
)
from datahub.metadata.urns import FormUrn

log = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)

# Create a form that will be dynamically assigned to all Snowflake datasets
# in the "Finance" domain

# Define the form metadata
form_urn = FormUrn("snowflake_finance_compliance")
form_info = FormInfoClass(
    name="Snowflake Finance Data Compliance",
    description="Compliance form for all Snowflake datasets in the Finance domain",
    type=FormTypeClass.VERIFICATION,
    actors=FormActorAssignmentClass(owners=True),
    prompts=[
        FormPromptClass(
            id="retention_time_prompt",
            title="Data Retention Period",
            description="Specify how long this data should be retained",
            type=FormPromptTypeClass.STRUCTURED_PROPERTY,
            structuredPropertyParams=StructuredPropertyParamsClass(
                urn="urn:li:structuredProperty:io.acryl.dataRetentionTime"
            ),
            required=True,
        ),
        FormPromptClass(
            id="pii_classification_prompt",
            title="PII Classification",
            description="Classify whether this dataset contains PII",
            type=FormPromptTypeClass.STRUCTURED_PROPERTY,
            structuredPropertyParams=StructuredPropertyParamsClass(
                urn="urn:li:structuredProperty:io.acryl.piiClassification"
            ),
            required=True,
        ),
    ],
)

# Define dynamic assignment filter
# This form will be assigned to all entities matching these criteria
dynamic_assignment = DynamicFormAssignmentClass(
    filter=FilterClass(
        criteria=[
            CriterionClass(
                field="platform",
                value="urn:li:dataPlatform:snowflake",
                condition="EQUAL",
            ),
            CriterionClass(
                field="domains",
                value="urn:li:domain:finance",
                condition="EQUAL",
            ),
            CriterionClass(
                field="_entityType",
                value="urn:li:entityType:dataset",
                condition="EQUAL",
            ),
        ]
    )
)

# Create rest emitter
rest_emitter = DatahubRestEmitter(
    gms_server=os.getenv("DATAHUB_GMS_URL", "http://localhost:8080"),
    token=os.getenv("DATAHUB_GMS_TOKEN"),
)

# Emit the form info
form_info_event = MetadataChangeProposalWrapper(
    entityUrn=str(form_urn),
    aspect=form_info,
)
rest_emitter.emit(form_info_event)

# Emit the dynamic assignment
dynamic_assignment_event = MetadataChangeProposalWrapper(
    entityUrn=str(form_urn),
    aspect=dynamic_assignment,
)
rest_emitter.emit(dynamic_assignment_event)

log.info(
    f"Created form {form_urn} with dynamic assignment to Snowflake datasets in Finance domain"
)

Python SDK: Assign a form to specific entities
import logging

from datahub.emitter.mcp import MetadataChangeProposalWrapper
from datahub.emitter.rest_emitter import DatahubRestEmitter
from datahub.metadata.schema_classes import FormAssociationClass, FormsClass
from datahub.metadata.urns import DatasetUrn, FormUrn

log = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)

# Form to assign
form_urn = FormUrn("metadata_initiative_2024")

# Entities to assign the form to
dataset_urns = [
    DatasetUrn(platform="snowflake", name="prod.analytics.customer_data", env="PROD"),
    DatasetUrn(platform="snowflake", name="prod.analytics.sales_data", env="PROD"),
]

# Create rest emitter
rest_emitter = DatahubRestEmitter(gms_server="http://localhost:8080")

# Assign the form to each entity by updating the forms aspect
for dataset_urn in dataset_urns:
    # Create a forms aspect with the form marked as incomplete
    forms_aspect = FormsClass(
        incompleteForms=[FormAssociationClass(urn=str(form_urn))],
        completedForms=[],
        verifications=[],
    )

    # Emit the forms aspect for the entity
    event = MetadataChangeProposalWrapper(
        entityUrn=str(dataset_urn),
        aspect=forms_aspect,
    )
    rest_emitter.emit(event)
    log.info(f"Assigned form {form_urn} to entity {dataset_urn}")

log.info(f"Successfully assigned form to {len(dataset_urns)} entities")

Python SDK: Remove a form from entities
import logging

from datahub.emitter.mcp import MetadataChangeProposalWrapper
from datahub.emitter.rest_emitter import DatahubRestEmitter
from datahub.metadata.schema_classes import FormsClass
from datahub.metadata.urns import DatasetUrn, FormUrn

log = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO)

# Form to remove
form_urn = FormUrn("metadata_initiative_2024")

# Entities to remove the form from
dataset_urns = [
    DatasetUrn(platform="snowflake", name="prod.analytics.customer_data", env="PROD"),
    DatasetUrn(platform="snowflake", name="prod.analytics.sales_data", env="PROD"),
]

# Create rest emitter
rest_emitter = DatahubRestEmitter(gms_server="http://localhost:8080")

# Remove the form from each entity by setting empty forms aspect
for dataset_urn in dataset_urns:
    # Create an empty forms aspect
    # This will remove all form assignments from the entity
    forms_aspect = FormsClass(
        incompleteForms=[],
        completedForms=[],
        verifications=[],
    )

    # Emit the forms aspect for the entity
    event = MetadataChangeProposalWrapper(
        entityUrn=str(dataset_urn),
        aspect=forms_aspect,
    )
    rest_emitter.emit(event)
    log.info(f"Removed forms from entity {dataset_urn}")

log.info(f"Successfully removed forms from {len(dataset_urns)} entities")

Querying Forms​

Forms can be queried via the REST API to retrieve form definitions and check their assignment status.

Query a form via REST API
curl 'http://localhost:8080/entities/urn%3Ali%3Aform%3AmetadataInitiative2024'

This will return the form entity with all its aspects including formInfo, dynamicFormAssignment (if configured), and ownership.

Check form assignments on an entity
curl 'http://localhost:8080/entities/urn%3Ali%3Adataset%3A(urn%3Ali%3AdataPlatform%3Asnowflake,mydb.schema.table,PROD)?aspects=forms'

This returns the forms aspect showing all incomplete and completed forms for the dataset.

Integration Points​

Relationship with Other Entities​

Forms have several key integration points in the DataHub ecosystem:

  1. Structured Properties: Forms primarily interact with structured properties through prompts. Each prompt can require users to fill out specific structured properties on entities or schema fields.

  2. Assets (Datasets, Dashboards, Charts, etc.): Forms are assigned to data assets to collect metadata. The relationship is tracked through:

    • The forms aspect on the target entity
    • The dynamicFormAssignment filter criteria on the form

  3. Users and Groups: Forms are assigned to specific actors (users or groups) who are responsible for completing them. This creates a workflow where:

    • Forms appear in assignees' task lists
    • Assignees receive notifications about pending forms
    • Completion tracking is maintained per user

  4. Domains and Containers: Forms can be automatically assigned based on domain or container membership, enabling governance teams to apply compliance requirements at the domain or container level.

Typical Usage Patterns​

  1. Compliance Initiatives: Organizations create verification forms to ensure critical assets meet compliance requirements (e.g., PII classification, data retention policies)

  2. Metadata Enrichment: Completion forms help gather missing documentation, ownership, and domain assignments for high-value assets

  3. Governance Workflows: Forms enable systematic metadata collection by routing tasks to domain experts who are best positioned to provide accurate information

  4. Quality Improvement: Forms can be used to incrementally improve metadata quality by focusing on the most important assets first

Notable Exceptions​

Form Prompt Types​

Currently, forms only support structured property prompts. Each prompt requires users to set values for structured properties on entities or their schema fields. Future extensions may include prompts for other metadata types (e.g., documentation requirements, tag requirements).

Schema Field Prompts​

When using FIELDS_STRUCTURED_PROPERTY prompt types, these prompts should not be marked as required, as they apply to an indeterminate number of schema fields. The form is considered complete when users have appropriately responded to the field-level prompts.

Assignment Logic​

Forms are assigned to entities through two independent mechanisms:

  • Direct assignment: Explicitly specified entity URNs
  • Dynamic assignment: Filter-based automatic assignment

An entity can have the same form assigned through both mechanisms. The forms aspect on the entity consolidates all assignments regardless of how they were made.

Form Lifecycle​

Once a form is created and published, assignees will see it in their task queue. Deleting a form requires:

  1. Hard deleting the form entity itself
  2. Removing all references to the form from entities that have it assigned

This two-step process ensures referential integrity across the metadata graph.

Technical Reference​

For technical details about fields, searchability, and relationships, view the Columns tab in DataHub.