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 Contribution Guide (https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst)
Here are some useful points:

• Pay attention to the quality of your code (flake8, 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.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: [email protected]
  Slack: https://s.apache.org/airflow-slack

Keeping my eyeballs on this PR

Hello~ Following up to see if this needs anything else for review? 🥲

@epapineau can you post a screenshot of this working on single tenant based on your code changes?

@sungchun12 Added to the PR. Please let me know if you'd like any others :)

@epapineau How would users typically connect to these single-tenant instances? Does one API key have access to multiple, single-tenant instances or should one API only have access to an individual single-tenant instance?

The reason I ask is, if it's the latter, it might be better to have users provide a URL or tenant name in the Airflow connection rather than having to pass it in every operator/sensor instance.

@josh-fell one API key for one single tenant instance. A large customer may have multiple single tenant instances. Considering that, do you think it would be best to have the single tenant option on the connection, but leave it override-able at the operator/sensor, like with account_id?

@josh-fell Added tenant as an option to the Airflow connection. Let me know what you think.

... one API key for one single tenant instance.

So it sounds like users would need to configure an Airflow connection per API key they use? Or is it possible that a single API key could access a single tenant instance + a multi-tenant instance therefore the need for an override when using a single Airflow connection?

@josh-fell It is not possible that a single API key could access a single tenant instance + a multi-tenant instance. Each instance will require its own API key

@josh-fell It is not possible that a single API key could access a single tenant instance + a multi-tenant instance. Each instance will require its own API key

So if users will always need to use a separate connection for each API key they want to use because said key can't cross-pollinate with other instances, it seems like the move is to have tenant optionally configurable in the connection rather than overriding at the task level. If it's not populated in the connection, then it's assumed to be a multi-tenant instance which always has a "cloud" tenant name. Am I in the ballpark with that? Or what is a good use case to have the override at the task level?

@josh-fell Yep, totally follow. I removed the arg from the Operator. Do you have an opinion whether the tenant should be stored in the connection as the schema or as an extra?

@josh-fell Yep, totally follow. I removed the arg from the Operator. Do you have an opinion whether the tenant should be stored in the connection as the schema or as an extra?

Schema is great I think.

@josh-fell okay changes pushed 🎉

@josh-fell I was able to resolve the unit test failures locally by updating the value of dbt_cloud_conn_id to dbt_cloud (the name of my connection locally). Should I push those changes?

@josh-fell I was able to resolve the unit test failures locally by updating the value of dbt_cloud_conn_id to dbt_cloud (the name of my connection locally). Should I push those changes?

I don't think that will work for CI unfortunately. The test will be looking for a connection that's accessible by the CI environment. I can see a couple options to help resolve the failure:

1. Add a Connection and db.merge_conn() in the setup_class() of TestDbtCloudJobRunSensor like what's in TestDbtCloudHook. This would ensure a connection is always available for the test. Most straightforward to resolve the failure.
2. Remove the self.base_url logic from the constructor of the DbtCloudHook and implement a cached_property instead. This would make the hook constructor a little cleaner IMO, but would probably warrant a little bit of regression testing. The mock patch would skip over all of this logic and there would be no change to the test itself.

    @cached_property
    def base_url(self) -> str:
        tenant = self.connection.schema if self.connection.schema else "cloud"

        return f"https://{tenant}.getdbt.com/api/v2/accounts/"

I don't have a strong opinion either way.

@josh-fell cool I went with option one & it's passing now 🥳

Looks fantastic!! Would you mind rebasing to the latest main when you get a chance please? We can go ahead and merge afterwards.