New Optional dbt Cloud Job Operator Params

[ginone]

Summary

Allow triggering dbt Cloud jobs using project_name, environment_name, and job_name instead of job_id.

It is not a breaking change, but optional, alternative way of triggering a job - both options will work:

# regular job run by id
trigger_job_run1 = DbtCloudRunJobOperator(
    task_id="trigger_job_run1",
    job_id=48617,
    check_interval=10,
    timeout=300,
)

# equivalent job run by name
trigger_job_run3 = DbtCloudRunJobOperator(
    task_id="trigger_job_run3",
    project_name="my_dbt_project",
    environment_name="prod",
    job_name="my_dbt_job",
    check_interval=10,
    timeout=300,
)

This is beneficial in dynamically configured environments (e.g. managed by Infrastructure as Code) when job_id is not known upfront (or may change over time) and therefore hardcoding it is not convenient.

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide (https://github.com/apache/airflow/blob/main/contributing-docs/README.rst)
Here are some useful points:

• Pay attention to the quality of your code (ruff, mypy and type annotations). Our pre-commits will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example DAG that shows how users should use it.
• Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
• Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[jaklan]

@joellabes another MR related to dbt Cloud Operator from our side - could you have a look if you are fine with the approach?

[pikachuev]

Hi @ginone , great proposal and change. We use in our Airflow environment an own helper and retrieve job_ids by names. We figure out that in our case the payload from dbt Cloud REST API is quite big to retrieve it for every dbt cloud job triggering. To reduce payload size we additionally filter rest api calls by filtering project name in API call, for example:
project_url = (
f"{base_url_v3}accounts/{dbt_cloud_account_id}/projects/"
f"?account_id={dbt_cloud_account_id}"
f"&name__icontains={project_name}"
)

or
job_url = (
f"{base_url_v2}accounts/{dbt_cloud_account_id}/jobs/"
f"?account_id={dbt_cloud_account_id}"
f"&project_id={project_id}"
f"&name__icontains={job_name}"
)
What do you think about such improvement?

[ginone]

@pikachuev thanks for the feedback!

If you look at the code, get_job_by_name function specifically, you will notice there already is an optimization that works similarly to your suggestion. In order to minimize the response sizes, the process is done in 3 steps:

1. list_projects using project_name as filter (name__icontains={project_name})
2. list_environments using project_id as a filter
3. list_jobs using both project_id & environment_id as filters

I decided to only use name__icontains for first step to limit the response size as retrieving all projects would not make sense. In 2nd & 3rd stages my assumption was that it should not be necessary to use such filtering because there should be no more than a few environments in each project and a dozen or so jobs in each environment.

Now that you mentioned it, I think it would make sense to use name__icontains at least also for retrieving jobs, as there may be use cases when there are 100+ jobs in a single environment, so I'll submit that change shortly.

[jaklan]

Except of these very minor comments below - I think about one more approach.

Replace:

        self.job_id = job_id
        self.project_name = project_name
        self.environment_name = environment_name
        self.job_name = job_name

with*:

        self.job = job
        self.project = project
        self.environment = environment

where all these parameters can be either string or int.

• if job is int - that's all we need
• if it's string - we need project and environment
• if any of these two is int - it's passed directly to get_job_by_name as resource ID
• if any of these two is string - resource ID is retrieved by calling get_<project | environment>_by_name

This way we allow for more flexible setup (e.g. job by name, environment and project by id). Ofc instead of checking the type - we could keep separate parameters explicitly (== have <resource>_id and <resource>_name params for all three resources), but then we would need to either:

• add logic for mutually exclusive params
• document that <resource>_id has always higher priority than <resource>_name

*for backward compatibility we would need to keep job_id parameter, but we could mark it as deprecated

[joellabes]

could you have a look if you are fine with the approach?

I'm sorry that it's necessary (and have linked this internally for us to work on a better strategy!) but I think it's a sensible workaround for now, and is a pattern we've seen customers use before 👍

[ginone]

@josh-fell @Lee-W thank you for the code review, I've addressed all feedback. Could you please look again?

BTW. What do you think regarding @jaklan suggestion (#45634 (review))? My thoughts:

1. this (implicit argument types) feels wrong/confusing:

self.job: str | int,
self.project: str | int,
self.environment: str | int,

I think the params should be explicit, so I would go with:

self.job_id: int,
self.project_id: int,
self.environment_id: int,
self.job_name: str,
self.project_name: str,
self.environment_name: str,

2. I see this as an enhancement that could allow more flexible setups, but I would prefer to release this PR "as is" and unblock using project_name, environment_name, and job_name ASAP. I would create a new PR with the suggested enhancements.

[ginone]

@josh-fell @Lee-W thank you for the code review, I've addressed all feedback. Could you please look again?

BTW. What do you think regarding @jaklan suggestion (#45634 (review))? My thoughts:

1. this (implicit argument types) feels wrong/confusing:

self.job: str | int,
self.project: str | int,
self.environment: str | int,

I think the params should be explicit, so I would go with:

self.job_id: int,
self.project_id: int,
self.environment_id: int,
self.job_name: str,
self.project_name: str,
self.environment_name: str,

2. I see this as an enhancement that could allow more flexible setups, but I would prefer to release this PR "as is" and unblock using project_name, environment_name, and job_name ASAP. I would create a new PR with the suggested enhancements.

New PR that includes the changes described above: #46184

Feel free to select just one of my PRs and close the other one.

[Lee-W]

@josh-fell @Lee-W thank you for the code review, I've addressed all feedback. Could you please look again?

BTW. What do you think regarding @jaklan suggestion (#45634 (review))? My thoughts:

1. this (implicit argument types) feels wrong/confusing:

self.job: str | int,
self.project: str | int,
self.environment: str | int,

I think the params should be explicit, so I would go with:

self.job_id: int,
self.project_id: int,
self.environment_id: int,
self.job_name: str,
self.project_name: str,
self.environment_name: str,

2. I see this as an enhancement that could allow more flexible setups, but I would prefer to release this PR "as is" and unblock using `project_name`, `environment_name`, and `job_name` ASAP. I would create a new PR with the suggested enhancements.

Hi, sorry for the late reply. I prefer this idea more. It would be great if you could rebase from the main branch and resolve the conflict, then I'll review it again. Thanks a lot!

[ginone]

@Lee-W rebasing is done, please have a look again. Thanks!

[Lee-W]

Looks good to me. Thanks for your prompt reply. I'll keep it open for one to two days so others can take a look.

[ginone]

Looks good to me. Thanks for your prompt reply. I'll keep it open for one to two days so others can take a look.

Fixed one incorrect "include" path in docs. Checks should be all green now.

Thanks!

[boring-cyborg]

Awesome work, congrats on your first merged pull request! You are invited to check our Issue Tracker for additional contributions.