Introduce the concept of an interceptor in Cosmos tasks

[tatiana]

Sometimes users may need to run commands before the dbt command is executed in Cosmos tasks.

This behaviour is similar to the callback feature, but instead of happening after dbt command execution, it happens before.

A use case is to calculate dbt or environment variables at runtime and use them during dbt command execution.

Example:

def my_interceptor(context, operator):
    # Modify vars (e.g. add execution date)
    if operator.vars is None:
        operator.vars = {}
    operator.vars["run_date"] = str(context["data_interval_start"].date())
    # Modify env
    if operator.env is None:
        operator.env = {}
    operator.env["MY_VAR"] = "value"

# Example of usage in an operator
dbt_build = DbtBuildLocalOperator(
    ...,
    interceptors=[my_interceptor],
)

# Example of usage in a DbtTaskGroup
dbt_task_group = DbtTaskGroup(
    ...,
    operator_args={"interceptors": [my_interceptor]},
)

The current implementation works with all existing execution modes:

• Local
• Virtualenv
• Docker
• Kubernetes
• AWS EKS
• AWS ECS
• GCP Cloud Run Job
• Azure Container Instance
• Async (e.g. BigQuery)
• Watcher
• Watcher Kubernetes

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 97.96%. Comparing base (6f2403c) to head (65adf48).

Additional details and impacted files

@@           Coverage Diff           @@
##             main    #2419   +/-   ##
=======================================
  Coverage   97.95%   97.96%           
=======================================
  Files         102      102           
  Lines        7009     7019   +10     
=======================================
+ Hits         6866     6876   +10     
  Misses        143      143           

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[Copilot]

Pull request overview

Adds a new “interceptor” hook to Cosmos dbt operators so users can run callables before the dbt command is built/executed, enabling runtime injection/modification of vars and environment variables across execution modes.

Changes:

• Add interceptors to the base operator API and invoke them prior to building dbt commands.
• Wire interceptor invocation into container-based execution modes (Kubernetes, Docker, ECS, ACI, Cloud Run Job) and local execution.
• Add unit tests across execution modes and document the new operator_args option.

Reviewed changes

Copilot reviewed 15 out of 15 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
cosmos/operators/base.py	Introduces interceptors param and base invoke_interceptors() helper.
cosmos/operators/local.py	Invokes interceptors before build_cmd() for local execution.
cosmos/operators/kubernetes.py	Invokes interceptors before building kube args / executing pod.
cosmos/operators/docker.py	Invokes interceptors before building docker command / executing container.
cosmos/operators/aws_ecs.py	Invokes interceptors before building ECS overrides/command.
cosmos/operators/azure_container_instance.py	Invokes interceptors before building ACI command.
cosmos/operators/gcp_cloud_run_job.py	Invokes interceptors before building Cloud Run Job command.
docs/configuration/operator-args.rst	Documents the new interceptors operator arg + example usage.
tests/operators/test_local.py	Adds unit test asserting interceptors run and affect vars/env for local.
tests/operators/test_virtualenv.py	Adds unit test for virtualenv operators (inherited behavior).
tests/operators/test_kubernetes.py	Adds unit test for interceptor behavior in Kubernetes mode.
tests/operators/test_docker.py	Adds unit test for interceptor behavior in Docker mode.
tests/operators/test_aws_ecs.py	Adds unit test for interceptor behavior in ECS mode.
tests/operators/test_azure_container_instance.py	Adds unit test for interceptor behavior in ACI mode.
tests/operators/test_gcp_cloud_run_job.py	Adds unit test for interceptor behavior in Cloud Run Job mode.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull request overview

Copilot reviewed 15 out of 15 changed files in this pull request and generated no new comments.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[pankajkoti]

LGTM. Have a minor question inline.

And maybe you have thought of this already and have your rationale, but wondering if the name injectors could suit better instead of interceptors for the use case we're tackling here?

[tatiana]

Thanks a lot for the review, @pankajkoti !

And maybe you have thought of this already and have your rationale, but wondering if the name injectors could suit better instead of interceptors for the use case we're tackling here?

Interceptor typically refers to a mechanism that observes, modifies, or blocks an operation at a well-defined point in its lifecycle. It often implies hooking into a control flow to inspect or change something - exactly what this PR does (it lets users run code before dbt commands). Interceptors are common in middleware, request pipelines, event systems, and AOP (aspect-oriented programming).

Injectors, on the other hand, usually suggest adding or providing something (usually a dependency). In a dependency injection system, injectors provide objects with what they need; they typically don't observe or intervene in behaviour. The term focuses more on the insertion of dependencies than on the control of execution.

I think in this particular feature, interceptor is a better fit because:

1. It hooks into the execution flow, running beofre dbt commadns are built and executed, what interceptors are meant to do
2. It implies extension/modification of behaviour, not just insertion of data. Users could modify existing environment variables, dbt variables, and even the Airflow context, potentially changing the operator's behaviour.
3. Conceptually, interceptors sit along callbacks as lifecycle hooks

In contrast, injector would imply that the primary purpose is to insert something (like environment variables or config), which is part of what the new hook does, but not its defining behaviour.

We could evolve the current version of the interceptor to, depending on its output, not even run the dbt command. At the moment, users could accomplish this by raising an exception, for instance. I'm curious about the community's feedback and how they'll use it.

[pankajkoti]

Injectors, on the other hand, usually suggest adding or providing something (usually a dependency). In a dependency injection system, injectors provide objects with what they need; they typically don't observe or intervene in behaviour. The term focuses more on the insertion of dependencies than on the control of execution.

In contrast, injector would imply that the primary purpose is to insert something (like environment variables or config), which is part of what the new hook does, but not its defining behaviour.

I see the association with dependency injection, which does come up in framework discussions. I am reading that injection as a term can be used more broadly across domains, for example, in Code injection, Command injection, Script injection, SQL injection, Chaos injection, fault injection (in testing), environment variable injection, feature flag injection, sidecar injection in Kubernetes, or trace context (trace ID) injection in observability, all of which influence runtime behaviour. Even dependency injection itself ultimately changes behaviour by supplying different implementations.

I support interceptor :), I think it fits well, just wanted to share why injector also felt reasonable to me and why it came to mind.