Refactor operator `DbtConsumerWatcherSensor` for reusability by tatiana · Pull Request #2245 · astronomer/astronomer-cosmos

tatiana · 2026-01-05T16:23:34Z

This PR extracts basic watcher operator features that will be reused by the ExecutionMode.WATCHER_KUBERNETES implementation in #2207, so that the last PR related to introducing ExecutionMode.WATCHER_KUBERNETES becomes smaller and easier to review.

This PR does not add any new features or functionality. The main changes are:

Remove definitions that will be shared between ExecutionMode.WATCHER and ExecutionMode.WATCHER_KUBERNETES from cosmos.operators.watcher.BaseConsumerSensor to cosmos.operators._watcher.base.DbtConsumerWatcherSensor
Move cosmos.operators.watcher._store_dbt_resource_status_from_log to cosmos.operators._watchers.base._store_dbt_resource_status_from_log

How the changes were validated

In addition to the automated tests we run in the CI, I ran manually three different DAGs that use ExecutionMode.WATCHER:

example_watcher that is in our repo in dev/dags/example_watcher.py
example_watcher_synchronous that is in our repo in dev/dags/example_watcher.py
modified version of example_watcher to validate subprocess, where I changed execution_config to:

    execution_config=ExecutionConfig(
        execution_mode=ExecutionMode.WATCHER,
        invocation_mode=InvocationMode.SUBPROCESS,
        dbt_executable_path="/Users/tatiana.alchueyr/Code/astronomer-cosmos/venv/bin/dbt"
        ),

Screenshots with the successful runs:

DAG overview:

Using InvocationMode.DBT_RUNNER

Using InvocationMode.SUBPROCESS

Consumers continue sensing:

Runs with synchronous sensors:

netlify · 2026-01-05T16:23:39Z

✅ Deploy Preview for astronomer-cosmos canceled.

Name	Link
🔨 Latest commit	`5d93d86`
🔍 Latest deploy log	https://app.netlify.com/projects/astronomer-cosmos/deploys/695cfc91e881280007d82526

…he K8s watcher operator

Copilot

Pull request overview

This PR refactors the watcher operator implementation by extracting common functionality into a new base module (cosmos/operators/_watcher/base.py), preparing for the upcoming Kubernetes watcher implementation. The refactoring moves shared sensor logic, utility functions, and constants to reusable components while maintaining existing functionality.

Key changes:

Created BaseConsumerSensor class to encapsulate common sensor behavior previously in DbtConsumerWatcherSensor
Moved utility functions (_store_dbt_resource_status_from_log) and constants to shared locations
Updated import paths throughout test files to reflect the new module structure

Reviewed changes

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

Show a summary per file

File	Description
`cosmos/constants.py`	Added new watcher-related constants for priority weights and weight rules
`cosmos/operators/watcher.py`	Refactored to use new `BaseConsumerSensor` parent class and removed duplicated code
`cosmos/operators/_watcher/base.py`	New base module containing `BaseConsumerSensor` class and shared utility functions
`tests/operators/test_watcher.py`	Updated test patches to reference new module paths and renamed test methods
`tests/hooks/test_subprocess.py`	Updated test patches to reference new module location

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

codecov · 2026-01-05T17:07:38Z

Codecov Report

❌ Patch coverage is 98.59155% with 2 lines in your changes missing coverage. Please review.
✅ Project coverage is 97.97%. Comparing base (1085a3c) to head (a993b0d).

Files with missing lines	Patch %	Lines
cosmos/operators/_watcher/base.py	98.40%	2 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #2245      +/-   ##
==========================================
- Coverage   97.99%   97.97%   -0.03%     
==========================================
  Files          96       97       +1     
  Lines        6245     6273      +28     
==========================================
+ Hits         6120     6146      +26     
- Misses        125      127       +2

☔ 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.

#2245

pankajkoti

Looks solid refactor to me!

Would be nice to test DAGs once with astro CLI / airflow standalone if not done already.

…b.com/astronomer/astronomer-cosmos/pull/2245\#discussion_r2664617989

pankajastro

LGTM

…pull/2245/changes\#r2664673872

Co-authored-by: Pankaj Koti <pankajkoti699@gmail.com>

…ull/2245/changes\#r2664693660

@pankajkoti

Features * Support cross-referencing models across dbt projects using dbt-loom by @pankajkoti in #2271 * Support use of YAML selectors when using ``LoadMode.DBT_MANIFEST`` by @YourRoyalLinus in #2261 * Introduce ``ExecutionMode.WATCHER_KUBERNETES`` to use the watcher with ``KubernetesPodOperator`` by @tatiana in #2207 * Add support for StarRocks profile mapping by @kurkim0661 in #2256 * Allow pushing URIs as XComs for Cosmos tasks by @corsettigyg in #2275 * Support defining custom callbacks alongside the ``WATCHER_KUBERNETES`` callback by @johnhoran in #2307 Enhancements * Refactor: remove duplicate ``_construct_dest_file_path`` by @jx2lee in #2077 * Leverage Airflow ``::group::`` to group logs associated with DAG parsing by @tatiana in #2235 * Refactor ``DbtConsumerWatcherSensor`` for reusability by @tatiana in #2245 * Restore plain text output when using ``ExecutionMode.WATCHER`` by @tiovader in #2241 Bug Fixes * Fix running empty models or ephemeral nodes in ``ExecutionMode.WATCHER`` by @tatiana in #2279 * Improve watcher producer task priority in scheduling and the UI by @tatiana in #2237 * Fix typos and formatting issues in documentation by @pankajkoti in #2259 * Allow watcher producer retries without erroring by @tatiana in #2283 * Fix ``TestBehavior.AFTER_ALL`` is missing project_name information when loading project using manifest file by @tuantran0910 in #2242 * Fix duplicate log lines in watcher subprocess execution and format timestamps by @pankajkoti in #2301 Docs * Add Watcher Kubernetes documentation by @tatiana in #2303 * Document newly added telemetry metrics in the privacy notice by @pankajkoti in #2249 * Add compatibility policy document by @pankajastro in #2251 * Improve watcher documentation related to dbt threads by @tatiana in #2273 * Fix link in watcher execution mode documentation by @jedcunningham in #2277 * Update Apache Airflow minimum compatibility policy by @tatiana in #2285 * Clarify Cosmos runtime support until "End of Basic Support" by @jedcunningham in #2286 * Update watcher docs by @tatiana in #2298 * Update watcher kubernetes documentation by @tatiana in #2306 Others * Add Airflow 3 DAG versioning tests for Cosmos by @michal-mrazek in #2177 * Add dbt Core 1.11 to the test matrix by @tatiana in #2230 * Add integration tests using InvocationMode.SUBPROCESS and validate output by @tatiana in #2287 * Fix main branch failing tests by @tatiana in #2296 * Update pre-commit hooks to the latest versions by @jedcunningham in #2289 * Pre-commit autoupdates by @pre-commit in #2222, #2264, #2274 and #2290 * Dependabot updates by @dependabot in #2218, #2219, #2220, #2280 and #2284 * Add Scarf metrics to understand Cosmos feature usage patterns - Add telemetry tracking for dbt docs plugin usage by @pankajkoti in #2240 - Add DAG run telemetry metrics for load mode, invocation, and render_config parameters by @pankajkoti in #2223 - Collect profile metrics for DAG runs by @pankajastro in #2228 - Compress telemetry metadata to reduce serialized DAG size by @pankajkoti in #2252 - Skip storing telemetry metadata when emission is disabled by @pankajkoti in #2278 - Hide telemetry metadata parameters from the Airflow trigger UI by @pankajkoti in #2247 closes: astronomer/oss-integrations-private#317 --------- Co-authored-by: Tatiana Al-Chueyr <tatiana.alchueyr@gmail.com>

… sensor DbtConsumerWatcherSensor did not pass deferrable (or producer_task_id) to super().__init__() after the refactor in PR #2245, so operator_args={"deferrable": False} was ignored and sensors always deferred. Pass both through so the base class receives them. related: #2245

… sensor (#2384) `DbtConsumerWatcherSensor` did not pass `deferrable` (or `producer_task_id`) to `super().__init__()` after the refactor in PR #2245, so `operator_args={"deferrable": False}` was ignored and sensors always deferred. Pass both through so the base class receives them. related: #2245

tatiana temporarily deployed to internal January 5, 2026 16:23 — with GitHub Actions Inactive

tatiana changed the title ~~Refactor watcher operators for reusable base definitions~~ Refactor watcher operators so they can be reused by kubernetes_watcher Jan 5, 2026

tatiana changed the title ~~Refactor watcher operators so they can be reused by kubernetes_watcher~~ Refactor watcher operators for reusability Jan 5, 2026

tatiana temporarily deployed to internal January 5, 2026 16:32 — with GitHub Actions Inactive

tatiana added 2 commits January 5, 2026 16:33

Refactor watcher operators so the base definitions can be reused by t…

25bc89c

…he K8s watcher operator

Fix unittests

7c96d1b

tatiana force-pushed the refactor-watcher-into-base-class branch from f13fa25 to 7c96d1b Compare January 5, 2026 16:34

tatiana marked this pull request as ready for review January 5, 2026 16:35

Copilot AI review requested due to automatic review settings January 5, 2026 16:35

tatiana temporarily deployed to internal January 5, 2026 16:35 — with GitHub Actions Inactive

tatiana requested review from pankajastro and pankajkoti January 5, 2026 16:35

Copilot AI reviewed Jan 5, 2026

View reviewed changes

Comment thread cosmos/constants.py

Comment thread cosmos/operators/_watcher/base.py

Comment thread cosmos/operators/_watcher/base.py Outdated

tatiana temporarily deployed to internal January 5, 2026 16:38 — with GitHub Actions Inactive

Improve test coverage

a993b0d

tatiana force-pushed the refactor-watcher-into-base-class branch from b3ef8b6 to a993b0d Compare January 5, 2026 17:52

tatiana temporarily deployed to internal January 5, 2026 17:53 — with GitHub Actions Inactive

tatiana changed the title ~~Refactor watcher operators for reusability~~ Refactor operator DbtConsumerWatcherSensor for reusability Jan 5, 2026

tatiana added a commit that referenced this pull request Jan 6, 2026

Fix unittests after rebase based on

d407ae2

#2245

tatiana mentioned this pull request Jan 6, 2026

Introduce ExecutionMode.WATCHER_KUBERNETES to use watcher with KubernetesPodOperator #2207

Merged

pankajkoti approved these changes Jan 6, 2026

View reviewed changes

Comment thread cosmos/operators/_watcher/base.py Outdated

Comment thread cosmos/operators/watcher.py Outdated

Comment thread cosmos/operators/watcher.py Outdated

Simplify DbtConsumerWatcherSensor based on code review: https://githu…

0b5b237

…b.com/astronomer/astronomer-cosmos/pull/2245\#discussion_r2664617989

tatiana temporarily deployed to internal January 6, 2026 11:59 — with GitHub Actions Inactive

pankajastro approved these changes Jan 6, 2026

View reviewed changes

Comment thread cosmos/operators/_watcher/base.py Outdated

Comment thread cosmos/operators/_watcher/base.py Outdated

Address CR feedback: https://github.com/astronomer/astronomer-cosmos/…

6e077eb

…pull/2245/changes\#r2664673872

tatiana temporarily deployed to internal January 6, 2026 12:03 — with GitHub Actions Inactive

Apply suggestion from @pankajkoti

7107ec0

Co-authored-by: Pankaj Koti <pankajkoti699@gmail.com>

tatiana temporarily deployed to internal January 6, 2026 12:04 — with GitHub Actions Inactive

Address CR feedback https://github.com/astronomer/astronomer-cosmos/p…

75243ff

…ull/2245/changes\#r2664693660

tatiana temporarily deployed to internal January 6, 2026 12:07 — with GitHub Actions Inactive

Fix broken tests after applying CR feedback

5d93d86

tatiana temporarily deployed to internal January 6, 2026 12:14 — with GitHub Actions Inactive

tatiana merged commit acf6790 into main Jan 6, 2026
87 checks passed

tatiana deleted the refactor-watcher-into-base-class branch January 6, 2026 12:30

tatiana added this to the Cosmos 1.13.0 milestone Jan 29, 2026

pankajastro mentioned this pull request Jan 29, 2026

Release 1.13.0 #2297

Merged

pankajkoti mentioned this pull request Feb 20, 2026

fix(watcher): Respect deferrable=False from operator_args on consumer sensor #2384

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Refactor operator `DbtConsumerWatcherSensor` for reusability#2245

Refactor operator `DbtConsumerWatcherSensor` for reusability#2245
tatiana merged 8 commits into
mainfrom
refactor-watcher-into-base-class

tatiana commented Jan 5, 2026 •

edited

Loading

Uh oh!

netlify Bot commented Jan 5, 2026 •

edited

Loading

Uh oh!

Copilot AI left a comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

codecov Bot commented Jan 5, 2026 •

edited

Loading

Uh oh!

pankajkoti left a comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

pankajastro left a comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

Conversation

tatiana commented Jan 5, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

netlify Bot commented Jan 5, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

✅ Deploy Preview for astronomer-cosmos canceled.

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Uh oh!

Uh oh!

Uh oh!

codecov Bot commented Jan 5, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

pankajkoti left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

pankajastro left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

tatiana commented Jan 5, 2026 •

edited

Loading

netlify Bot commented Jan 5, 2026 •

edited

Loading

codecov Bot commented Jan 5, 2026 •

edited

Loading