Jinja2 SQL Templating — Conditional Blocks & Reusable Fragments

Jinja2 lets you write SQL templates with conditional blocks, loops, and macros — keeping query logic readable while adapting to runtime conditions. It is the engine behind dbt, SQLFluff, and many query builders.

from jinja2 import Environment, BaseLoader
import psycopg2
from psycopg2.extras import RealDictCursor

env = Environment(loader=BaseLoader())

# ── Template with conditional blocks ─────────────────────────────
TWEET_REPORT_TEMPLATE = """
SELECT
    u.username,
    {{ "u.verified," if include_verified else "" }}
    COUNT(t.id)        AS tweet_count,
    SUM(t.like_count)  AS total_likes
    {% if include_impressions %},
    SUM(t.impression_count) AS total_impressions
    {% endif %}
FROM tw_users u
JOIN tw_tweets t ON t.user_id = u.id
WHERE t.retweet_of_id IS NULL
  {% if since %}
  AND t.created_at >= :since
  {% endif %}
  {% if hashtags %}
  AND t.id IN (
      SELECT th.tweet_id
      FROM tw_tweet_hashtags th
      JOIN tw_hashtags h ON h.id = th.hashtag_id
      WHERE h.tag IN ({{ hashtags | map("lower") | list | join(", ") | tojson }})
  )
  {% endif %}
GROUP BY u.username {{ ", u.verified" if include_verified else "" }}
HAVING COUNT(t.id) >= :min_tweets
ORDER BY total_likes DESC
LIMIT :limit
"""

def render_tweet_report(
    include_verified: bool = True,
    include_impressions: bool = False,
    since: str | None = None,
    hashtags: list[str] | None = None,
    min_tweets: int = 5,
    limit: int = 20,
):
    template = env.from_string(TWEET_REPORT_TEMPLATE)
    sql = template.render(
        include_verified=include_verified,
        include_impressions=include_impressions,
        since=since,
        hashtags=hashtags or [],
    )
    # Replace :param with %s for psycopg2
    params = {
        "since": since,
        "min_tweets": min_tweets,
        "limit": limit,
    }
    return sql, params

sql, params = render_tweet_report(
    include_verified=True,
    include_impressions=True,
    since="2024-06-01",
    hashtags=["#sql", "#python"],
)
print(sql)