🛠️ Designing an Entity‑First YAML → SQL Generator (dbt & SQLMesh)

The hands‑on guide with schema, code skeletons, and CI/CD.

Series navigation: A1 — Why We Need a Business‑Friendly Data Transformation Model • B1 — How dbt & SQLMesh Organize Metadata and Code • B2 (you are here) • A2 — Entities, Subject Areas & Transformation Steps

Summary

We define a small, strict Entity‑First metadata schema (YAML) for entities, roles, patterns, relationships, and step types.

A lightweight Python generator validates YAML, builds a DAG, and emits dbt and SQLMesh models/tests/macros, plus profiling/DQ queries and docs diagrams.

Everything runs in Git + VSCode + CI and remains platform‑agnostic while allowing dialect‑specific optimizations via macros.

1) Goals & design principles

Single source of truth: logical entities, not files or tables, are the primary modeling unit.
Separation of concerns: logical modeling in YAML; physical materialization and platform details in emitters/macros.
Deterministic generation: same YAML ⇒ same SQL (idempotent builds, snapshot tests).
Portable: emitters for dbt and SQLMesh; add others later with the same AST.
Validatable: JSON Schema + semantic validators (graph, keys, relationships).
Extensible: step types and macros for complex patterns (incremental, SCD2, windowed agg, dedupe, translate).

2) Entity YAML v0.1 (spec & examples)

Each entity lives at entities/<subject_area>/<entity>/<entity>.entity.yml.

2.1 Minimal fields

entity: order_enriched               # logical name (unique within subject area)  
subject_area: sales                  # business grouping  
role: intermediate                   # source | intermediate | delivery  
pattern: core                        # core | characteristic | associative | reference  
primary_key: order_id                # PK at the entity’s grain

structure:  
  columns:  
    order_id:        {type: integer, description: "Surrogate key"}  
    customer_id:     {type: integer}  
    country_code:    {type: string}  
    status_code:     {type: string}  
    status_label:    {type: string}  
    order_date:      {type: date}  
    ship_date:       {type: date}  
    amount:          {type: numeric(18,2)}step:  
  type: derive                         # derive | filter | aggregate | union | translate | windowed_agg | deduplicateinputs:  
  - entity: sales.order  
  - entity: sales.customer  
  - entity: shared.code_translationrelationships:  
  - name: order_to_customer  
    left:  sales.order.customer_id  
    right: sales.customer.customer_id  
    join_type: inner  
  - name: order_status_translate  
    left:  sales.order.status_code  
    right: shared.code_translation.code  
    join_type: left  
    filter: shared.code_translation.domain = 'order_status'mappings:  
  order_id:      sales.order.order_id  
  customer_id:   sales.order.customer_id  
  country_code:  sales.customer.country_code  
  status_code:   sales.order.status_code  
  status_label:  fn.translate_code('order_status', sales.order.status_code)  
  order_date:    sales.order.order_date  
  ship_date:     sales.order.ship_date  
  amount:        sales.order.amount  
  days_to_ship:  date_diff('day', sales.order.order_date, sales.order.ship_date)constraints:  
  not_null: [order_id]  
  unique:   [order_id]  
  foreign_keys:  
    - {column: customer_id, ref: sales.customer.customer_id}

2.2 Step‑type specifics

filter

step:  
  type: filter  
  predicate: "status_code != 'X'"

aggregate

step:  
  type: aggregate  
  inputs:  
    - entity: sales.order_enriched  
  group_by: [order_date, country_code]  
  aggregations:  
    orders_cnt: count(1)  
    amount_sum: sum(amount)  
  post_derivations:  
    pk_day_country: concat(order_date, '_', country_code)  
primary_key: pk_day_country

union

step:  
  type: union  
  inputs:  
    - entity: source.sales.order_region_a  
    - entity: source.sales.order_region_b  
# Generator verifies aligned structure

windowed_agg (optional extension)

step:  
  type: windowed_agg  
  partition_by: [customer_id]  
  order_by: [{column: order_date, direction: asc}]  
  windows:  
    order_running_amount: {expr: sum(amount), frame: rows between unbounded preceding and current row}

deduplicate (optional extension)

step:  
  type: deduplicate  
  key: [business_id]  
  order_by: [{column: updated_at, direction: desc}]

3) JSON Schema (excerpt)

{  
  "$schema": "https://json-schema.org/draft/2020-12/schema",  
  "type": "object",  
  "required": ["entity", "subject_area", "role", "pattern", "primary_key", "structure", "step"],  
  "properties": {  
    "entity": {"type": "string"},  
    "subject_area": {"type": "string"},  
    "role": {"enum": ["source", "intermediate", "delivery"]},  
    "pattern": {"enum": ["core", "characteristic", "associative", "reference"]},  
    "primary_key": {"type": "string"},  
    "structure": {"type": "object"},  
    "step": {"type": "object", "properties": {"type": {"type": "string"}}},  
    "inputs": {"type": "array"},  
    "relationships": {"type": "array"},  
    "mappings": {"type": "object"},  
    "constraints": {"type": "object"}  
  }  
}

Add semantic validators (Python) for: PK existence in columns, FK targets, DAG acyclicity, structure alignment for unions, aggregate PK grain consistency, etc.

4) Generator architecture

entities/ → loader → validator → AST(graph) → emitters → generated/

Loader: discover *.entity.yml, parse YAML, enforce JSON Schema.
Validator: semantic checks (keys, relationships, steps, graph rules).
AST: normalized in‑memory model (Entities, Columns, Inputs, Relationships, Steps).
Emitters: dbt_emitter, sqlmesh_emitter, testing_emitter, docs_emitter.
Outputs: SQL files, YAML tests, sources, macros; profiling/DQ queries; diagrams.

4.1 Data classes (Python)

from dataclasses import dataclass, field  
from typing import List, Dict, Optional  

@dataclass  
class Column:  
    name: str  
    type: Optional[str] = None  
    expr: Optional[str] = None  
@dataclass  
class Relationship:  
    name: str  
    left: str  
    right: str  
    join_type: str = "inner"  
    filter: Optional[str] = None  
@dataclass  
class Step:  
    type: str  
    predicate: Optional[str] = None  
    group_by: List[str] = field(default_factory=list)  
    aggregations: Dict[str, str] = field(default_factory=dict)  
    post_derivations: Dict[str, str] = field(default_factory=dict)  
    partition_by: List[str] = field(default_factory=list)  
    order_by: List[Dict[str, str]] = field(default_factory=list)  
@dataclass  
class Entity:  
    fqname: str              # e.g. "sales.order_enriched"  
    role: str                # source|intermediate|delivery  
    pattern: str             # core|characteristic|associative|reference  
    primary_key: str  
    columns: Dict[str, Column]  
    inputs: List[str]  
    relationships: List[Relationship]  
    step: Step

4.2 Topological order & graph

def topo_sort(entities):  
    # edges: input → entity  
    indeg = {e: 0 for e in entities}  
    for e in entities.values():  
        for i in e.inputs:  
            indeg[e.fqname] += 1  
    # Kahn’s algorithm …

4.3 Rendering templates

Use Jinja2 templates for dbt and SQLMesh targets.

dbt_model.sql.j2

{{ config(materialized=materialization) }}  
{% for cte in ctes %}{{ cte.name }} as (  
  {{ cte.sql }}  
){% if not loop.last %},{% endif %}  
{% endfor %}  
select {{ select_list }}  
from {{ base_from }}  
{% for j in joins %}{{ j }}{% endfor %}  
{% if where_clause %}  
where {{ where_clause }}  
{% endif %}  
{% if group_by %}  
group by {{ group_by|join(', ') }}  
{% endif %}

sqlmesh_model.sql.j2

MODEL (  
  name {{ model_name }},  
  kind {{ kind }},  
  audits ({{ audits|join(', ') }})  
);  
{{ body_sql }}

5) Emitting dbt artifacts

Paths: map roles to folders (configurable)

source → models/staging/  
intermediate → models/integration/  
delivery → models/marts/

Models: one SQL per entity; ref() for upstream entities, source() for raw sources.
Tests: generate schema.yml with not_null, unique, relationships, etc.
Sources: consolidate into sources.yml per subject area.
Macros: ship generic macros (e.g., translate_code, date_diff dialect wrappers).
Analysis / profiling: emit helper queries into analysis/ or tests/profiling/.

6) Emitting SQLMesh artifacts

Models: one SQL with MODEL(...) header (name, kind, audits).
Lineage: use ref()/source() helpers or let SQLMesh infer from SQL.
Audits: translate PK/UK/NN/FK constraints.
Macros: provide compatible macros for translations and platform hints.

7) End‑to‑end example (derive → aggregate)

7.1 YAML entities

sales.order (source, filter step)
sales.customer (source)
sales.order_enriched (intermediate, derive)
sales.order_daily_metrics (delivery, aggregate)

7.2 Emitted dbt (shortened)

-- models/integration/order_enriched.sql  
{{ config(materialized='table') }}  
with o as (select * from {{ ref('stg_order') }}),  
     c as (select * from {{ ref('stg_customer') }})  
select o.order_id, o.customer_id, c.country_code, o.amount,  
       datediff('day', o.order_date, o.ship_date) as days_to_ship  
from o join c on o.customer_id = c.customer_id

-- models/marts/order_daily_metrics.sql  
{{ config(materialized='table') }}  
select order_date, country_code,  
       count(1) as orders_cnt,  
       sum(amount) as amount_sum,  
       concat(order_date, '_', country_code) as pk_day_country  
from {{ ref('order_enriched') }}  
group by 1,2

7.3 Emitted SQLMesh (shortened)

MODEL (name analytics.order_enriched, kind FULL, audits (not_null(order_id)));  
with o as (select * from ref('analytics.stg_order')),  
     c as (select * from ref('analytics.stg_customer'))  
select o.order_id, o.customer_id, c.country_code, o.amount,  
       datediff(day, o.order_date, o.ship_date) as days_to_ship  
from o join c on o.customer_id = c.customer_id;

8) CI/CD & developer workflow

Pre‑commit: YAML formatting (prettier), JSON Schema validation, semantic validators.
Unit tests: generator functions (AST transforms, renderers).
Snapshot tests: freeze generated SQL; diff on PR to detect unintended changes.
dbt/SQLMesh jobs: compile on PR; run smoke tests in ephemeral schema.
Docs build: generate draw.io/Mermaid ERDs + lineage; publish to Confluence/SharePoint.
Linting: sqlfluff (dialect per target) on emitted SQL.

9) Extensions & patterns

Incremental strategies: declarative incremental: {unique_key: ..., predicate: ...} in YAML → materialization config.
Dual SCD2 & PIT: declarative historization rules; macros emit platform‑optimized SQL.
Reference data translation: translate step + function catalog (logic names → dialect macros).
Platform hints: clustering/partitioning as emitter hints (Snowflake, BigQuery, Databricks).
Security tags: column/classification metadata → propagate to catalogs or masking policies.

10) Diagrams & docs from metadata

ERD: entity boxes (role/pattern badges), PK/FK, subject‑area swimlanes.
Data movement: step‑type glyphs along edges (derive/filter/aggregate/union).
Generators: Python → draw.io XML or Mermaid; embed into Markdown; publish to Confluence.

11) Tying back to the series (links)

Business context: A1 — Why We Need a Business‑Friendly Data Transformation Model
Technical mapping: B1 — How dbt & SQLMesh Organize Metadata and Code

This article delivers the practical bridge: a schema, code skeletons, and a reproducible workflow.

12) Next steps

Initialize a repo with /entities, /tooling, and /generated folders.
Add the JSON Schema and validators.
Implement the dbt and sqlmesh emitters using the templates above.
Wire CI to validate, generate, compile, and publish docs.
Migrate one subject area from hand‑written SQL to Entity‑First and compare results.

🔗 Series footer

Previous: B1 — How dbt & SQLMesh Organize Metadata and Code
This article: B2 — Designing an Entity‑First YAML → SQL Generator
Next: A2 — Entities, Subject Areas & Transformation Steps