Build a Custom Application

Applications are powerful, reusable groups of derived Components bundled into a single visual unit within your Flow Graph. Custom Applications give you full programmatic control over Component generation, allowing you to implement complex logic that goes beyond what template-based Applications can provide.

When to use Custom Applications

Choose Custom Applications when you need:

• Complex conditional logic for Component generation
• Dynamic number of Components based on configuration
• Custom validation or processing of configuration
• Integration with external systems during build time
• Full control over Component structure and relationships

For simpler use cases with template-based generation, see Simple Applications.

Create a Custom Application

1. Define the configuration model

Use Pydantic to define a typed configuration schema:

src/applications/quality_app.py

from pydantic import BaseModel, Field
from typing import Optional

class QualityConfig(BaseModel):
    """Configuration for the data quality Application."""
    input_table: str = Field(..., description="Source table to validate")
    threshold: int = Field(default=50, description="Quality score threshold")
    enable_detailed_checks: bool = Field(default=False, description="Run additional checks")
    output_format: Optional[str] = Field(default="parquet", description="Output format")

2. Implement the Application class

src/applications/quality_app.py

import yaml
from ascend.application.application import Application, ApplicationBuildContext, application
from ascend.models.component.component import Component
from ascend.resources import ComponentBuilder

@application(name="data_quality")
class DataQualityApp(Application[QualityConfig]):
    config_model = QualityConfig

    def components(self, config: QualityConfig, context: ApplicationBuildContext) -> list[Component | ComponentBuilder]:
        components = []

        # Create validation Component
        validation_name = context.fully_qualified_component_name("validate")
        validation_yaml = f"""
        component:
          name: {validation_name}
          transform:
            sql: |
              SELECT *
              FROM {{{{ ref('{config.input_table}') }}}}
              WHERE quality_score > {config.threshold}
        """
        components.append(Component(**yaml.safe_load(validation_yaml)))

        # Conditionally add detailed checks
        if config.enable_detailed_checks:
            detail_name = context.fully_qualified_component_name("detailed_checks")
            detail_yaml = f"""
            component:
              name: {detail_name}
              transform:
                sql: |
                  SELECT *
                  FROM {{{{ ref('{validation_name}') }}}}
                  WHERE quality_score < 100
            """
            components.append(Component(**yaml.safe_load(detail_yaml)))

        return components

3. Configure the Application

Create a YAML file that references your Custom Application:

flows/my_flow/quality_check.yaml

component:
  application:
    application_id: data_quality  # Matches @application(name="...")
    config:
      input_table: raw_events
      threshold: 75
      enable_detailed_checks: true
      output_format: json

ApplicationBuildContext reference

The context parameter provides access to build-time information:

Property	Description
application_component_name	Name of the parent Application Component
flow_name	Name of the containing Flow
flow_build_context	Full FlowBuildContext instance
flow_options	FlowOptions with parameters and configuration
fully_qualified_component_name(name)	Creates fully qualified sub-component name

Access Flow parameters

Use context.flow_options.parameters to access Flow and Profile parameters at build time:

def components(self, config: MyConfig, context: ApplicationBuildContext):
    # Access flow parameters
    flow_params = context.flow_options.parameters

    # Conditionally generate components based on parameters
    if flow_params.get("enable_advanced_features", False):
        # Generate additional components
        pass

Create fully qualified names

Sub-components use the parent__child naming pattern. Use fully_qualified_component_name() to generate correct names:

def components(self, config: MyConfig, context: ApplicationBuildContext):
    # If Application is named "my_app", this creates "my_app__transform"
    transform_name = context.fully_qualified_component_name("transform")

    # Use in component definition
    yaml_def = f"""
    component:
      name: {transform_name}
      transform:
        sql: SELECT * FROM ...
    """

Reference Components

Reference sibling sub-components

Within an Application, reference other sub-components using fully qualified names:

def components(self, config: MyConfig, context: ApplicationBuildContext):
    read_name = context.fully_qualified_component_name("read")
    transform_name = context.fully_qualified_component_name("transform")

    # First component: read
    read_yaml = f"""
    component:
      name: {read_name}
      read:
        connection: warehouse
        table: {config.source_table}
    """

    # Second component: references the read component
    transform_yaml = f"""
    component:
      name: {transform_name}
      transform:
        sql: |
          SELECT * FROM {{{{ ref('{read_name}') }}}}
          WHERE value > {config.threshold}
    """

    return [
        Component(**yaml.safe_load(read_yaml)),
        Component(**yaml.safe_load(transform_yaml)),
    ]

Reference external Components

Reference Components outside the Application using standard ref():

transform_yaml = f"""
component:
  name: {transform_name}
  transform:
    sql: |
      SELECT * FROM {{{{ ref('external_component') }}}}
      -- Or from another Flow
      UNION ALL
      SELECT * FROM {{{{ ref('other_component', flow='other_flow') }}}}
"""

Reference Application sub-components from outside

From other Components in your Flow, reference Application sub-components using fully qualified names:

-- Reference a sub-component of my_app Application
SELECT * FROM {{ ref('my_app__transform') }}

Complete example

Here's a full Custom Application that creates a parameterized ETL pipeline:

src/applications/etl_pipeline.py

import yaml
from pydantic import BaseModel, Field
from typing import Optional
from ascend.application.application import Application, ApplicationBuildContext, application
from ascend.models.component.component import Component

class ETLConfig(BaseModel):
    source_table: str = Field(..., description="Source table name")
    destination_schema: str = Field(..., description="Target schema")
    filter_column: Optional[str] = Field(default=None, description="Column to filter on")
    filter_value: Optional[int] = Field(default=None, description="Filter threshold")
    include_aggregations: bool = Field(default=True, description="Include summary aggregations")

@application(name="etl_pipeline")
class ETLPipelineApp(Application[ETLConfig]):
    config_model = ETLConfig

    def components(self, config: ETLConfig, context: ApplicationBuildContext):
        components = []
        flow_name = context.flow_name

        # 1. Extract component
        extract_name = context.fully_qualified_component_name("extract")
        extract_sql = f"SELECT * FROM {{{{ ref('{config.source_table}') }}}}"

        if config.filter_column and config.filter_value:
            extract_sql += f" WHERE {config.filter_column} > {config.filter_value}"

        extract_yaml = f"""
        component:
          name: {extract_name}
          transform:
            sql: |
              {extract_sql}
        """
        components.append(Component(**yaml.safe_load(extract_yaml)))

        # 2. Transform component
        transform_name = context.fully_qualified_component_name("transform")
        transform_yaml = f"""
        component:
          name: {transform_name}
          transform:
            sql: |
              SELECT
                *,
                CURRENT_TIMESTAMP() as processed_at
              FROM {{{{ ref('{extract_name}') }}}}
        """
        components.append(Component(**yaml.safe_load(transform_yaml)))

        # 3. Optional aggregations
        if config.include_aggregations:
            agg_name = context.fully_qualified_component_name("aggregations")
            agg_yaml = f"""
            component:
              name: {agg_name}
              transform:
                sql: |
                  SELECT
                    COUNT(*) as total_records,
                    MIN(processed_at) as first_processed,
                    MAX(processed_at) as last_processed
                  FROM {{{{ ref('{transform_name}') }}}}
            """
            components.append(Component(**yaml.safe_load(agg_yaml)))

        return components

Usage:

flows/sales/sales_etl.yaml

component:
  application:
    application_id: etl_pipeline
    config:
      source_table: raw_sales
      destination_schema: analytics
      filter_column: amount
      filter_value: 0
      include_aggregations: true

Best practices

1. Use Pydantic Field descriptions: Document parameters with Field(description="...") for better discoverability
2. Validate configuration: Add Pydantic validators for complex validation logic
3. Single purpose: Each Application should have one focused goal
4. Type hints: Use proper type hints for all configuration fields
5. Escape Jinja: Use double braces {{{{ }}}} in f-strings for Jinja expressions
6. Test locally: Validate your Application logic before deploying

Next steps

• Learn about Simple Applications for template-based generation
• Review the Applications concept guide