DBT-CORE-INTERFACE

Lightweight, thread-safe, multi-project Python interface to dbt-core

Built with the tools and technologies:

---

Overview

dbt-core-interface is a lightweight, high-performance Python interface for working directly with dbt-core (v1.8+). It allows developers to manage and run dbt projects entirely in memory using an intuitive Python API—enabling runtime SQL compilation, macro evaluation, SQLFluff linting/formatting, and more, all through FastAPI or local usage.

It supports dynamic multi-project environments, automatic re-parsing, file watchers, and asynchronous usage. It is the foundation for more complex interfaces such as dbt-fastapi and is designed to rapidly prototype ideas outside the constraints of the dbt-core repo itself.

---

Features

• 🧐 In-memory dbt-core 1.8+ interface with full RuntimeConfig hydration
• ⚡ Fast, thread-safe SQL compilation and execution via FastAPI
• 🔬 Interactive linting and formatting with SQLFluff
• 🌐 Live REST API server via FastAPI
• 🌍 Supports multiple projects simultaneously using DbtProjectContainer
• 🚀 Dynamic macro parsing, Jinja rendering, manifest manipulation
• 🔄 Background file watching for auto-reparsing
• ⚖ Direct dbt command passthrough (e.g. run, test, docs serve, etc.)

---

Requirements

• Python 3.9+
• dbt-core >= 1.8.0

Install via PyPI:

pip install dbt-core-interface

---

Usage

Programmatic

from dbt_core_interface import DbtProject

# Load your project
project = DbtProject(project_dir="/path/to/dbt_project")

# Run a simple SQL query
res = project.execute_sql("SELECT current_date AS today")
print(res.table)

# Compile SQL (but don't run it)
compiled = project.compile_sql("SELECT * FROM {{ ref('my_model') }}")
print(compiled.compiled_code)

# Execute a ref() lookup
node = project.ref("my_model")
print(node.resource_type, node.name)

# Load a source node
source = project.source("my_source", "my_table")
print(source.description)

# Incrementally parse the project
project.parse_project(write_manifest=True)

# Re-parse a specific path
project.parse_paths("models/my_model.sql")

# Compile a node from path
node = project.get_node_by_path("models/my_model.sql")
compiled = project.compile_node(node)
print(compiled.compiled_code)

# Run a dbt command programmatically
project.run("-s +orders")
project.test()

# SQLFluff linting
lint_result = project.lint(sql="select 1 AS foo")
lint_result = project.lint(sql=Path("models/my_model.sql"))
print(lint_result)

# SQLFluff formatting
success, formatted_sql = project.format(sql="Select * FROM orders as o")
success, formatted_sql = project.format(sql=Path("models/my_model.sql"))
print(formatted_sql)

# Use the DbtProjectContainer to manage multiple projects
from dbt_core_interface import DbtProjectContainer

container = DbtProjectContainer()
container.create_project(project_dir="/path/to/dbt_project_1")
container.create_project(project_dir="/path/to/dbt_project_2")
print(container.registered_projects())

Server Mode (FastAPI)

Run:

python -m dbt_core_interface.server --host 0.0.0.0 --port 8581

Register a project:

curl -X POST 'http://localhost:8581/register?project_dir=/your/dbt_project'

Compile SQL:

curl -X POST 'http://localhost:8581/compile' \
     -H 'X-dbt-Project: /your/dbt_project' \
     -d 'select * from {{ ref("orders" }}'

Client Usage

Run the server and use the bundled client to interact with it:

from dbt_core_interface.client import DbtInterfaceClient, ServerError

client = DbtInterfaceClient(
    project_dir="/path/to/project",
    profiles_dir="/path/to/profiles.yml",
    target="dev",
    base_url="http://localhost:8581",
    timeout=(5.0, 15.0)
)

# Health & heartbeat
print(client.health_check())  # {'status': 'ok', ...}
print(client.heartbeat())     # {'alive': True, 'uptime': ...}

# Run SQL with limit & path which allows resolving {{ this }}
result = client.run_sql("SELECT * FROM {{ this }} ORDER BY id", limit=500, path="models/my_model.sql")
print(result.table.rows)

# Compile without execution
comp = client.compile_sql("SELECT * FROM {{ ref('users') }}")
print(comp.compiled_code)

# Lint & format
lint = client.lint_sql(raw_sql="select * from {{ ref('users') }}")
print(lint.violations)
fmt = client.format_sql(raw_sql="select * from {{ ref('users') }}")
print(fmt.formatted_code)

# Arbitrary dbt command
docs = client.command("docs", "generate")
print(docs)

# On object deletion, project is unregistered automatically
del client

---

License

This project is licensed under the MIT License. See the LICENSE file for more info.

---

Acknowledgments

Thanks to the dbt-core maintainers and contributors whose work makes this project possible.