Hey all, I'm still new to Prefect but haven't been able resolve a deployment error. I'm able to successfully run my main flow file (daily_run.py), but when I run

deployment build

I get the following errors (in thread). Perhaps they are related, but at this point I'm stumped. Not sure why it runs but cant be deployed. Any help would be appreciated and I'm happy to provide more info as needed. Thanks!

Hi Michael. Can you please share your flow code and build command?

@Jeff Hale here you go: Flow

from prefect import task, flow
from prefect.blocks.notifications import SlackWebhook
from prefect.filesystems import GitHub
from prefect.filesystems import GCS
from prefect_github import GitHubCredentials
from prefect_dbt.cli.commands import trigger_dbt_cli_command
from prefect_dbt.cli.configs import BigQueryTargetConfigs
from prefect_gcp import GcpCredentials
import shutil
import os

#Slack Credentials Block
slack_webhook_block = SlackWebhook.load("slack-webhook")


# GitHub Credentials Block
github_credentials_block = GitHubCredentials.load("get-dbt-repo")
token = github_credentials_block.token
repo = "<https://github.com/[owner]/[repo].git>"
    
github_configs = GitHub(
    repository=repo,
    access_token=token,
)

# BigQuery Credentials Block
gcp_credentials_block = GcpCredentials.load("gcp-credentials")
target_configs = BigQueryTargetConfigs(
    schema="[dataset]",
    project="[project]",
    credentials=gcp_credentials_block,
)

# Google Storage Block
gcs_block = GCS.load("google-storage-block")


@task
def clone_dbt_repo():
    shutil.rmtree("dbt_temp/", ignore_errors=True)
    github_configs.get_directory(local_path="dbt_temp/")
    print("Cloned GitHub directory.")


@flow
def run_dbt_deps():
    try:
        trigger_dbt_cli_command("dbt deps")
    except Exception as e:
        error_msg = repr(e)
        print('error caught:: ', error_msg)
        slack_webhook_block.notify(error_msg)

@flow
def run_dbt_build():
    try:
        trigger_dbt_cli_command("dbt build")
    except Exception as e:
        error_msg = repr(e)
        print('error caught:: ', error_msg)
        slack_webhook_block.notify(error_msg)

@task
def delete_dbt_repo():
    shutil.rmtree("dbt_temp/", ignore_errors=True)

@task
def send_slack_message():
    slack_webhook_block.notify("Run completed")


@flow
def main_wrapper_flow():
    clone_dbt_repo()
    os.chdir("dbt_temp")
    run_dbt_deps()
    run_dbt_build()
    os.chdir("..")
    delete_dbt_repo()
    send_slack_message()
    
if __name__ == "__main__":
    main_wrapper_flow()

Deploy

prefect deployment build daily_run.py:main_wrapper_flow --name "Test Deployment" --storage-block gcs/google-storage-block

Thanks, Michael. I believe you need to load that GitHub block in a task or flow, or use async.

Appreciate the help. I put them within the tasks and that error seems to be resolved. I have a new one w/ certificates and publishing w/ Google Storage but I'll dig into that. Generally speaking, is it best practice to put the load blocks within tasks/flows?

Cool. Yep!