https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> teach me prefect-dbt step by step (db...
# ask-marvin
p
@Marvin teach me prefect-dbt step by step (dbt-Core)
m
✅ thought for 552.6 seconds
Awesome — let’s walk through prefect-dbt with dbt Core step by step. I’ll assume you’re on Prefect 3.x. If you’re on 2.x, let me know and I’ll adjust. Step 0 — Prereqs - Python 3.10+ - dbt-core 1.7+ (installed via extras below) - Prefect 3.4.9+ (required for the modern dbt runner) - An existing dbt project (or create one with 
dbt init
) Helpful docs: - prefect-dbt integration docs - Run dbt with Prefect (example) - prefect-dbt release notes - prefect-dbt API reference Step 1 — Install Pick the adapter(s) you use. The recommended install is via Prefect extras:
Copy code
# Base dbt-core support
pip install "prefect[dbt]"

# Add your adapter(s) as needed
pip install "prefect[dbt,snowflake]"
pip install "prefect[dbt,bigquery]"
pip install "prefect[dbt,postgres]"

# Optional: all adapters
pip install "prefect[dbt,all_extras]"
Register dbt blocks in Prefect (enables UI configuration of dbt-related blocks):
Copy code
prefect block register -m prefect_dbt
Step 2 — Point Prefect to your dbt project and profiles You can set these via environment variables or pass them via settings. Environment variables are auto-detected by PrefectDbtSettings:
Copy code
export DBT_PROJECT_DIR=/path/to/your/dbt_project
export DBT_PROFILES_DIR=/path/to/your/.dbt
# Optional:
export DBT_LOG_LEVEL=info
export DBT_TARGET_PATH=target
Step 3 — Minimal flow using PrefectDbtRunner (recommended) The modern way in Prefect 3.x is PrefectDbtRunner. It creates per-dbt-node tasks and Prefect assets automatically.
Copy code
# flow.py
from prefect import flow, task
from prefect_dbt import PrefectDbtRunner, PrefectDbtSettings

@task
def dbt_invoke(args: list[str], project_dir: str | None = None, profiles_dir: str | None = None):
    settings = PrefectDbtSettings(project_dir=project_dir, profiles_dir=profiles_dir)
    runner = PrefectDbtRunner(settings=settings)
    return runner.invoke(args)

@flow
def dbt_pipeline(project_dir: str = "./dbt_project", profiles_dir: str = "~/.dbt"):
    # 1) install packages
    dbt_invoke(["deps"], project_dir, profiles_dir)
    # 2) seed (optional)
    dbt_invoke(["seed"], project_dir, profiles_dir)
    # 3) run models
    dbt_invoke(["run"], project_dir, profiles_dir)
    # 4) run tests
    dbt_invoke(["test"], project_dir, profiles_dir)

if __name__ == "__main__":
    dbt_pipeline()
Notes: - Use 
["build"]
if you want models + tests (+ snapshots/seeds if configured) in one go. - You can pass CLI args as usual: 
["run", "--select", "stg_*", "--state", "state/"]
. Step 4 — Profiles and credentials Option A: Use your existing profiles.yml as-is (point 
DBT_PROFILES_DIR
to it). Option B: Template profiles.yml using Prefect blocks (great for secrets) In `profiles.yml`:
Copy code
my_profile:
  target: prod
  outputs:
    prod:
      type: snowflake
      account: "{{ prefect.blocks.snowflake-credentials.prod-creds.account }}"
      user: "{{ prefect.blocks.snowflake-credentials.prod-creds.user }}"
      password: "{{ prefect.blocks.snowflake-credentials.prod-creds.password }}"
      database: "{{ prefect.blocks.snowflake-connector.prod-conn.database }}"
      schema: "{{ prefect.blocks.snowflake-connector.prod-conn.schema }}"
      warehouse: "{{ prefect.blocks.snowflake-connector.prod-conn.warehouse }}"
      threads: 4
Prefect resolves those 
{{ prefect.blocks... }}
references automatically during 
runner.invoke(...)
. Option C: Environment variables in profiles.yml
Copy code
password: "{{ env_var('SNOWFLAKE_PASSWORD') }}"
Set it in your flow or worker environment before running dbt. Step 5 — Passing vars, selectors, and other CLI flags Just pass them in the args list (exactly as you’d use in dbt CLI): ``` dbt_invoke(["run", "--select", "tag:daily"]) dbt_invoke(["run", "--vars", '{"date": "2025-01-01"}'])
dbt_invoke(["build", "--threads", "8"])
Copy code
Step 6 — Handle failures and results
By default, PrefectDbtRunner raises on dbt failures. You can change that:
from prefect import flow from prefect_dbt import PrefectDbtRunner, PrefectDbtSettings @flow def tolerate_test_failures(project_dir="./dbt_project", profiles_dir="~/.dbt"): settings = PrefectDbtSettings(project_dir=project_dir, profiles_dir=profiles_dir) runner = PrefectDbtRunner(settings=settings, raise_on_failure=False) result = runner.invoke(["test"]) if not result.success: print("Tests failed, but continuing downstream...")
Copy code
Step 7 — Assets and lineage (automatic)
- Using PrefectDbtRunner automatically creates Prefect assets for dbt models/seeds/snapshots and links lineage.
- You can include compiled SQL in asset descriptions:
runner = PrefectDbtRunner(settings=settings, include_compiled_code=True)
Copy code
- Disable assets entirely if you want:
runner = PrefectDbtRunner(settings=settings, disable_assets=True)
Copy code
Step 8 — Run locally
python flow.py
Copy code
You’ll see each dbt node as a Prefect task with logs.

Step 9 — Deploy and run on a worker
- Create a process work pool (if you don’t already have one):
prefect work-pool create "dbt-pool" --type process
Copy code
- Start a worker to pick up runs:
prefect worker start --pool "dbt-pool"
Copy code
- Create a deployment (use your flow’s entrypoint and a name):
prefect deploy flow.py:dbt_pipeline --name "dbt-daily" --schedule "0 2 * * *"
Copy code
- Trigger a run:
prefect deployment run "dbt_pipeline/dbt-daily"
Copy code
Step 10 — Common variations
- Run only a subset:
dbt_invoke(["run", "--select", "my_model+"])
Copy code
- Retry failed nodes (useful with flow retries):
dbt_invoke(["retry"])
Copy code
- Separate steps with retries:
@task(retries=2, retry_delay_seconds=10) def run_models(...): ... ``` Notes on legacy APIs - In Prefect 3.x, prefer 
PrefectDbtRunner
. The older shell-based 
prefect_dbt.cli.commands.DbtCoreOperation
and single-command tasks exist for backward compatibility but are being phased out in favor of the runner model (better observability and assets). Where to go next - Integration docs: prefect-dbt integration docs - Full example: Run dbt with Prefect (example) - API reference: prefect-dbt API reference Want me to tailor this to your stack? - Which adapter are you using (Snowflake, BigQuery, Postgres, DuckDB, etc.)? - Where do you want to run flows (local process, Docker, Kubernetes)? - Do you prefer managing credentials via Prefect blocks or environment variables?
p
@Marvin i use dbt CLI POSTGRES TARGET CONFIGS block which credentails: SQLAlchemy Connector . how I reference in my profiles.yml? dev: dbname: "{{ prefect.blocks.postgres-dev-dbt-creds.database }}" host: "{{ prefect.blocks.postgres-dev-dbt-creds.host }}" pass: "{{ prefect.blocks.postgres-dev-dbt-creds.password }}" port: "{{ prefect.blocks.postgres-dev-dbt-creds.port }}" schema: public threads: 4 type: postgres user: "{{ prefect.blocks.postgres-dev-dbt-creds.user }}" not work
m
✅ thought for 204.4 seconds
Great question — a couple of key points to get this working: 1) The placeholder must include the block type slug, not just the block name - The pattern is: 
{{ prefect.blocks.<block-type-slug>.<block-name>.<field-path> }}
- In your case: - Postgres Target Configs block type slug is likely 
dbt-cli-postgres-target-configs
- The nested SQLAlchemy Connector values live under 
credentials.connection_info
2) Use dbt’s correct field names - dbt expects 
password
(not 
pass
) - dbt expects 
dbname
for Postgres 3) These placeholders are only resolved when you run through Prefect’s dbt integration - Plain 
dbt
won’t expand 
{{ prefect.blocks... }}
on its own. You need to run with Prefect’s resolver (e.g., 
PrefectDbtSettings.resolve_profiles_yml()
or the 
prefect-dbt
tasks). Here’s a corrected 
profiles.yml
example assuming your Postgres Target Configs block is named `postgres-dev-dbt-creds`:
Copy code
your_profile_name:
  target: dev
  outputs:
    dev:
      type: postgres
      host: "{{ prefect.blocks.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.credentials.connection_info.host }}"
      port: "{{ prefect.blocks.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.credentials.connection_info.port }}"
      user: "{{ prefect.blocks.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.credentials.connection_info.username }}"
      password: "{{ prefect.blocks.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.credentials.connection_info.password }}"
      dbname: "{{ prefect.blocks.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.credentials.connection_info.database }}"
      schema: "{{ prefect.blocks.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.schema }}"
      threads: "{{ prefect.blocks.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.threads }}"
How to run so it resolves those placeholders: - If you’re using the prefect-dbt integration, wrap the run with 
PrefectDbtSettings.resolve_profiles_yml()
(this creates a temporary 
profiles.yml
with the Prefect references resolved) or use the 
trigger_dbt_cli_command
task with a 
DbtCliProfile
. Minimal example using the resolver:
Copy code
from prefect_dbt.core.settings import PrefectDbtSettings
from prefect_dbt.core.runner import PrefectDbtRunner

settings = PrefectDbtSettings(profiles_dir="/path/to/your/profiles_dir")

with settings.resolve_profiles_yml() as temp_profiles_dir:
    runner = PrefectDbtRunner(
        settings=PrefectDbtSettings(profiles_dir=temp_profiles_dir)
    )
    runner.invoke(["run"])
Alternatively, if you want to reference the SQLAlchemy connector block directly (if you saved it separately), use the 
sqlalchemy-connector
block type slug, e.g.:
Copy code
host: "{{ prefect.blocks.sqlalchemy-connector.<your-connector-block-name>.connection_info.host }}"
If this still “doesn’t work,” can you share: - The exact name of your Postgres Target Configs block (as shown in the UI) - Whether you’re running dbt via Prefect (prefect-dbt) or calling 
dbt
directly - Any error message you see I’ll tailor the fix right away once I have those.
p
@Marvin dev: dbname: "{{ prefect.blocks.blocks.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.credentials.postgres-dev-conn.database }}" host: "{{ prefect.blocks.blocks.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.credentials.postgres-dev-conn.host }}" pass: "{{ prefect.blocks.blocks.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.credentials.postgres-dev-conn.password }}" port: "{{ prefect.blocks.blocks.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.credentials.postgres-dev-conn.port }}" schema: public threads: 4 type: postgres user: "{{ prefect.blocks.blocks.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.credentials.postgres-dev-conn.user }}"
m
✅ thought for 44.9 seconds
Thanks for sharing the snippet — a few issues to fix: - Remove the extra 
blocks
segment (
prefect.blocks.blocks...
→ 
prefect.blocks...
) - Don’t use the nested block’s name inside the path (
credentials.postgres-dev-conn...
is not valid) - Use dbt’s field names (
password
, not 
pass
) - Map SQLAlchemy connector fields under 
credentials.connection_info
Assuming your block type is Dbt CLI Postgres Target Configs and the block name is 
postgres-dev-dbt-creds
, here’s the corrected mapping:
Copy code
dev:
  type: postgres
  host: "{{ prefect.blocks.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.credentials.connection_info.host }}"
  port: "{{ prefect.blocks.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.credentials.connection_info.port }}"
  user: "{{ prefect.blocks.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.credentials.connection_info.username }}"
  password: "{{ prefect.blocks.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.credentials.connection_info.password }}"
  dbname: "{{ prefect.blocks.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.credentials.connection_info.database }}"
  schema: "{{ prefect.blocks.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.schema }}"
  threads: "{{ prefect.blocks.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.threads }}"
Notes: - The correct block type slug is likely 
dbt-cli-postgres-target-configs
. - The SQLAlchemy connector fields live under 
credentials.connection_info
. - If you want to reference the SQLAlchemy Connector block directly (if saved separately), use the 
sqlalchemy-connector
type slug, e.g.:
Copy code
host: "{{ prefect.blocks.sqlalchemy-connector.postgres-dev-conn.connection_info.host }}"
Important: These 
{{ prefect.blocks... }}
placeholders are only resolved when you run dbt via Prefect’s dbt integration (e.g., using 
PrefectDbtSettings.resolve_profiles_yml()
or 
trigger_dbt_cli_command
). Plain 
dbt
won’t resolve them on its own. Quick way to debug what Prefect resolves:
Copy code
from pathlib import Path
from prefect_dbt.core.settings import PrefectDbtSettings

settings = PrefectDbtSettings(profiles_dir="/path/to/your/profiles_dir")
with settings.resolve_profiles_yml() as temp_profiles_dir:
    print(Path(temp_profiles_dir, "profiles.yml").read_text())
If anything still fails, please share: - The exact name of your Postgres Target Configs block - Whether you’re running dbt via Prefect (prefect-dbt) or calling 
dbt
directly - Any error messages you see
p
@Marvin dev: dbname: "{{ prefect.dbt-cli-postgres-target-configs.blocks.postgres-dev-dbt-creds.credentials.connection_info.database }}" host: "{{ prefect.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.credentials.connection_info.host }}" pass: "{{ prefect.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.credentials.connection_info.password }}" port: "{{ prefect.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.credentials.connection_info.port }}" schema: public threads: 4 type: postgres user: "{{ prefect.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.credentials.connection_info.username }}" Worker 'DockerWorker de2e1e4e-54a3-4572-90e6-8bcccd50965d' submitting flow run '019aec25-8840-710b-b429-f5a5d529096e' 093452 AM prefect.flow_runs.worker Completed submission of flow run '019aec25-8840-710b-b429-f5a5d529096e' 093452 AM prefect.flow_runs.worker Opening process... 093453 AM prefect.flow_runs.runner > Running set_working_directory step... 093454 AM prefect.flow_runs Beginning flow run 'copper-agama' for flow 'main' 093500 AM prefect.flow_runs Task run failed with exception: ObjectNotFound(None) Traceback (most recent call last): File "/opt/prefect/.venv/lib/python3.11/site-packages/prefect/client/orchestration/_blocks_types/client.py", line 367, in read_block_document_by_name response = await self.request( ^^^^^^^^^^^^^^^^^^^ File "/opt/prefect/.venv/lib/python3.11/site-packages/prefect/client/orchestration/base.py", line 53, in request return await self._client.send(request) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/opt/prefect/.venv/lib/python3.11/site-packages/prefect/client/base.py", line 379, in send response.raise_for_status() File "/opt/prefect/.venv/lib/python3.11/site-packages/prefect/client/base.py", line 163, in raise_for_status raise PrefectHTTPStatusError.from_httpx_error(exc) from exc.cause prefect.exceptions.PrefectHTTPStatusError: Client error '404 Not Found' for url 'http://prefect-server:4200/api/block_types/slug/postgres-prod-dbt-creds/block_documents/name/database?include_secrets=true' Response: {'detail': 'Block document not found'} For more information check: https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/404 The above exception was the direct cause of the following exception: Traceback (most recent call last): File "/opt/prefect/.venv/lib/python3.11/site-packages/prefect/task_engine.py", line 884, in run_context yield self File "/opt/prefect/.venv/lib/python3.11/site-packages/prefect/task_engine.py", line 1538, in run_task_sync engine.call_task_fn(txn) File "/opt/prefect/.venv/lib/python3.11/site-packages/prefect/task_engine.py", line 901, in call_task_fn result = call_with_parameters(self.task.fn, parameters) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/opt/prefect/.venv/lib/python3.11/site-packages/prefect/utilities/callables.py", line 210, in call_with_parameters return fn(*args, **kwargs) ^^^^^^^^^^^^^^^^^^^ File "/opt/prefect/flows/flow.py", line 94, in run_dbt_job runner.invoke(["build"]) File "/opt/prefect/.venv/lib/python3.11/site-packages/prefect_dbt/core/runner.py", line 1066, in invoke with self.settings.resolve_profiles_yml() as profiles_dir: File "/usr/local/lib/python3.11/contextlib.py", line 137, in enter return next(self.gen) ^^^^^^^^^^^^^^ File "/opt/prefect/.venv/lib/python3.11/site-packages/prefect_dbt/core/settings.py", line 94, in resolve_profiles_yml profiles_yml = run_coro_as_sync( ^^^^^^^^^^^^^^^^^ File "/opt/prefect/.venv/lib/python3.11/site-packages/prefect/utilities/asyncutils.py", line 207, in run_coro_as_sync return call.result() ^^^^^^^^^^^^^ File "/opt/prefect/.venv/lib/python3.11/site-packages/prefect/_internal/concurrency/calls.py", line 365, in result return self.future.result(timeout=timeout) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/opt/prefect/.venv/lib/python3.11/site-packages/prefect/_internal/concurrency/calls.py", line 192, in result return self.__get_result() ^^^^^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.11/concurrent/futures/_base.py", line 401, in __get_result raise self._exception File "/opt/prefect/.venv/lib/python3.11/site-packages/prefect/_internal/concurrency/calls.py", line 441, in _run_async result = await coro ^^^^^^^^^^ File "/opt/prefect/.venv/lib/python3.11/site-packages/prefect/utilities/asyncutils.py", line 188, in coroutine_wrapper return await task ^^^^^^^^^^ File "/opt/prefect/.venv/lib/python3.11/site-packages/prefect/client/utilities.py", line 99, in with_injected_client return await fn(*args, **kwargs) ^^^^^^^^^^^^^^^^^^^^^^^^^ File "/opt/prefect/.venv/lib/python3.11/site-packages/prefect/utilities/templating.py", line 317, in resolve_block_document_references updated_value = await resolve_block_document_references( ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/opt/prefect/.venv/lib/python3.11/site-packages/prefect/client/utilities.py", line 99, in with_injected_client return await fn(*args, **kwargs) ^^^^^^^^^^^^^^^^^^^^^^^^^ File "/opt/prefect/.venv/lib/python3.11/site-packages/prefect/utilities/templating.py", line 317, in resolve_block_document_references updated_value = await resolve_block_document_references( ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/opt/prefect/.venv/lib/python3.11/site-packages/prefect/client/utilities.py", line 99, in with_injected_client return await fn(*args, **kwargs) ^^^^^^^^^^^^^^^^^^^^^^^^^ File "/opt/prefect/.venv/lib/python3.11/site-packages/prefect/utilities/templating.py", line 317, in resolve_block_document_references updated_value = await resolve_block_document_references( ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/opt/prefect/.venv/lib/python3.11/site-packages/prefect/client/utilities.py", line 99, in with_injected_client return await fn(*args, **kwargs) ^^^^^^^^^^^^^^^^^^^^^^^^^ File "/opt/prefect/.venv/lib/python3.11/site-packages/prefect/utilities/templating.py", line 317, in resolve_block_document_references updated_value = await resolve_block_document_references( ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/opt/prefect/.venv/lib/python3.11/site-packages/prefect/client/utilities.py", line 99, in with_injected_client return await fn(*args, **kwargs) ^^^^^^^^^^^^^^^^^^^^^^^^^ File "/opt/prefect/.venv/lib/python3.11/site-packages/prefect/utilities/templating.py", line 349, in resolve_block_document_references block_document = await client.read_block_document_by_name( ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/opt/prefect/.venv/lib/python3.11/site-packages/prefect/client/orchestration/_blocks_types/client.py", line 375, in read_block_document_by_name raise ObjectNotFound(http_exc=e) from e prefect.exceptions.ObjectNotFound: None 093500 AM run_dbt_job-822 prefect.task_runs Finished in state Failed('Task run encountered an exception ObjectNotFound: None')
m
Your message was too long, here's your message at the allowed limit: dev dbname {{ prefect dbt-cli-postgres-target-configs blocks postgres-dev-dbt-creds credentials connection_info database }} host {{ prefect dbt-cli-postgres-target-configs postgres-dev-dbt-creds credentials connection_info host }} pass {{ prefect dbt-cli-postgres-target-configs postgres-dev-dbt-creds credentials connection_info password }} port {{ prefect dbt-cli-postgres-target-configs postgres-dev-dbt-creds credentials connection_info port }} schema public threads 4 type postgres user {{ prefect dbt-cli-postgres-target-configs postgres-dev-dbt-creds credentials connection_info username }} Worker 'DockerWorker de2e1e4e-54a3-4572-90e6-8bcccd50965d' submitting flow run '019aec25-8840-710b-b429-f5a5d529096e' 09 34 52 AM prefect flow_runs worker Completed submission of flow run '019aec25-8840-710b-b429-f5a5d529096e' 09 34 52 AM prefect flow_runs worker Opening process 09 34 53 AM prefect flow_runs runner > Running set_working_directory step 09 34 54 AM prefect flow_runs Beginning flow run 'copper-agama' for flow 'main' 09 35 00 AM prefect flow_runs Task run failed with exception ObjectNotFound(None) Traceback (most recent call last) File /opt/prefect/ venv/lib/python3 11/site-packages/prefect/client/orchestration/_blocks_types/client py line 367 in read_block_document_by_name response = await self request( ^^^^^^^^^^^^^^^^^^^ File /opt/prefect/ venv/lib/python3 11/site-packages/prefect/client/orchestration/base py line 53 in request return await self _client send(request) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File /opt/prefect/ venv/lib/python3 11/site-packages/prefect/client/base py line 379 in send response raise_for_status() File /opt/prefect/ venv/lib/python3 11/site-packages/prefect/client/base py line 163 in raise_for_status raise PrefectHTTPStatusError from_httpx_error(exc) from exc cause prefect exceptions PrefectHTTPStatusError Client error '404 Not Found' for url '<http //prefect-server 4200/api/block_types/slug/postgres-prod-dbt-creds/block_documents/name/database?include_secrets=true>' Response {'detail' 'Block document not found'} For more information check <https //developer mozilla org/en-US/docs/Web/HTTP/Status/404> The above exception was the direct cause of the following exception Traceback (most recent call last) File /opt/prefect/ venv/lib/python3 11/site-packages/prefect/task_engine py line 884 in run_context yield self File /opt/prefect/ venv/lib/python3 11/site-packages/prefect/task_engine py line 1538 in run_task_sync engine call_task_fn(txn) File /opt/prefect/ venv/lib/python3 11/site-packages/prefect/task_engine py line 901 in call_task_fn result = call_with_parameters(self task fn parameters) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File /opt/prefect/ venv/lib/python3 11/site-packages/prefect/utilities/callables py line 210 in call_with_parameters return fn(*args *kwargs) ^^^^^^^^^^^^^^^^^^^ File /opt/prefect/flows/flow py line 94 in run_dbt_job runner invoke([ build ]) File /opt/prefect/ venv/lib/python3 11/site-packages/prefect_dbt/core/runner py line 1066 in invoke with self settings resolve_profiles_yml() as profiles_dir File /usr/local/lib/python3 11/contextlib py line 137 in enter return next(self gen) ^^^^^^^^^^^^^^ File /opt/prefect/ venv/lib/python3 11/site-packages/prefect_dbt/core/settings py line 94 in resolve_profiles_yml profiles_yml = run_coro_as_sync( ^^^^^^^^^^^^^^^^^ File /opt/prefect/ venv/lib/python3 11/site-packages/prefect/utilities/asyncutils py line 207 in run_coro_as_sync return call result() ^^^^^^^^^^^^^ File /opt/prefect/ venv/lib/python3 11/site-packages/prefect/_internal/concurrency/calls py line 365 in result return self future result(timeout=timeout) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File /opt/prefect/ venv/lib/python3 11/site-packages/prefect/_internal/concurrency/calls py line 192 in result return self __get_result() ^^^^^^^^^^^^^^^^^^^ File /usr/local/lib/python3 11/concurrent/futures/_b
ase py line 401 in __get_result raise self _exception File /opt/prefect/ venv/lib/python3 11/site-packages/prefect/_internal/concurrency/calls py line 441 in _run_async result = await coro ^^^^^^^^^^ File /opt/prefect/ venv/lib/python3 11/site-packages/prefect/utilities/asyncutils py line 188 in coroutine_wrapper return await task ^^^^^^^^^^ File /opt/prefect/ venv/lib/python3 11/site-packages/prefect/client/utilities py line 99 in with_injected_client return await fn(*args *kwargs) ^^^^^^^^^^^^^^^^^^^^^^^^^ File /opt/prefect/ venv/lib/python3 11/site-packages/prefect/utilities/templating py line 317 in resolve_block_document_references updated_value = await resolve_block_document_references( ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File /opt/prefect/ venv/lib/python3 11/site-packages/prefect/client/utilities py line 99 in with_injected_client return await fn(*args *kwargs) ^^^^^^^^^^^^^^^^^^^^^^^^^ File /opt/prefect/ venv/lib/python3 11/site-packages/prefect/utilities/templating py line 317 in resolve_block_document_references updated_value = await resolve_block_document_references( ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File /opt/prefect/ venv/lib/python3 11/site-packages/prefect/client/utilities py line 99 in with_injected_client return await fn(*args *kwargs) ^^^^^^^^^^^^^^^^^^^^^^^^^ File /opt/prefect/ venv/lib/python3 11/site-packages/prefect/utilities/templating py line 317 in resolve_block_document_references updated_value = await resolve_block_document_references( ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File /opt/prefect/ venv/lib/python3 11/site-packages/prefect/client/utilities py line 99 in with_injected_client return await fn(*args **kwargs) ^^^^^^^^^^^^^^^^^^^^^^^^^
p
@Marvin dev: dbname: "{{ prefect.dbt-cli-postgres-target-configs.blocks.postgres-dev-dbt-creds.credentials.connection_info.database }}" host: "{{ prefect.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.credentials.connection_info.host }}" pass: "{{ prefect.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.credentials.connection_info.password }}" port: "{{ prefect.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.credentials.connection_info.port }}" schema: public threads: 4 type: postgres user: "{{ prefect.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.credentials.connection_info.username }}" Task run failed with exception: ObjectNotFound(None) Traceback (most recent call last): File "/opt/prefect/.venv/lib/python3.11/site-＼
m
✅ thought for 68.5 seconds
Thanks — the 
ObjectNotFound
is happening because the Jinja path is wrong. A few concrete fixes: - The prefix must be 
prefect.blocks
, not 
prefect.dbt-cli-postgres-target-configs
or 
prefect.blocks.blocks
. - Don’t include a nested block name inside the path (e.g., 
credentials.postgres-dev-conn...
). - Use dbt’s 
password
(not 
pass
). - SQLAlchemy connector fields live under 
credentials.connection_info
. Assuming your block type is “Dbt CLI Postgres Target Configs” and the block name is 
postgres-dev-dbt-creds
, use this:
Copy code
dev:
  type: postgres
  host: "{{ prefect.blocks.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.credentials.connection_info.host }}"
  port: "{{ prefect.blocks.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.credentials.connection_info.port }}"
  user: "{{ prefect.blocks.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.credentials.connection_info.username }}"
  password: "{{ prefect.blocks.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.credentials.connection_info.password }}"
  dbname: "{{ prefect.blocks.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.credentials.connection_info.database }}"
  schema: public
  threads: 4
Important: - Those 
{{ prefect.blocks... }}
placeholders are only resolved when you run dbt via Prefect’s dbt integration (e.g., using 
PrefectDbtSettings.resolve_profiles_yml()
or 
trigger_dbt_cli_command
). Running 
dbt
directly won’t resolve them. If you still see 
ObjectNotFound
, double‑check: - The exact block type slug is 
dbt-cli-postgres-target-configs
- The exact block name is 
postgres-dev-dbt-creds
- You’re running through Prefect’s resolver Quick debug to see what Prefect resolves:
Copy code
from pathlib import Path
from prefect_dbt.core.settings import PrefectDbtSettings

settings = PrefectDbtSettings(profiles_dir="/path/to/your/profiles_dir")
with settings.resolve_profiles_yml() as tmp:
    print(Path(tmp, "profiles.yml").read_text())
If this doesn’t fix it, please share: - The exact block name and type from the Prefect UI - How you’re running dbt (Prefect task/flow or calling dbt directly) - The full error message (a few more lines after 
ObjectNotFound(None)
help)
p
@Marvin dev: dbname: "{{ prefect.blocks.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.credentials.connection_info.database }}" host: "{{ prefect.blocks.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.credentials.connection_info.host }}" pass: "{{ prefect.blocks.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.credentials.connection_info.password }}" port: "{{ prefect.blocks.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.credentials.connection_info.port }}" schema: public threads: 4 type: postgres user: "{{ prefect.blocks.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.credentials.connection_info.username }}"
m
✅ thought for 290.7 seconds
You’re very close — two things to fix and a couple of checks to stop the ObjectNotFound: 1) Use dbt’s field name password (not pass) 2) Keep the prefix exactly prefect.blocks.<block-type-slug>.<block-name>... Corrected snippet (assuming the block name is exactly postgres-dev-dbt-creds):
Copy code
dev:
  type: postgres
  host: "{{ prefect.blocks.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.credentials.connection_info.host }}"
  port: "{{ prefect.blocks.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.credentials.connection_info.port }}"
  user: "{{ prefect.blocks.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.credentials.connection_info.username }}"
  password: "{{ prefect.blocks.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.credentials.connection_info.password }}"
  dbname: "{{ prefect.blocks.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.credentials.connection_info.database }}"
  schema: public
  threads: 4
If you still get ObjectNotFound: - Verify the exact block slug. Run:
Copy code
prefect block ls
or:
Copy code
prefect block inspect dbt-cli-postgres-target-configs/postgres-dev-dbt-creds
If inspect fails, the type slug or block name is off. - If the type doesn’t exist yet, register prefect-dbt blocks:
Copy code
prefect block register -m prefect_dbt
- Make sure you’re running dbt through Prefect’s resolver (plain dbt will not resolve 
{{ prefect.blocks... }}
):
Copy code
from pathlib import Path
  from prefect_dbt.core.settings import PrefectDbtSettings

  settings = PrefectDbtSettings(profiles_dir="/path/to/profiles_dir")
  with settings.resolve_profiles_yml() as tmp:
      print(Path(tmp, "profiles.yml").read_text())
This will show you exactly what Prefect resolved. Also double‑check your profiles.yml structure matches dbt’s expected shape:
Copy code
your_profile_name:
  target: dev
  outputs:
    dev:
      ...the mapping above...
If it still errors, please share: - The exact output of 
prefect block inspect dbt-cli-postgres-target-configs/postgres-dev-dbt-creds
- How you’re invoking dbt (Prefect flow/task vs direct dbt) - The next lines of the stack trace after ObjectNotFound
p
@Marvin nyc_yellow_taxi_trips: target: "{{ prefect.variables.target }}" outputs: # ---------------------------------------------------- # Dev 環境配置 # ---------------------------------------------------- dev: dbname: "{{ prefect.blocks.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.credentials.connection_info.database }}" host: "{{ prefect.blocks.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.credentials.connection_info.host }}" password: "{{ prefect.blocks.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.credentials.connection_info.password }}" port: "{{ prefect.blocks.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.credentials.connection_info.port }}" schema: public threads: 4 type: postgres user: "{{ prefect.blocks.dbt-cli-postgres-target-configs.postgres-dev-dbt-creds.credentials.connection_info.username }}" still fail
m
✅ thought for 101.4 seconds
Got it — since it’s “still failing,” let’s quickly isolate where it’s breaking and get you unblocked. First, your profiles.yml keys now look correct. The most common remaining causes are: - The block slug or name doesn’t exist (or wrong workspace) - The Prefect variable 
target
isn’t set - You’re running dbt directly so the 
{{ prefect.blocks... }}
placeholders never get resolved - The resolver isn’t actually reading the profiles.yml you edited Do these quick checks: 1) Verify the block exists and the slug is correct
Copy code
prefect block inspect dbt-cli-postgres-target-configs/postgres-dev-dbt-creds
- If this errors, the type slug or block name is wrong, or you’re in a different workspace. Also confirm you’re authenticated (API URL/KEY). 2) Verify the variable exists
Copy code
prefect variable inspect target
# or set it if missing
prefect variable set target dev
3) Ensure you’re resolving the profiles.yml through Prefect’s dbt integration Plain 
dbt
will not resolve 
{{ prefect.blocks... }}
or 
{{ prefect.variables... }}
. Use the resolver and confirm which file it’s reading:
Copy code
from pathlib import Path
from prefect_dbt.core.settings import PrefectDbtSettings

# Point this to the directory that contains the profiles.yml you pasted
settings = PrefectDbtSettings(profiles_dir="/path/to/.dbt")

with settings.resolve_profiles_yml() as tmp:
    print("Resolved profiles.yml:")
    print(Path(tmp, "profiles.yml").read_text())
If this prints resolved values (not the 
{{ prefect... }}
placeholders), you can run dbt with the resolved dir:
Copy code
from prefect_dbt.core.runner import PrefectDbtRunner

with settings.resolve_profiles_yml() as tmp:
    runner = PrefectDbtRunner(settings=PrefectDbtSettings(profiles_dir=tmp))
    runner.invoke(["run"])
4) As an alternative, avoid templating entirely and let Prefect generate profiles.yml from your saved block This bypasses profiles.yml placeholders and eliminates slug/templating issues.
Copy code
from prefect_dbt.cli.credentials import DbtCliProfile
from prefect_dbt.cli.configs import PostgresTargetConfigs
from prefect_dbt.cli.commands import trigger_dbt_cli_command

# Load your saved target-configs block
pg_cfg = PostgresTargetConfigs.load("postgres-dev-dbt-creds")

profile = DbtCliProfile(
    name="nyc_yellow_taxi_trips",
    target="dev",  # or load from Prefect Variable if you want
    target_configs=pg_cfg,
)

# This writes a fully-resolved profiles.yml and runs dbt
trigger_dbt_cli_command(
    command="dbt run",
    dbt_cli_profile=profile,
    overwrite_profiles=True,         # let Prefect write the profiles.yml
    project_dir="/path/to/your/dbt", # adjust
)
If it still fails, please share: - The output of: - 
prefect block inspect dbt-cli-postgres-target-configs/postgres-dev-dbt-creds
- 
prefect variable inspect target
- How you are invoking dbt (Prefect flow/task vs calling dbt directly) - The next few lines of the error/trace after “Task run failed with exception: ObjectNotFound(None)” With that, we can pinpoint the exact issue quickly.
p
@Marvin how about i connect directly from SQLAlchemy Connector block
m
✅ thought for 29.4 seconds
Yes — you can reference a SQLAlchemy Connector block directly in profiles.yml. Use the block type slug sqlalchemy-connector and read from its connection_info fields. Assuming your connector block is named postgres-dev-conn, here’s the mapping:
Copy code
dev:
  type: postgres
  host: "{{ prefect.blocks.sqlalchemy-connector.postgres-dev-conn.connection_info.host }}"
  port: "{{ prefect.blocks.sqlalchemy-connector.postgres-dev-conn.connection_info.port }}"
  user: "{{ prefect.blocks.sqlalchemy-connector.postgres-dev-conn.connection_info.username }}"
  password: "{{ prefect.blocks.sqlalchemy-connector.postgres-dev-conn.connection_info.password }}"
  dbname: "{{ prefect.blocks.sqlalchemy-connector.postgres-dev-conn.connection_info.database }}"
  schema: public
  threads: 4
Important notes: - The prefix must be 
prefect.blocks
. - The type slug is 
sqlalchemy-connector
. - These placeholders are only resolved when you run dbt through Prefect’s dbt integration (e.g., with 
PrefectDbtSettings.resolve_profiles_yml()
or 
trigger_dbt_cli_command
). Plain 
dbt
will not resolve them. Quick checks if it still fails: - Verify the block exists and the name/slug are correct:
Copy code
prefect block inspect sqlalchemy-connector/postgres-dev-conn
- Print the resolved profiles.yml to confirm placeholders are being expanded:
Copy code
from pathlib import Path
  from prefect_dbt.core.settings import PrefectDbtSettings

  settings = PrefectDbtSettings(profiles_dir="/path/to/.dbt")
  with settings.resolve_profiles_yml() as tmp:
      print(Path(tmp, "profiles.yml").read_text())
Optional (avoids templating entirely): build the profile programmatically from the connector
Copy code
from prefect_dbt.cli.credentials import DbtCliProfile
from prefect_dbt.cli.configs import PostgresTargetConfigs
from prefect_sqlalchemy.database import SqlAlchemyConnector

conn = SqlAlchemyConnector.load("postgres-dev-conn")
pg_cfg = PostgresTargetConfigs(schema="public", threads=4, credentials=conn)

profile = DbtCliProfile(name="nyc_yellow_taxi_trips", target="dev", target_configs=pg_cfg)

from prefect_dbt.cli.commands import trigger_dbt_cli_command
trigger_dbt_cli_command(
  command="dbt run",
  dbt_cli_profile=profile,
  overwrite_profiles=True,
  project_dir="/path/to/dbt"
)
If you share your exact connector block name and how you’re invoking dbt, I can tailor the snippet further.
29 Views
Open in Slack
PreviousNext
Powered by