profiles.yml to authenticate dlt pipelines — without duplicating credentials or managing separate config files.
What the Script Expects from profiles.yml
The loader reads your active target directly fromdbt_project.yml and looks it up in profiles.yml. A typical BigQuery profile looks like this:
my_project:
target: prod
outputs:
dev:
type: bigquery
method: service-account
project: my-gcp-project
dataset: my_dataset
location: US
keyfile_json:
type: service_account
project_id: my-gcp-project
private_key_id: abc123
private_key: "-----BEGIN RSA PRIVATE KEY-----\n..."
client_email: my-sa@my-gcp-project.iam.gserviceaccount.com
target: in your profile — in this case prod.
ingestion target is optional. The get_credentials_by_environment() method uses an ingestion target as a fallback when your default target is prod, to avoid accidentally running pipelines against production. This is only relevant if you call that specific method — the rest of this guide uses get_active_credentials(), which works with any target name.1. Set Up the Credentials Loader
Create adlthub folder at the root of your dbt™ project and add the profile_connection_credentials.py script inside it.
your-dbt-project/
├── models/
├── dbt_project.yml
├── dlthub/
│ ├── profile_connection_credentials.py ← add this script here
│ └── my_pipeline.py
2. Use the Credentials in Your dlt Pipeline
In your dlt pipeline file (e.g.,my_pipeline.py), import the loader and pass the credentials to your BigQuery destination.
import dlt
from profile_connection_credentials import ProfileConnectionCredentialsLoader
# Load credentials from your active dbt™ profile
loader = ProfileConnectionCredentialsLoader()
creds = loader.get_active_credentials()
# Configure your dlt pipeline using the fetched credentials
pipeline = dlt.pipeline(
pipeline_name="my_pipeline",
destination=dlt.destinations.bigquery(
project_id=creds.get("project"),
credentials=creds.get("keyfile_json"),
location=creds.get("location", "US"),
),
dataset_name="my_dataset",
)
# Define and run your pipeline as normal
@dlt.resource
def my_data():
yield [{"id": 1, "value": "hello"}, {"id": 2, "value": "world"}]
load_info = pipeline.run(my_data())
print(load_info)
NoteThe loader automatically detects your
dbt_project.yml and profiles.yml paths, as long as profile_connection_credentials.py is placed inside a dlthub folder at the root of your dbt™ project as shown above. If you place it elsewhere, pass project_dir and profiles_dir explicitly to ProfileConnectionCredentialsLoader()Appendix: Profile Connection Credentials Loader Script
Add the following script asprofile_connection_credentials.py in your dlthub folder.
#!/usr/bin/env python3
"""
Profile Connection Credentials Loader Module
A reusable utility module for fetching connection credentials from dbt's profiles.yml
on demand. Can be imported and used in different Python scripts.
This module directly parses profiles.yml without requiring dbt-core.
Usage:
from profile_connection_credentials import ProfileConnectionCredentialsLoader
loader = ProfileConnectionCredentialsLoader(
project_dir="/path/to/dbt/project",
profiles_dir="/path/to/profiles/dir"
)
# Get credentials for a specific profile and target
creds = loader.get_credentials(profile_name="my_profile", target_name="dev")
# Or get the active target credentials (from dbt_project.yml)
creds = loader.get_active_credentials()
# Access credential attributes
print(creds['type']) # e.g., "bigquery"
print(creds['project_id']) # database/project specific
"""
import sys
from pathlib import Path
from typing import Optional, Dict, Any
try:
import yaml
except ImportError:
print("Error: PyYAML is not installed")
print("Install it with: pip install pyyaml")
sys.exit(1)
class ProfileConnectionCredentialsLoader:
"""
Load and manage connection credentials from profiles.yml by directly parsing YAML.
This class provides a simple interface to fetch credentials on demand
without requiring dbt-core to be installed.
"""
def __init__(
self,
project_dir: Optional[str] = None,
profiles_dir: Optional[str] = None
):
"""
Initialize the credentials loader.
Args:
project_dir: Path to the dbt project directory (auto-detected if not provided)
profiles_dir: Path to the directory containing profiles.yml
(auto-detected if not provided)
Raises:
FileNotFoundError: If project_dir or profiles_dir don't exist
FileNotFoundError: If dbt_project.yml or profiles.yml not found
"""
# Auto-detect paths if not provided
if not project_dir:
project_dir = self._detect_project_dir()
if not profiles_dir:
profiles_dir = self._detect_profiles_dir(project_dir)
self.project_dir = project_dir
self.profiles_dir = profiles_dir
self._profiles_data = None
self._project_data = None
# Validate directories and files
self._validate_paths()
@staticmethod
def _detect_project_dir() -> str:
"""
Auto-detect the dbt project directory.
Looks for dbt_project.yml relative to this script's location.
Falls back to /workspace/repository/dbt if not found.
Returns:
Path to the dbt project directory
"""
# Try to find dbt_project.yml relative to this script
# This file is at: <repo>/dbt/dlthub/profile_connection_credentials.py
# So we go up 2 levels to get to <repo>/dbt
script_file = Path(__file__).resolve()
script_dir = script_file.parent.parent # Go up to dbt project root
if (script_dir / "dbt_project.yml").exists():
return str(script_dir.resolve())
# Fall back to default location
return "/workspace/repository/dbt"
@staticmethod
def _detect_profiles_dir(project_dir: str) -> str:
"""
Auto-detect the profiles directory.
Looks for profiles.yml in the parent directory of the project.
Falls back to /workspace if not found.
Args:
project_dir: Path to the dbt project directory
Returns:
Path to the profiles directory
"""
# Try to find profiles.yml in the parent directory of project
# project_dir is typically <repo>/dbt, so parent is <repo>
project_path = Path(project_dir).resolve()
potential_profiles_dir = project_path.parent
if (potential_profiles_dir / "profiles.yml").exists():
return str(potential_profiles_dir.resolve())
# Fall back to default location
return "/workspace"
def _validate_paths(self) -> None:
"""Validate that required directories and files exist."""
project_path = Path(self.project_dir)
if not project_path.exists():
raise FileNotFoundError(f"Project directory does not exist: {self.project_dir}")
dbt_project_yml = project_path / 'dbt_project.yml'
if not dbt_project_yml.exists():
raise FileNotFoundError(f"dbt_project.yml not found in: {self.project_dir}")
profiles_path = Path(self.profiles_dir)
if not profiles_path.exists():
raise FileNotFoundError(f"Profiles directory does not exist: {self.profiles_dir}")
profiles_yml = profiles_path / 'profiles.yml'
if not profiles_yml.exists():
raise FileNotFoundError(f"profiles.yml not found in: {self.profiles_dir}")
def _load_profiles_yaml(self) -> Dict[str, Any]:
"""
Load and parse the profiles.yml file.
Returns:
Dictionary containing the parsed profiles.yml content
"""
if self._profiles_data is not None:
return self._profiles_data
try:
profiles_path = Path(self.profiles_dir) / 'profiles.yml'
with open(profiles_path, 'r') as f:
self._profiles_data = yaml.safe_load(f) or {}
return self._profiles_data
except Exception as e:
raise RuntimeError(f"Failed to load profiles.yml: {e}")
def _load_project_yaml(self) -> Dict[str, Any]:
"""
Load and parse the dbt_project.yml file.
Returns:
Dictionary containing the parsed dbt_project.yml content
"""
if self._project_data is not None:
return self._project_data
try:
project_path = Path(self.project_dir) / 'dbt_project.yml'
with open(project_path, 'r') as f:
self._project_data = yaml.safe_load(f) or {}
return self._project_data
except Exception as e:
raise RuntimeError(f"Failed to load dbt_project.yml: {e}")
def get_active_credentials(self) -> Dict[str, Any]:
"""
Get the credentials for the active target (from dbt_project.yml).
Returns:
Dictionary of credentials for the active target
Raises:
RuntimeError: If credentials cannot be loaded
"""
project_data = self._load_project_yaml()
profiles_data = self._load_profiles_yaml()
# Get profile name from dbt_project.yml
profile_name = project_data.get('profile')
if not profile_name:
raise RuntimeError("No 'profile' specified in dbt_project.yml")
# Get the profile from profiles.yml
if profile_name not in profiles_data:
raise RuntimeError(f"Profile '{profile_name}' not found in profiles.yml")
profile = profiles_data[profile_name]
# Get the target name (default or specified)
target_name = profile.get('target')
if not target_name:
raise RuntimeError(f"No 'target' specified for profile '{profile_name}'")
# Get the target credentials
outputs = profile.get('outputs', {})
if target_name not in outputs:
raise RuntimeError(
f"Target '{target_name}' not found in profile '{profile_name}'"
)
credentials = outputs[target_name]
return credentials
def get_credentials(
self,
profile_name: Optional[str] = None,
target_name: Optional[str] = None
) -> Dict[str, Any]:
"""
Get credentials for a specific profile and target.
If profile_name or target_name are not provided, uses the active
target from dbt_project.yml.
Args:
profile_name: Name of the profile (optional)
target_name: Name of the target within the profile (optional)
Returns:
Dictionary of credentials for the specified target
Raises:
RuntimeError: If credentials cannot be loaded
"""
profiles_data = self._load_profiles_yaml()
# If no specific profile/target requested, use active credentials
if not profile_name and not target_name:
return self.get_active_credentials()
# Use provided profile_name or get from dbt_project.yml
if not profile_name:
project_data = self._load_project_yaml()
profile_name = project_data.get('profile')
if not profile_name:
raise RuntimeError("No 'profile' specified in dbt_project.yml")
# Get the profile from profiles.yml
if profile_name not in profiles_data:
raise RuntimeError(f"Profile '{profile_name}' not found in profiles.yml")
profile = profiles_data[profile_name]
# Use provided target_name or get default from profile
if not target_name:
target_name = profile.get('target')
if not target_name:
raise RuntimeError(f"No 'target' specified for profile '{profile_name}'")
# Get the target credentials
outputs = profile.get('outputs', {})
if target_name not in outputs:
raise RuntimeError(
f"Target '{target_name}' not found in profile '{profile_name}'"
)
credentials = outputs[target_name]
return credentials
def get_credentials_dict(self) -> Dict[str, Any]:
"""
Get the active credentials as a dictionary.
Returns:
Dictionary of credential attributes
Raises:
RuntimeError: If credentials cannot be loaded
"""
credentials = self.get_active_credentials()
# Filter out private attributes (starting with _)
creds_dict = {
key: value for key, value in credentials.items()
if not key.startswith('_')
}
return creds_dict
def get_profile_info(self) -> Dict[str, Any]:
"""
Get information about the loaded profile.
Returns:
Dictionary with profile metadata
"""
project_data = self._load_project_yaml()
profiles_data = self._load_profiles_yaml()
profile_name = project_data.get('profile')
profile = profiles_data.get(profile_name, {})
target_name = profile.get('target')
return {
'project_name': project_data.get('name'),
'project_root': str(Path(self.project_dir).resolve()),
'profiles_dir': str(Path(self.profiles_dir).resolve()),
'profile_name': profile_name,
'target_name': target_name,
}
def get_credential_attribute(self, attribute_name: str) -> Any:
"""
Get a specific attribute from the active credentials.
Args:
attribute_name: Name of the credential attribute to retrieve
Returns:
The value of the requested attribute
Raises:
KeyError: If the attribute doesn't exist
RuntimeError: If credentials cannot be loaded
"""
credentials = self.get_active_credentials()
if attribute_name not in credentials:
raise KeyError(
f"Credentials dictionary has no key '{attribute_name}'"
)
return credentials[attribute_name]
def get_credentials_by_environment(self, non_prod_target: str = 'ingestion') -> Dict[str, Any]:
"""
Get credentials based on the environment (target).
If the default target is 'prod', uses the non-prod target instead.
Otherwise, uses the default target.
Args:
non_prod_target: Target name to use if default is 'prod' (default: 'ingestion')
Returns:
Dictionary of credentials for the selected target
Raises:
RuntimeError: If credentials cannot be loaded
"""
profile_info = self.get_profile_info()
default_target = profile_info.get('target_name')
# If default target is 'prod', use non-prod target instead
if default_target == 'prod':
return self.get_credentials(target_name=non_prod_target)
else:
# Use the default target
return self.get_active_credentials()
# Example usage and testing
if __name__ == "__main__":
import json
if len(sys.argv) < 2:
print("Usage:")
print(" python profile_connection_credentials.py <project_dir> [profiles_dir]")
print("\nExample:")
print(" python profile_connection_credentials.py /workspace/repository/dbt /workspace")
sys.exit(1)
project_dir = sys.argv[1]
profiles_dir = sys.argv[2] if len(sys.argv) > 2 else "/workspace"
try:
# Initialize the loader
loader = ProfileConnectionCredentialsLoader(project_dir, profiles_dir)
# Get profile info
print("\n" + "=" * 70)
print("PROFILE INFORMATION")
print("=" * 70)
profile_info = loader.get_profile_info()
for key, value in profile_info.items():
print(f"{key}: {value}")
# Get active credentials
print("\n" + "=" * 70)
print("ACTIVE TARGET CREDENTIALS")
print("=" * 70)
creds = loader.get_active_credentials()
print(f"Type: {creds.get('type', 'N/A')}")
# Print all credential attributes
creds_dict = loader.get_credentials_dict()
for key, value in creds_dict.items():
# Redact sensitive information
if key in ['password', 'token', 'api_key', 'secret', 'private_key']:
value = "[REDACTED]"
print(f"{key}: {value}")
print("\n" + "=" * 70 + "\n")
except Exception as e:
print(f"❌ Error: {e}")
import traceback
traceback.print_exc()
sys.exit(1)