QueryBasedTransformer transforms raw metadata DataFrames into structured entities using SQL queries defined in YAML templates. It executes SQL transformations on raw dataframes using the daft engine and automatically creates nested structures from flat dataframes with dot notation. This transformer is ideal when you need SQL-based transformations with flexible template definitions.

QueryBasedTransformer

Class

📁application_sdk.transformers.query

Transforms metadata using SQL queries defined in YAML templates. Executes SQL transformations on raw dataframes using the daft engine and automatically creates nested structures from flat dataframes with dot notation. Uses YAML files to define SQL queries for each asset type, executes these queries on raw dataframes using daft's SQL engine, then automatically groups columns with dot notation into nested structs.

Methods5

▸

__init__

__init__(self, connector_name, tenant_id)

Initialize the transformer with connector name and tenant ID.

Parameters

connector_namestr

Required

Name of the connector

tenant_idstr

Required

ID of the tenant

▸

transform_metadata

transform_metadata(self, typename, dataframe, workflow_id, workflow_run_id, entity_class_definitions=None, **kwargs)

Transforms metadata using SQL queries from YAML templates. Loads YAML template for the typename, prepares default attributes and SQL template, generates SQL query from template, executes SQL on DataFrame using daft engine, groups columns with dot notation into nested structs, and returns transformed DataFrame. Returns None if input DataFrame is empty.

Parameters

typenamestr

Required

Type identifier (e.g., 'DATABASE', 'SCHEMA', 'TABLE', 'COLUMN')

dataframedaft.DataFrame

Required

Raw metadata as daft DataFrame

workflow_idstr

Required

Workflow identifier

workflow_run_idstr

Required

Workflow run identifier

entity_class_definitionsOptional[Dict[str, str]]

Optional

Custom template path mappings

**kwargsdict

Optional

Additional keyword arguments including connection_qualified_name (str) and connection_name (str)

Returns

Optional[daft.DataFrame] - Transformed DataFrame with nested structures, or None if input is empty

▸

generate_sql_query

generate_sql_query(self, yaml_path, dataframe, default_attributes)

Generates a SQL query from a YAML template and DataFrame. Loads YAML template from file, flattens nested columns dictionary, generates SQL column expressions, creates SELECT statement with column expressions, and returns SQL query and literal columns.

Parameters

yaml_pathstr

Required

Path to YAML template file

dataframedaft.DataFrame

Required

DataFrame to reference for column names

default_attributesDict[str, Any]

Required

Default attributes to add to SQL query

Returns

Tuple[str, List[str]] - Tuple of (SQL query string, list of literal columns)

▸

prepare_template_and_attributes

prepare_template_and_attributes(self, dataframe, workflow_id, workflow_run_id, connection_qualified_name=None, connection_name=None, entity_sql_template_path=None)

Prepares the SQL template and default attributes for transformation. Adds default attributes including connection_qualified_name, connection_name, tenant_id, workflow tracking fields, and connector_name.

Parameters

dataframedaft.DataFrame

Required

Input DataFrame

workflow_idstr

Required

Workflow identifier

workflow_run_idstr

Required

Workflow run identifier

connection_qualified_nameOptional[str]

Optional

Connection qualified name

connection_nameOptional[str]

Optional

Connection name

entity_sql_template_pathOptional[str]

Optional

Path to SQL template

Returns

Tuple[daft.DataFrame, str] - Tuple of (DataFrame with default attributes, SQL template string)

▸

get_grouped_dataframe_by_prefix

get_grouped_dataframe_by_prefix(self, dataframe)

Groups columns with dot notation into nested structs. Transforms flat columns like 'attributes.name' into nested structures with attributes.name grouped under attributes struct.

Parameters

dataframedaft.DataFrame

Required

Flat DataFrame with dot notation columns

Returns

daft.DataFrame - DataFrame with columns grouped into nested structs

Usage Examples

Basic transformation

Initialize transformer and transform metadata for tables using default YAML templates

from application_sdk.transformers.query import QueryBasedTransformer

# Initialize transformer
transformer = QueryBasedTransformer(
  connector_name="postgresql-connector",
  tenant_id="tenant-123"
)

# Transform metadata
transformed_df = transformer.transform_metadata(
  typename="TABLE",
  dataframe=raw_table_df,
  workflow_id="extract-tables",
  workflow_run_id="run-001",
  connection_qualified_name="tenant/postgresql/1",
  connection_name="production"
)

Processing multiple entity types

Transform different entity types in sequence

# Transform different entity types
databases_df = transformer.transform_metadata(
  typename="DATABASE",
  dataframe=raw_databases_df,
  workflow_id="workflow-123",
  workflow_run_id="run-456",
  connection_qualified_name=connection_qn,
  connection_name=connection_name
)

tables_df = transformer.transform_metadata(
  typename="TABLE",
  dataframe=raw_tables_df,
  workflow_id="workflow-123",
  workflow_run_id="run-456",
  connection_qualified_name=connection_qn,
  connection_name=connection_name
)

columns_df = transformer.transform_metadata(
  typename="COLUMN",
  dataframe=raw_columns_df,
  workflow_id="workflow-123",
  workflow_run_id="run-456",
  connection_qualified_name=connection_qn,
  connection_name=connection_name
)

Using custom templates

Use custom YAML templates by providing custom template path mappings

from application_sdk.transformers.query import QueryBasedTransformer
from application_sdk.transformers.common.utils import get_yaml_query_template_path_mappings

# Get custom template mappings
custom_mappings = get_yaml_query_template_path_mappings(
  custom_templates_path="./custom_templates",
  assets=["TABLE", "COLUMN"]
)

# Initialize transformer
transformer = QueryBasedTransformer(
  connector_name="my-connector",
  tenant_id="tenant-123"
)

# Use custom templates
transformed_df = transformer.transform_metadata(
  typename="TABLE",
  dataframe=raw_df,
  workflow_id="workflow-123",
  workflow_run_id="run-456",
  entity_class_definitions=custom_mappings,
  connection_qualified_name="tenant/connector/1"
)

Default template mappings

Asset types are automatically mapped to YAML template files when using default templates:

Asset Type	Template File	Location
TABLE	table.yaml	application_sdk/transformers/query/templates/
COLUMN	column.yaml	application_sdk/transformers/query/templates/
DATABASE	database.yaml	application_sdk/transformers/query/templates/
SCHEMA	schema.yaml	application_sdk/transformers/query/templates/
FUNCTION	function.yaml	application_sdk/transformers/query/templates/
EXTRAS-PROCEDURE	extras-procedure.yaml	application_sdk/transformers/query/templates/

Template structure

YAML templates define column mappings using nested dictionaries with dot notation for column paths.

Template format

table: Table
columns:
  attributes:
    name:
      source_query: table_name
    qualifiedName:
      source_query: concat(connection_qualified_name, '/', table_catalog, '/', table_schema, '/', table_name)
      source_columns: [connection_qualified_name, table_catalog, table_schema, table_name]
  customAttributes:
    table_type:
      source_query: table_type
  typeName:
    source_query: |
      CASE
        WHEN table_type = 'VIEW' THEN 'View'
        ELSE 'Table'
      END
    source_columns: [table_type]
  status:
    source_query: "'ACTIVE'"

Column definition fields

Field	Type	Required	Description
source_query	str	Yes	SQL expression or literal value
source_columns	List[str]	No	Source columns to validate before generating SQL query. If missing, column mapping is skipped with warning.

Column expression types

Type	Example	Notes
Direct column reference	source_query: table_name	Direct column mapping
SQL expression	source_query: concat(a, '/', b)	SQL function or expression
String literal	source_query: "'ACTIVE'"	Quoted string value
Boolean/Number literal	source_query: true	Unquoted boolean or number
Conditional expression	source_query: | CASE WHEN ... END	Multi-line SQL CASE statement

Default attributes

The prepare_template_and_attributes method adds these default attributes to all transformations:

Attribute	Source	Type
connection_qualified_name	**kwargs	str
connection_name	**kwargs	str
tenant_id	Initialization parameter	str
workflow_id	Method parameter	str
workflow_run_id	Method parameter	str
connector_name	Initialization parameter	str

Default attributes are available in all SQL expressions within YAML templates.

Nested structure creation

The get_grouped_dataframe_by_prefix method groups flat columns with dot notation into nested structs.

Input format (flat columns):

attributes.name
attributes.qualifiedName
attributes.description
customAttributes.table_type
typeName
status

Output format (nested structs):

{
    "attributes": {
        "name": "...",
        "qualifiedName": "...",
        "description": "..."
    },
    "customAttributes": {
        "table_type": "..."
    },
    "typeName": "...",
    "status": "..."
}

Column name handling: Columns with dots are automatically quoted in SQL (for example, attributes.name becomes "attributes.name").

Error handling

Error Type	Behavior
Template loading error	Raises exception, logs template path
Missing source column	Skips column mapping, logs warning
SQL execution error	Raises exception, logs SQL query and typename
Empty DataFrame	Returns None

Performance characteristics

Aspect	Behavior
Evaluation	Lazy evaluation (daft DataFrames)
Column processing	Only processes columns that exist in DataFrame
Struct building	Uses daft expressions (not row-by-row)
Template loading	On-demand per transformation (consider caching for high-volume scenarios)

See also

• Transformers: Overview of all transformers and the TransformerInterface
• Atlas transformer: Convert metadata into Atlas entities using pyatlan library classes
• Application SDK README: Overview of the Application SDK and its components