https://prefect.io
Join Slack
<@ULVA73B9P> Describe the use of a DbtCloudJob object to trigger a dbt flow. DbtCloudCredentials and...
# ask-marvin
j
@Marvin Describe the use of a DbtCloudJob object to trigger a dbt flow. DbtCloudCredentials and DbtCloudJob blocks have already been configured,
m
To help you with the usage of the 
DbtCloudJob
object for triggering a dbt flow in Prefect, we need to first determine the version of Prefect you are using because the implementation details can vary between Prefect 2.x and 3.x. Could you please confirm your Prefect version? If you are not sure, I can guide you based on typical usage patterns for recent versions.
j
@Marvin prefect 3.1
m
To use a 
DbtCloudJob
object to trigger a dbt flow in Prefect 3.1, you can follow these steps: 1. Copy the Job ID: - Navigate to your dbt Cloud account and go to the "Deploy" section. - Select the job you want to trigger. - Obtain the job ID from the URL, which will look like this: 
<https://cloud.getdbt.com/deploy/<ACCOUNT_ID>/projects/<PROJECT_ID>/jobs/<JOB_ID>>
. 2. Create the 
DbtCloudJob
object: - Use the following script, ensuring to replace the placeholders with your actual credentials block and job ID.
Copy code
python
from prefect_dbt.cloud import DbtCloudCredentials, DbtCloudJob

# Load your dbt cloud credentials
dbt_cloud_credentials = DbtCloudCredentials.load("CREDENTIALS-BLOCK-NAME")

# Configure the DbtCloudJob
dbt_cloud_job = DbtCloudJob(
    dbt_cloud_credentials=dbt_cloud_credentials,
    job_id="YOUR_JOB_ID"
).save("YOUR_JOB_BLOCK_NAME")
3. Load and Trigger the Job: - You can then load the saved block, which contains your credentials, to trigger the dbt job.
Copy code
python
from prefect_dbt.cloud import DbtCloudJob

# Load the previously saved job block
dbt_cloud_job = DbtCloudJob.load("YOUR_JOB_BLOCK_NAME")

# Trigger the job
dbt_cloud_job.run()
Make sure that the 
DbtCloudCredentials
and 
DbtCloudJob
blocks are set up correctly in your Prefect environment to allow these scripts to run. For more detailed information, you can refer to the Prefect documentation on dbt integration.
j
@Marvin This is what I entered:
Copy code
from prefect_dbt.cloud import DbtCloudJob
dbt_cloud_job = DbtCloudJob.load("mail")
dbt_cloud_job.run()
This is the response:
Copy code
AttributeError: 'DbtCloudJob' object has no attribute 'run'
m
It seems like there might be some confusion or a misunderstanding about how the 
DbtCloudJob
object is utilized in Prefect 3.x. Based on the error message you received, it appears that the 
DbtCloudJob
object doesn't have a 
run
method. Instead, the 
DbtCloudJob
is typically used to configure and schedule dbt Cloud jobs rather than directly executing them with a 
run
method. To trigger a dbt Cloud job, you might need to use a task or another mechanism that interacts with the dbt Cloud API using Prefect. Here's a corrected example of how you could structure it using Prefect tasks:
Copy code
python
from prefect import flow
from prefect_dbt.cloud import DbtCloudCredentials, DbtCloudJob

@flow
def trigger_dbt_job():
    # Load your dbt cloud job
    dbt_cloud_job = DbtCloudJob.load("mail")

    # Use a helper method or Prefect's task functionality to trigger the job if needed
    # Note: Ensure that the configuration is correct to trigger the job
    # This example is hypothetical as the function to directly trigger might differ
    # Check dbt and Prefect documentation for actual runnable task methods if provided

# Execute the flow
trigger_dbt_job()
If you have set up the 
DbtCloudCredentials
and 
DbtCloudJob
blocks correctly, you might need to look into how these jobs are scheduled and triggered from Prefect, as the 
run
method might not be applicable directly. It's also possible that this version of Prefect or the plugin has a specific way to schedule dbt jobs, which might require additional setup or methods provided by the integration. For the most accurate solution, double-check the Prefect 3.x documentation or your installed version's integration plugin details to ensure you use the correct method to trigger the job: Prefect 3.x Documentation. If this issue persists, consulting Prefect's community forums or their support might also provide more insights specific to your setup and version.
s
Hi @Jason Wiener! I made some changes to the documentation on that this afternoon. It hasn't made it into the actual docs yet, but here's a preview. Feel free to pass on any feedback on it.
The gist of it is here:
Copy code
from prefect import flow
from prefect_dbt.cloud import DbtCloudJob
from <http://prefect_dbt.cloud.jobs|prefect_dbt.cloud.jobs> import run_dbt_cloud_job
import asyncio

@flow
async def run_dbt_job_flow():
    result = await run_dbt_cloud_job(
        dbt_cloud_job = await DbtCloudJob.load("JOB-BLOCK-NAME-PLACEHOLDER"),
        targeted_retries = 0,
    )
    return await result

if __name__ == "__main__":
    asyncio.run(run_dbt_job_flow())
j
Thanks @Sean Williams! This snippet helped a lot. Interestingly, the job runs in dbt and I get a valid response that includes the correct job execution ID in each case but the function does not wait on the job to complete (I don't think the await in the return is needed but the only one that makes a difference to the error is the await on DbtCloudJob.load() - that's necessary.) and so the script always fails with an error like 
prefect_dbt.cloud.exceptions.DbtCloudJobRunFailed: dbt Cloud job ######## failed after 0 retries.
👀 1
But, in dbt, the job runs fine.
@Sean Williams the exception does not seem to be thrown when I use the trigger in a deployment rather than running it from my debugger so 🤷. In fact, this is working as hoped for in prod now. One note on the trigger method log line is that the URL format seems to be incorrect:
Copy code
<http://self.logger.info|self.logger.info>(
            f"dbt Cloud job {self.job_id} run {run_id} successfully triggered. "
            f"You can view the status of this run at "
            f"https://{self.dbt_cloud_credentials.domain}/#/accounts/"
            f"{self.dbt_cloud_credentials.account_id}/projects/"
            f"{run_data['project_id']}/runs/{run_id}/"
        )
Based on my execution this should be
Copy code
<http://self.logger.info|self.logger.info>(
            f"dbt Cloud job {self.job_id} run {run_id} successfully triggered. "
            f"You can view the status of this run at "
            f"https://{self.dbt_cloud_credentials.domain}/deploy/"
            f"{self.dbt_cloud_credentials.account_id}/projects/"
            f"{run_data['project_id']}/runs/{run_id}/"
        )
Thanks again for the timely pointer.
s
Oh, good catch. Thanks! I'll update that
Open in Slack
PreviousNext
Powered by

Prefect Community

Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Powered by