[Feature] - Provide User Defined Macros & Filters

[afranzi]

This new discussion aims to gather feedback from the community and the DBT team to enable the DBT developers to define their own Jinja Python Macros & Filters being able to integrate them into the code and enhancing their experience with new potential capabilities.

Proposal

Similar to how Airflow handles the capability to allow the Airflow developers to integrate new Macros and Filters into the variables and connections.

The main idea would be to define two new Env vars to import the packages.

• DBT_USER_DEFINED_MACROS: Jinja Macros
• DBT_USER_DEFINED_FILTERS: Jinja Filters

See the documentation proposal at dbt-labs/docs.getdbt.com#4872 and the required implementation at dbt-labs/dbt-common#72

Usages

Some potential use-cases we would enable could be:

• Credentials retrieval from external systems (i.e. AWS Secrets Manager / K8s Secrets).
• External Model Config (retrieve model configurations) from 3rd party APIs.
• Others

Example

export DBT_USER_DEFINED_MACROS=afranzi.macros:dbt_macros

## FILE: afranzi/macros.py

def universe_number() -> int:
    return 42

def secrets_client(region_name: str | None = None) -> SecretsManagerClient:
    return aws_session(region_name).client(service_name="secretsmanager")

def get_secret(secret_name: str) -> dict:
    client = secrets_client()

    try:
        get_secret_value_response = client.get_secret_value(SecretId=secret_name)
    except ClientError as e:
        raise e

    secret_str = get_secret_value_response["SecretString"]
    return json.loads(secret_str)

@lru_cache
def fetch_db_credentials() -> dict:
    db_credentials_secret = get_secret('my_db_credentials')
    return db_credentials_secret

def credentials(field: str) -> dict:
    return fetch_db_credentials()[field]


dbt_macros = {
    "today": dates.today,
    "universe_number": universe_number,
    "credentials": credentials,
}

dummy_gummy:
  target: dev
  outputs:
    dev:
      type: postgres
      host: "{{ credentials('host') }}"
      user: "{{ credentials('user') }}"
      password: "{{ credentials('password') }}"
      port: "{{ env_var('DBT_POSTGRES_TEST_PORT') | int }}" # RANDOM PYTEST PORT
      dbname: "{{ credentials('database') }}"
      schema: "{{ credentials('schema') }}"
      threads: 2

[ismailsimsek]

+1 for this feature

@afranzi and whoever interested in this feature. opendbt is already enabling this and much more features in an elegant way.

leaving here links extending adapters with new python methods exposed to jinja integrating dlt adapters examples
hope it helps