Operators & Integrations Catalog

54Compute & scripts#

What it does: The core scripting layer — BashOperator, PythonOperator, ExternalPythonOperator, PythonVirtualenvOperator, and BranchPythonOperator.

How it works:

• All five run as CPU-bound, local work, subject to the deployment's local-task concurrency cap — heavier fan-out of these operators competes for the same local-execution budget.
• BashOperator runs your command with bash -c "<your command>" from the orchestrator process's own working directory — not your DAGs repo checkout. Use an explicit cd $DAGS_REPO_PATH/... (or absolute paths) if your command needs to reference files checked out from Git.
• ExternalPythonOperator/PythonVirtualenvOperator re-import your DAG module fresh inside the target interpreter — that interpreter/venv must be able to import dag_parser and any packages your callable needs.
• BranchPythonOperator is the only one of the five that changes the scheduling graph (feature 33) — the other four simply run and report success/failure.
• None of the five inherit the orchestrator's own environment (feature 22), regardless of which one you pick.

Example:

# BashOperator: cwd is the orchestrator's own process dir, NOT the dags repo
fix_path_example = BashOperator(
    task_id="fix_path_example",
    bash_command="cd $DAGS_REPO_PATH/dags && ls -la",  # explicit cd required
)

55SQL databases#

What it does: Running a query against a data warehouse or database connection. SnowflakeOperator, SQLExecuteQueryOperator, MySqlOperator, PostgresOperator, and RedshiftOperator all share the same shape — sql= plus connection_id= — pick whichever matches your connection's type.

Parameters:

How it works:

• Use SnowflakeOperator for Snowflake, and SQLExecuteQueryOperator (database-agnostic) or the type-specific MySqlOperator/PostgresOperator/ RedshiftOperator for the matching database.
• parameters values are bound as real query parameters, not string-substituted into the SQL text — passing user-supplied values (e.g. from a manually triggered run's conf) is safe without any manual escaping on your part.
• Only the first result row is captured as return_value (feature 37) — a query returning many rows drops everything after row 1.
• SnowflakeOperator maintains a connection pool per connection_id that persists across task executions on the same worker — if you change a connection's credentials, an already-warm pool won't pick up the change until the worker restarts.

Example:

from dag_parser.dynamic.dag_context import PostgresOperator

extract = SnowflakeOperator(
    task_id="extract",
    connection_id="snowflake_prod",
    sql="SELECT * FROM sales.orders WHERE region = %(region)s",
    parameters={"region": "US-WEST"},   # safely bound, not string-substituted
)

load_summary = PostgresOperator(
    task_id="load_summary",
    connection_id="reporting_pg",
    sql="INSERT INTO summary (region, total) VALUES (%(region)s, %(total)s)",
    parameters={"region": "US-WEST", "total": 12000},
)

56Data movement#

What it does: S3ToRedshiftOperator loads data from S3 into Redshift via COPY. GCSToBigQueryOperator loads data from GCS into BigQuery via a load job.

Parameters (S3ToRedshiftOperator):

Parameters (GCSToBigQueryOperator):

How it works:

• truncate=True and the subsequent COPY run inside a single transaction — if COPY fails for any reason, the truncate is rolled back and the table keeps its original data.
• Set exactly one of iam_role/s3_conn_id — the operator validates this at DAG-parse time.
• GCSToBigQueryOperator's poll loop tolerates a transient status-check error (retrying rather than failing the task immediately) and is bounded by the task's execution_timeout (feature 16) — set that explicitly if you need a hard cap on how long a load job is allowed to run.
• autodetect and schema_fields are mutually exclusive — set exactly one; the operator validates this at DAG-parse time.
• destination_project/destination_dataset fall back to the connection's stored defaults if omitted from the task — the same connection can target different datasets depending on whether a given task fills these in.

Example:

load_from_s3 = S3ToRedshiftOperator(
    task_id="load_from_s3",
    s3_conn_id="s3_default",
    redshift_conn_id="redshift_prod",
    s3_bucket="my-data-bucket",
    s3_key="exports/2026/01/*.csv",
    table="orders",
    schema="public",
    copy_options="CSV IGNOREHEADER 1 GZIP",
    truncate=True,   # truncate + COPY run as a single atomic transaction
)

load_from_gcs = GCSToBigQueryOperator(
    task_id="load_from_gcs",
    bigquery_conn_id="bq_prod",
    source_bucket="my-gcs-bucket",
    source_object="exports/*.parquet",
    destination_dataset="analytics",
    destination_table="orders",
    source_format="PARQUET",
    write_disposition="WRITE_TRUNCATE",
    autodetect=True,
    poll_interval=10,
    execution_timeout=1800,   # ceiling for the load job
)

57HTTP (one-shot call)#

What it does: Makes a single HTTP request as a task step — distinct from HttpSensor (feature 46), which polls repeatedly waiting for a condition.

Parameters (HttpOperator):

How it works:

• HttpOperator (also importable as SimpleHttpOperator) is a ready-made class — no need to subclass BaseOperator yourself.
• url/headers/body render through Go templating (feature 38), not Jinja.
• A response is truncated at 1MB, and any 2xx counts as success — there's no response-body validation option here (that's HttpSensor-only); add a follow-up Python task reading the pushed return_value if you need to inspect the body.

Example:

from dag_parser.dynamic.operators import HttpOperator

notify_downstream = HttpOperator(
    task_id="notify_downstream",
    url="https://api.example.com/v1/events",
    method="POST",
    headers={"Content-Type": "application/json"},
    connection_id="internal_api",
    body='{"event": "pipeline_complete", "dag_id": "{{ .DagID }}"}',
)

58SSH#

What it does: Opens an SSH connection and runs a single remote command.

Parameters:

How it works:

• SSH connections verify the remote host's key against the fingerprint stored on the connection — set this when creating the connection so Maestro-Pi can detect an unexpected host on the other end.
• environment={} may be silently ignored by the remote server depending on its sshd_config (most servers reject arbitrary SetEnv requests unless explicitly allowlisted) — a rejected env var is only logged as a warning, never surfaced as a task error. Prefer passing values through the command string itself if you need to guarantee they arrive.
• Output is return_value'd as stdout and stderr combined into one string (feature 37) — there's no way to capture them separately from this operator.
• The connection dial itself has a fixed 30-second timeout, separate from the command's own execution time — use execution_timeout (feature 16) to bound the whole task if the command itself might run long.

Example:

from dag_parser.dynamic.operators import SSHOperator

remote_cleanup = SSHOperator(
    task_id="remote_cleanup",
    connection_id="prod_bastion",   # host key verified against the connection's stored fingerprint
    command="rm -rf /tmp/staging/{{ .DS }}",
    environment={"STAGE": "prod"},  # may be silently dropped by sshd's AcceptEnv config
)

59Email / Slack (as task steps)#

What it does: EmailOperator and SlackAPIPostOperator send a message as a normal task step in the graph — distinct from the callback-based alert channels in Part 7, which fire on an event rather than running as their own scheduled step.

Parameters (EmailOperator):

Parameters (SlackAPIPostOperator):

How it works:

• Set from_email to override the "from" address for that specific task; leave it unset to use the deployment's globally-configured SMTP address.
• subject and html_content are both token-substituted the same way.
• files attaches the listed local file paths to the outgoing email; custom_headers are added as extra header lines on the outgoing message.
• SlackAPIPostOperator's return_value shape depends on the connection's mode: webhook mode returns Slack's raw response body (typically "ok"); token mode returns {"ts": "<message timestamp>"}. A downstream task reading this XCom should know which mode the connection uses.

Example:

from dag_parser.dynamic.dag_context import EmailOperator
from dag_parser.dynamic.dag_context import SlackAPIPostOperator

send_report = EmailOperator(
    task_id="send_report",
    to=["team@company.com"],
    subject="Report for {{dag_id}}",       # substituted
    html_content="<p>Report for {{ dag_id }} run {{ run_id }}</p>",
    from_email="pipeline-alerts@company.com",
    files=["/tmp/report.csv"],
    custom_headers={"X-Priority": "1"},
)

post_to_slack = SlackAPIPostOperator(
    task_id="post_to_slack",
    connection_id="slack_bot_token",   # token-mode: channel is REQUIRED
    channel="#data-alerts",
    text="Report generation complete",
    unfurl_links=True,
)

60Databricks#

What it does: DatabricksSubmitRunOperator submits a one-off run via the Databricks Jobs API, on either an existing or an ephemeral cluster, and waits for it to finish.

Parameters:

How it works:

• Set exactly one of cluster_id or new_cluster per task — setting both (or neither, with no connection default available) raises a clear validation error at DAG-parse time.
• cluster_id falls back to the connection's own stored default if you omit it from the task — handy if most tasks target the same cluster.
• The poll loop tolerates a single transient HTTP error (retrying rather than failing the task immediately) and is bounded by the task's execution_timeout (feature 16) — set that explicitly to cap how long you're willing to wait for a Databricks job.
• A notebook_task that calls dbutils.notebook.exit(...) is the most reliable way to get a meaningful return_value back — spark_python_task/ spark_jar_task runs typically don't populate one the same way.
• If you set a static idempotency_token on a task that also has retries configured, a retry using the same token may be deduplicated by Databricks itself — only do this if you specifically want that dedup behavior.

Example:

from dag_parser.dynamic.dag_context import DatabricksSubmitRunOperator

run_notebook = DatabricksSubmitRunOperator(
    task_id="run_notebook",
    connection_id="databricks_prod",
    task_type="notebook_task",
    notebook_path="/Shared/etl/daily_aggregate",
    base_parameters={"run_date": "{{ .DS }}"},
    new_cluster={  # wins over any cluster_id/connection default if both are set
        "spark_version": "14.3.x-scala2.12",
        "node_type_id": "i3.xlarge",
        "num_workers": 2,
    },
    poll_interval=15,
    execution_timeout=3600,   # bound the otherwise-unbounded poll loop
)

61Cross-DAG orchestration, summarized#

What it does: A recap of the operators that compose multiple DAGs together or wait on state outside the current DAG's own task graph: TriggerDagRunOperator (feature 27) plus the four sensors (Part 8).

How it works:

• TriggerDagRunOperator's wait_for_completion=True uses the same lightweight, slot-free waiting mechanism described in feature 27 — it does not hold a worker slot, unlike a mode="poke" sensor. Its poke_interval controls how often the background reconciler checks the child run's state while waiting.
• TriggerDagRunOperator is treated as lightweight coordination work for scheduling purposes, the same as the sensors — it is not subject to the same admission limits as CPU-heavy Python/Bash tasks, so it stays responsive even under system load.
• The sensors (Part 8) each wait on a condition (an HTTP check, a SQL query, a clock, or another task/DAG's state) using the poke/reschedule model (feature 51).

Example:

from dag_parser.dynamic.dag_context import TriggerDagRunOperator
from dag_parser.dynamic.operators import ExternalTaskSensor

with DAG(dag_id="orchestrator", schedule="@daily", start_date=datetime(2026, 1, 1)) as dag:

    trigger_child = TriggerDagRunOperator(
        task_id="trigger_child",
        trigger_dag_id="child_pipeline",
        wait_for_completion=True,
        poke_interval=30,   # how often the reconciler checks the child run's state
    )

    wait_for_sibling = ExternalTaskSensor(
        task_id="wait_for_sibling",
        external_dag_id="sibling_pipeline",
        allowed_states=["success"],
        mode="reschedule",
    )