Execute Components in order with inputs

Use inputs when your Component needs to process data from upstream Components. Use dependencies (a special kind of input) when your Component only needs upstream Components to complete first without accessing their data, such as for setup tasks or configuration.

To gain a deeper understanding of these concepts, see the Components guide.

tip

Looking to reference Components from other Flows? Use cross-Flow references.

Syntax

You can add inputs or dependencies to any Component type using the appropriate syntax:

Component type	Syntax	Location
YAML Components	inputs: / dependencies: field	Field in the Component configuration
SQL Components	inputs[] / dependencies=[] parameter	Parameter in the config block
Python Components	inputs[] / dependencies=[] parameter	Parameter in the Component decorator

note

Read Components support a dependencies field but not an inputs field, since they don't process data from upstream Components.

YAML example

Define dependencies using the dependencies field in your YAML configuration:

read.yaml

component:
  read:
    connection: read_gcs_lake
    gcs:
      path: ottos-expeditions/lakev0/generated/events/sales_store.parquet/year=
      include:
        - glob: "*/month=*/day=*/*.parquet"
    dependencies:
      - name: sales_website
        flow: extract-load-duckdb
      - name: sales_vendors
        flow: extract-load-duckdb

SQL examples

Specify inputs and dependencies in the configuration block at the top of your SQL file:

transform.sql

{{
    config(
        type="transform",
        inputs=[
            ref('sales_vendors', flow='transform-duckdb'),
            ref('read_sales_vendors', flow='extract-load-duckdb'),
        ]
    )
}}

SELECT * FROM {{ ref('sales_vendors', flow='transform-duckdb') }}

Python examples

Add inputs and dependencies as parameters in your Component decorator. Tables referenced via ref() must also be passed as function arguments. The implementation varies by Component type:

Transform

goats.py

import ibis
from ascend.application.context import ComponentExecutionContext
from ascend.resources import ref, transform


@transform(
    inputs=[
        ref("ascenders", flow="transform-duckdb"),
        ref("routes", flow="transform-duckdb"),
        ref("telemetry", flow="transform-duckdb"),
    ]
)
def goats(
    ascenders: ibis.Table,
    routes: ibis.Table,
    telemetry: ibis.Table,
    context: ComponentExecutionContext,
) -> ibis.Table:
    return ascenders.sample(0.01)

Incremental Python Read Component

read_inlinked.py

import polars as pl
import pyarrow as pa
from ascend.application.context import ComponentExecutionContext
from ascend.resources import read, ref


@read(
    strategy="incremental",
    incremental_strategy="merge",
    unique_key="id",
    on_schema_change="sync_all_columns",
    inputs=[ref("read_sales", flow="extract-load-duckdb")],
)
def read_inlinked(context: ComponentExecutionContext) -> pa.Table:
    df = pl.read_parquet("gs://ascend-io-gcs-public/ottos-expeditions/lakev0/generated/events/inlinked.parquet/year=*/month=*/day=*/*.parquet")
    ...

    return df.to_arrow()

Smart Python Read Component

yellow_cab.py

from datetime import datetime

import pandas as pd
from ascend.application.context import ComponentExecutionContext
from ascend.common.filters import ListItem
from ascend.resources import CustomPythonPartitionedReader, ref

sales_1 = {
    "sale_id": [1001, 1002, 1003],
    "date": [datetime(2024, 1, 15), datetime(2024, 1, 16), datetime(2024, 1, 17)],
    "product": ["Laptop", "Mouse", "Keyboard"],
    "quantity": [2, 5, 3],
    "customer_id": ["CUST001", "CUST002", "CUST003"],
}

sales_2 = {
    "sale_id": [1004, 1005, 1006],
    "date": [datetime(2024, 1, 18), datetime(2024, 1, 19), datetime(2024, 1, 20)],
    "product": ["Monitor", "Webcam", "Headset"],
    "quantity": [1, 2, 1],
    "customer_id": ["CUST004", "CUST005", "CUST001"],
}

reader = CustomPythonPartitionedReader(
    name="read_sales",
    dependencies=[ref("read_sales_vendors", flow="extract-load-duckdb")],
)


@reader.list()
def list_partitions(context):
    yield ListItem(name="sales_1")
    current_data = context.current_data()
    if current_data is not None:
        yield ListItem(name="sales_2")


@reader.read()
def read_partition(context: ComponentExecutionContext, item: ListItem) -> pd.DataFrame:
    if item.name == "sales_1":
        return pd.DataFrame(sales_1)
    else:
        return pd.DataFrame(sales_2)

Tasks

Tasks use the same syntax for dependencies and inputs as Transforms. Python Tasks follow the Python Transform syntax, while SQL Tasks follow the SQL Transform syntax.

Python Task

task_respond_on_socials.py

import ibis
from ascend.application.context import ComponentExecutionContext
from ascend.common.events import log
from ascend.resources import ref, task


@task(
    dependencies=[
        ref("social_media", flow="transform-duckdb"),
    ]
)
def task_respond_on_socials(social_media: ibis.Table, context: ComponentExecutionContext) -> None:
    for i in range(1000):
        log("Thank you for your comment!")

SQL Task

task.sql

{{
  config(
    type="task",
    dependencies=[ref("guides", flow="transform-duckdb")]
  )
}}
SELECT 1 as id