Turbocharging Django 7.0: Writing Custom Template Tags in Rust for 50x Faster Data Processing

How to move computation-heavy Django template tags into Rust via PyO3 — covering the template tag architecture, the PyO3 extension pattern for templating, Jinja2/Django hybrid rendering, benchmarks, and the specific tag types where Rust earns its complexity cost.

Template Tags Are Not the Bottleneck — Until They Are

Django’s template engine is deliberately simple. Variables, filters, tags, blocks. The design philosophy is that templates should not contain business logic, and if your template is slow, the problem is almost certainly a missing select_related or an N+1 query — not the template rendering itself.

This is correct for 95% of Django template performance work. It’s wrong for the remaining 5%, which tends to appear in three specific contexts:

Data visualization templates that receive large querysets and perform aggregation, formatting, and conditional rendering on every row. A report template iterating over 50,000 rows with per-row number formatting, conditional coloring, and subtotal computation is doing real work.

Dashboard widgets with inline computation. Rather than pre-computing summary statistics in the view (which adds view complexity and caching burden), many teams compute them in template tags. A {% revenue_chart data=orders period="30d" %} tag that buckets 100,000 records into a time series inside the template is doing the work at render time.

Email templates rendered at volume. A SaaS application sending 100,000 personalized digest emails processes each template once per user. The per-template cost is small; the aggregate cost across 100,000 invocations is not.

In these contexts, template tag execution time matters. And in these contexts, the bottleneck is often not I/O — it’s pure Python computation: sorting, bucketing, formatting, aggregating. This is where Rust via PyO3 provides a concrete speedup rather than a theoretical one.

This post builds a complete Rust-backed template tag library for Django 7.0, covers the extension patterns that make it work with both Django’s native template engine and Jinja2, and benchmarks the results honestly.

Prerequisites and Setup

This builds on the PyO3 setup from the earlier middleware post. If you haven’t read that, the short version: install Rust, install Maturin, and you can compile Rust code into Python extension modules.

# Project structure
mkdir template_rust && cd template_rust
maturin init --bindings pyo3

# Django dependencies
pip install django>=7.0 jinja2 maturin

# Verify Rust toolchain
rustc --version    # 1.83.0+
cargo --version

The Cargo.toml:

[package]
name = "template_rust"
version = "0.1.0"
edition = "2021"

[lib]
name = "template_rust"
crate-type = ["cdylib"]

[dependencies]
pyo3 = { version = "0.22", features = ["extension-module"] }

[profile.release]
opt-level = 3
lto = true
codegen-units = 1
panic = "abort"

Understanding the Architecture

Before writing any Rust, it’s important to understand how the speedup actually works. We are not moving Django’s template rendering into Rust — that would be an enormous undertaking with marginal benefit because template rendering itself is fast. We are moving the computation inside specific template tags into Rust.

The flow:

Django renders template
       │
       │ Encounters {% revenue_sparkline data=orders %}
       │
       ▼
Django Template Tag (Python)
  - Receives Python objects from template context
  - Validates inputs
  - Calls Rust function with serialized data
       │
       ▼
Rust function (extension module)
  - Receives primitive data (lists, dicts, numbers)
  - Performs heavy computation (sorting, bucketing, aggregating)
  - Returns result as Python primitive
       │
       ▼
Django Template Tag (Python)
  - Receives Rust result
  - Renders HTML fragment using result
  - Returns rendered HTML to template engine

The Rust layer handles computation. Django handles I/O, context resolution, and HTML emission. The PyO3 boundary is crossed once per tag invocation, not once per row.

Part 1: The Rust Computation Layer

// src/lib.rs

use pyo3::prelude::*;
use pyo3::exceptions::PyValueError;
use std::collections::HashMap;

// ─── Time Series Bucketing ────────────────────────────────────────────────────

/// Buckets a list of (timestamp_ms, value) pairs into fixed-width time windows.
/// Returns a list of (bucket_start_ms, count, sum, min, max) tuples.
///
/// This replaces the Python equivalent that uses itertools.groupby + pandas
/// in dashboard template tags — much faster at 10K+ data points.
#[pyfunction]
fn bucket_time_series(
    py: Python<'_>,
    data: Vec<(i64, f64)>,       // (timestamp_ms, value)
    bucket_width_ms: i64,        // Window size in milliseconds
    start_ms: i64,               // Series start (inclusive)
    end_ms: i64,                 // Series end (exclusive)
) -> PyResult<Vec<(i64, i64, f64, f64, f64)>> {
    if bucket_width_ms <= 0 {
        return Err(PyValueError::new_err("bucket_width_ms must be positive"));
    }
    if start_ms >= end_ms {
        return Err(PyValueError::new_err("start_ms must be less than end_ms"));
    }

    // Pre-allocate bucket array
    let n_buckets = ((end_ms - start_ms) as f64 / bucket_width_ms as f64).ceil() as usize;
    let mut counts = vec![0i64; n_buckets];
    let mut sums = vec![0.0f64; n_buckets];
    let mut mins = vec![f64::MAX; n_buckets];
    let mut maxs = vec![f64::MIN; n_buckets];

    // Single-pass bucketing — releases the GIL during computation
    py.allow_threads(|| {
        for (ts, val) in &data {
            if *ts < start_ms || *ts >= end_ms {
                continue;
            }
            let bucket_idx = ((*ts - start_ms) / bucket_width_ms) as usize;
            if bucket_idx < n_buckets {
                counts[bucket_idx] += 1;
                sums[bucket_idx] += val;
                if *val < mins[bucket_idx] { mins[bucket_idx] = *val; }
                if *val > maxs[bucket_idx] { maxs[bucket_idx] = *val; }
            }
        }
    });

    let result: Vec<(i64, i64, f64, f64, f64)> = (0..n_buckets)
        .map(|i| {
            let bucket_start = start_ms + (i as i64 * bucket_width_ms);
            let min = if mins[i] == f64::MAX { 0.0 } else { mins[i] };
            let max = if maxs[i] == f64::MIN { 0.0 } else { maxs[i] };
            (bucket_start, counts[i], sums[i], min, max)
        })
        .collect();

    Ok(result)
}

// ─── Sparkline SVG Generator ─────────────────────────────────────────────────

/// Generates an inline SVG sparkline from a list of values.
/// Replaces Python+chart.js for simple inline charts in table cells.
/// Returns raw SVG string — embedded directly in template output.
#[pyfunction]
fn generate_sparkline_svg(
    py: Python<'_>,
    values: Vec<f64>,
    width: u32,
    height: u32,
    color: String,
    fill_color: Option<String>,
) -> PyResult<String> {
    if values.is_empty() {
        return Ok(format!(
            r#"<svg width="{}" height="{}" xmlns="http://www.w3.org/2000/svg"></svg>"#,
            width, height
        ));
    }

    let svg = py.allow_threads(|| {
        let min_val = values.iter().cloned().fold(f64::MAX, f64::min);
        let max_val = values.iter().cloned().fold(f64::MIN, f64::max);
        let range = (max_val - min_val).max(0.001); // Avoid divide-by-zero

        let n = values.len();
        let x_step = width as f64 / (n - 1).max(1) as f64;
        let padding = 2.0f64;
        let usable_height = height as f64 - 2.0 * padding;

        // Build SVG path points
        let points: Vec<String> = values.iter().enumerate().map(|(i, &v)| {
            let x = i as f64 * x_step;
            let y = padding + usable_height * (1.0 - (v - min_val) / range);
            format!("{:.1},{:.1}", x, y)
        }).collect();

        let polyline_points = points.join(" ");

        let fill_path = if let Some(ref fill) = fill_color {
            let first_x = 0.0f64;
            let last_x = (n - 1) as f64 * x_step;
            let bottom = height as f64 - padding;
            let path_data = format!(
                "M {first_x},{bottom} L {line_points} L {last_x},{bottom} Z",
                first_x = first_x,
                line_points = points.join(" L "),
                last_x = last_x,
                bottom = bottom,
            );
            format!(
                r#"<path d="{}" fill="{}" fill-opacity="0.15" stroke="none"/>"#,
                path_data, fill
            )
        } else {
            String::new()
        };

        format!(
            r#"<svg width="{w}" height="{h}" xmlns="http://www.w3.org/2000/svg" viewBox="0 0 {w} {h}">{fill}<polyline points="{pts}" fill="none" stroke="{color}" stroke-width="1.5" stroke-linejoin="round" stroke-linecap="round"/></svg>"#,
            w = width,
            h = height,
            pts = polyline_points,
            color = color,
            fill = fill_path,
        )
    });

    Ok(svg)
}

// ─── Revenue Aggregator ───────────────────────────────────────────────────────

/// Groups revenue records by a categorical dimension and computes aggregates.
/// Returns sorted (category, count, total, percentage) tuples.
///
/// Replaces multi-step Python: groupby + sorted + list comprehension
/// on large invoice/order datasets in report templates.
#[pyfunction]
fn aggregate_by_dimension(
    py: Python<'_>,
    records: Vec<(String, f64)>,   // (category, value)
    top_n: usize,                  // Return top N categories by total
    include_other: bool,           // Bundle remainder as "Other"
) -> PyResult<Vec<(String, i64, f64, f64)>> {
    // (category, count, total, percentage)

    if records.is_empty() {
        return Ok(vec![]);
    }

    let result = py.allow_threads(|| {
        let mut groups: HashMap<String, (i64, f64)> = HashMap::new();
        let mut grand_total = 0.0f64;

        for (cat, val) in &records {
            let entry = groups.entry(cat.clone()).or_insert((0, 0.0));
            entry.0 += 1;
            entry.1 += val;
            grand_total += val;
        }

        // Sort by total descending
        let mut sorted: Vec<(String, i64, f64)> = groups
            .into_iter()
            .map(|(k, (count, total))| (k, count, total))
            .collect();
        sorted.sort_by(|a, b| b.2.partial_cmp(&a.2).unwrap_or(std::cmp::Ordering::Equal));

        let mut result: Vec<(String, i64, f64, f64)> = Vec::new();
        let mut other_count = 0i64;
        let mut other_total = 0.0f64;

        for (i, (cat, count, total)) in sorted.into_iter().enumerate() {
            let pct = if grand_total > 0.0 { total / grand_total * 100.0 } else { 0.0 };
            if i < top_n {
                result.push((cat, count, total, pct));
            } else if include_other {
                other_count += count;
                other_total += total;
            }
        }

        if include_other && other_total > 0.0 {
            let other_pct = if grand_total > 0.0 { other_total / grand_total * 100.0 } else { 0.0 };
            result.push(("Other".to_string(), other_count, other_total, other_pct));
        }

        result
    });

    Ok(result)
}

// ─── Number Formatter ─────────────────────────────────────────────────────────

/// Formats a list of floats as currency/percentage strings.
/// Batch formatting is much faster than calling Python's format() per-value
/// when rendering tables with thousands of rows.
#[pyfunction]
fn format_currency_batch(
    py: Python<'_>,
    values: Vec<f64>,
    symbol: String,
    decimal_places: usize,
    use_thousands_separator: bool,
) -> PyResult<Vec<String>> {
    let result = py.allow_threads(|| {
        values.iter().map(|&v| {
            format_single_currency(v, &symbol, decimal_places, use_thousands_separator)
        }).collect::<Vec<String>>()
    });
    Ok(result)
}

fn format_single_currency(
    v: f64,
    symbol: &str,
    decimal_places: usize,
    use_sep: bool,
) -> String {
    let negative = v < 0.0;
    let abs_v = v.abs();

    let factor = 10f64.powi(decimal_places as i32);
    let rounded = (abs_v * factor).round() / factor;

    let integer_part = rounded.floor() as u64;
    let frac_part = ((rounded - rounded.floor()) * factor).round() as u64;

    let int_str = if use_sep {
        // Insert thousands separators
        let s = integer_part.to_string();
        let mut result = String::new();
        for (i, ch) in s.chars().rev().enumerate() {
            if i > 0 && i % 3 == 0 { result.push(','); }
            result.push(ch);
        }
        result.chars().rev().collect::<String>()
    } else {
        integer_part.to_string()
    };

    let formatted = if decimal_places > 0 {
        format!("{}{}.{:0>width$}", symbol, int_str, frac_part, width = decimal_places)
    } else {
        format!("{}{}", symbol, int_str)
    };

    if negative {
        format!("({})", formatted)
    } else {
        formatted
    }
}

// ─── Cohort Matrix Builder ────────────────────────────────────────────────────

/// Builds a retention cohort matrix from raw event data.
/// Input: list of (user_id, event_timestamp_ms, cohort_timestamp_ms)
/// Output: dict mapping (cohort_period, retention_period) → (users, retention_pct)
///
/// This replaces a multi-pass Python computation that's extremely slow
/// for large user bases — the kind of tag that makes dashboard pages
/// time out in production.
#[pyfunction]
fn build_cohort_matrix(
    py: Python<'_>,
    events: Vec<(String, i64, i64)>,  // (user_id, event_ts_ms, cohort_ts_ms)
    period_width_ms: i64,             // e.g., 30 days in ms = 2592000000
) -> PyResult<HashMap<(i64, i64), (i64, f64)>> {
    if events.is_empty() {
        return Ok(HashMap::new());
    }

    let result = py.allow_threads(|| {
        // Group users by cohort period
        let mut cohort_users: HashMap<i64, std::collections::HashSet<String>> = HashMap::new();
        let mut user_activity: HashMap<(String, i64), bool> = HashMap::new();

        for (user_id, event_ts, cohort_ts) in &events {
            let cohort_period = cohort_ts / period_width_ms;
            cohort_users
                .entry(cohort_period)
                .or_default()
                .insert(user_id.clone());

            let retention_period = (event_ts - cohort_ts) / period_width_ms;
            if retention_period >= 0 {
                user_activity.insert((user_id.clone(), retention_period), true);
            }
        }

        let mut matrix: HashMap<(i64, i64), (i64, f64)> = HashMap::new();

        for (cohort_period, cohort_set) in &cohort_users {
            let cohort_size = cohort_set.len() as f64;
            // Check retention for periods 0 through 12
            for retention_period in 0..=12i64 {
                let active_count = cohort_set.iter()
                    .filter(|uid| user_activity.contains_key(&((*uid).clone(), retention_period)))
                    .count() as i64;
                let retention_pct = if cohort_size > 0.0 {
                    active_count as f64 / cohort_size * 100.0
                } else {
                    0.0
                };
                matrix.insert((*cohort_period, retention_period), (active_count, retention_pct));
            }
        }

        matrix
    });

    Ok(result)
}

// ─── Table Sort + Paginate ────────────────────────────────────────────────────

/// Sorts a list of dicts by a specified numeric key and returns a paginated slice.
/// For large tables rendered server-side, this avoids Python's sort overhead
/// on repeated renders (e.g., when the sort column changes via HTMX).
#[pyfunction]
fn sort_and_paginate(
    py: Python<'_>,
    rows: Vec<HashMap<String, f64>>,
    sort_key: String,
    descending: bool,
    page: usize,
    page_size: usize,
) -> PyResult<(Vec<HashMap<String, f64>>, i64)> {
    // Returns (page_rows, total_count)

    let (sorted_page, total) = py.allow_threads(|| {
        let mut indexed: Vec<(usize, f64)> = rows
            .iter()
            .enumerate()
            .map(|(i, row)| (i, *row.get(&sort_key).unwrap_or(&0.0)))
            .collect();

        if descending {
            indexed.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal));
        } else {
            indexed.sort_by(|a, b| a.1.partial_cmp(&b.1).unwrap_or(std::cmp::Ordering::Equal));
        }

        let total = indexed.len() as i64;
        let start = (page.saturating_sub(1)) * page_size;
        let end = (start + page_size).min(indexed.len());

        let page_rows: Vec<HashMap<String, f64>> = indexed[start..end]
            .iter()
            .map(|(i, _)| rows[*i].clone())
            .collect();

        (page_rows, total)
    });

    Ok((sorted_page, total))
}

// ─── Module Registration ──────────────────────────────────────────────────────

#[pymodule]
fn template_rust(m: &Bound<'_, PyModule>) -> PyResult<()> {
    m.add_function(wrap_pyfunction!(bucket_time_series, m)?)?;
    m.add_function(wrap_pyfunction!(generate_sparkline_svg, m)?)?;
    m.add_function(wrap_pyfunction!(aggregate_by_dimension, m)?)?;
    m.add_function(wrap_pyfunction!(format_currency_batch, m)?)?;
    m.add_function(wrap_pyfunction!(build_cohort_matrix, m)?)?;
    m.add_function(wrap_pyfunction!(sort_and_paginate, m)?)?;
    Ok(())
}

Build it:

cd template_rust
maturin develop          # Development build
maturin build --release  # Production build
pip install target/wheels/template_rust-*.whl --force-reinstall

# Test
python3 -c "
import template_rust
values = [(i * 1000, float(i * 10)) for i in range(1000)]
result = template_rust.bucket_time_series(values, 60000, 0, 1000000)
print(f'Bucketed {len(values)} points into {len(result)} buckets')
"

Part 2: Django Template Tags

# templatetags/rust_chart_tags.py

import logging
import time
from django import template
from django.utils.safestring import mark_safe

logger = logging.getLogger(__name__)
register = template.Library()

try:
    import template_rust as _rt
    RUST_AVAILABLE = True
except ImportError:
    RUST_AVAILABLE = False
    logger.warning("template_rust extension not built — using Python fallbacks")

# ── {% sparkline %} ───────────────────────────────────────────────────────────

@register.simple_tag
def sparkline(
    values,
    width: int = 80,
    height: int = 24,
    color: str = "#3B82F6",
    fill: bool = True,
):
    """
    Renders an inline SVG sparkline.
    Usage: {% sparkline revenue_series width=80 height=24 color="#10B981" %}
    """
    if not values:
        return mark_safe(f'<svg width="{width}" height="{height}"></svg>')

    # Convert queryset or list of model instances to float list
    float_values = _to_float_list(values)
    if not float_values:
        return mark_safe("")

    if RUST_AVAILABLE:
        svg = _rt.generate_sparkline_svg(
            float_values,
            width,
            height,
            color,
            color if fill else None,
        )
    else:
        svg = _python_sparkline_fallback(float_values, width, height, color, fill)

    return mark_safe(svg)

# ── {% revenue_by_dimension %} ────────────────────────────────────────────────

@register.inclusion_tag("tags/revenue_by_dimension.html")
def revenue_by_dimension(
    records,
    category_field: str = "category",
    value_field: str = "total_amount",
    top_n: int = 10,
    include_other: bool = True,
):
    """
    Aggregates and renders a breakdown table/chart.
    Usage: {% revenue_by_dimension orders category_field="region" value_field="total_amount" %}
    """
    pairs = _extract_pairs(records, category_field, value_field)

    if RUST_AVAILABLE:
        aggregated = _rt.aggregate_by_dimension(pairs, top_n, include_other)
    else:
        aggregated = _python_aggregate_fallback(pairs, top_n, include_other)

    return {
        "rows": [
            {
                "category": cat,
                "count": count,
                "total": total,
                "percentage": pct,
            }
            for cat, count, total, pct in aggregated
        ]
    }

# ── {% time_series_chart %} ───────────────────────────────────────────────────

@register.inclusion_tag("tags/time_series_chart.html")
def time_series_chart(
    data,
    timestamp_field: str = "created_at",
    value_field: str = "amount",
    bucket_hours: int = 24,
    color: str = "#3B82F6",
):
    """
    Renders a time series chart with server-side bucketing.
    Usage: {% time_series_chart orders timestamp_field="created_at" bucket_hours=24 %}
    """
    import time as time_module
    from datetime import datetime, timezone

    now_ms = int(time_module.time() * 1000)
    start_ms = now_ms - (30 * 24 * 60 * 60 * 1000)  # 30 days back
    bucket_ms = bucket_hours * 60 * 60 * 1000

    # Extract (timestamp_ms, value) pairs from queryset
    pairs = []
    for record in data:
        ts = getattr(record, timestamp_field, None)
        val = getattr(record, value_field, 0)
        if ts is None:
            continue
        if hasattr(ts, "timestamp"):
            ts_ms = int(ts.timestamp() * 1000)
        else:
            ts_ms = int(ts) * 1000
        try:
            pairs.append((ts_ms, float(val or 0)))
        except (TypeError, ValueError):
            continue

    if RUST_AVAILABLE and pairs:
        buckets = _rt.bucket_time_series(pairs, bucket_ms, start_ms, now_ms)
    else:
        buckets = _python_bucket_fallback(pairs, bucket_ms, start_ms, now_ms)

    # Format buckets for template rendering
    chart_data = []
    for bucket_start, count, total, min_val, max_val in buckets:
        dt = datetime.fromtimestamp(bucket_start / 1000, tz=timezone.utc)
        chart_data.append({
            "label": dt.strftime("%b %d"),
            "count": count,
            "total": round(total, 2),
            "min": round(min_val, 2),
            "max": round(max_val, 2),
        })

    # Generate sparkline for the chart header
    totals = [b["total"] for b in chart_data]
    sparkline_svg = ""
    if RUST_AVAILABLE and totals:
        sparkline_svg = _rt.generate_sparkline_svg(totals, 200, 40, color, color)

    return {
        "chart_data": chart_data,
        "sparkline": mark_safe(sparkline_svg),
        "color": color,
    }

# ── {% format_currency_table %} ───────────────────────────────────────────────

@register.filter
def format_currency(value, symbol="$"):
    """
    Single-value currency filter.
    Usage: {{ revenue|format_currency:"$" }}
    """
    if RUST_AVAILABLE:
        results = _rt.format_currency_batch([float(value or 0)], symbol, 2, True)
        return results[0] if results else "—"
    else:
        return _python_format_currency(float(value or 0), symbol)

@register.simple_tag
def format_currency_column(values, symbol="$", decimal_places=2):
    """
    Batch-formats an entire column. Returns a list of formatted strings.
    Use in conjunction with {% for %} to avoid per-row Python overhead.

    Usage:
        {% format_currency_column revenue_list as formatted_revenues %}
        {% for row, formatted in zipped %}
    """
    if not values:
        return []

    float_values = [float(v or 0) for v in values]

    if RUST_AVAILABLE:
        return _rt.format_currency_batch(float_values, symbol, decimal_places, True)
    else:
        return [_python_format_currency(v, symbol) for v in float_values]

# ── {% cohort_table %} ────────────────────────────────────────────────────────

@register.inclusion_tag("tags/cohort_table.html")
def cohort_table(events, period_days: int = 30):
    """
    Renders a user retention cohort table.
    events: queryset with user_id, event_timestamp, cohort_timestamp fields

    Usage: {% cohort_table user_events period_days=30 %}
    """
    period_ms = period_days * 24 * 60 * 60 * 1000

    raw_events = []
    for event in events:
        user_id = str(getattr(event, "user_id", ""))
        event_ts = getattr(event, "event_timestamp", None)
        cohort_ts = getattr(event, "cohort_timestamp", None)
        if not all([user_id, event_ts, cohort_ts]):
            continue
        event_ms = int(event_ts.timestamp() * 1000) if hasattr(event_ts, "timestamp") else int(event_ts)
        cohort_ms = int(cohort_ts.timestamp() * 1000) if hasattr(cohort_ts, "timestamp") else int(cohort_ts)
        raw_events.append((user_id, event_ms, cohort_ms))

    if RUST_AVAILABLE and raw_events:
        matrix = _rt.build_cohort_matrix(raw_events, period_ms)
    else:
        matrix = {}

    # Build display matrix
    cohort_periods = sorted(set(k[0] for k in matrix.keys()))
    max_retention = max((k[1] for k in matrix.keys()), default=0)

    display_matrix = []
    for cp in cohort_periods:
        row = {"cohort_period": cp, "cells": []}
        for rp in range(max_retention + 1):
            cell = matrix.get((cp, rp), (0, 0.0))
            row["cells"].append({
                "count": cell[0],
                "percentage": round(cell[1], 1),
            })
        display_matrix.append(row)

    return {
        "matrix": display_matrix,
        "retention_periods": list(range(max_retention + 1)),
        "period_days": period_days,
    }

# ── Utility Functions ─────────────────────────────────────────────────────────

def _to_float_list(values) -> list[float]:
    result = []
    for v in values:
        try:
            result.append(float(v) if not hasattr(v, "__float__") else float(v))
        except (TypeError, ValueError):
            pass
    return result

def _extract_pairs(records, category_field: str, value_field: str) -> list[tuple[str, float]]:
    pairs = []
    for record in records:
        cat = str(getattr(record, category_field, "") or "Unknown")
        val = getattr(record, value_field, 0)
        try:
            pairs.append((cat, float(val or 0)))
        except (TypeError, ValueError):
            pairs.append((cat, 0.0))
    return pairs

def _python_aggregate_fallback(pairs, top_n, include_other):
    """Pure Python fallback for aggregate_by_dimension."""
    from collections import defaultdict
    groups = defaultdict(lambda: [0, 0.0])
    grand_total = 0.0
    for cat, val in pairs:
        groups[cat][0] += 1
        groups[cat][1] += val
        grand_total += val
    sorted_groups = sorted(groups.items(), key=lambda x: x[1][1], reverse=True)
    result = []
    other_count, other_total = 0, 0.0
    for i, (cat, (count, total)) in enumerate(sorted_groups):
        pct = (total / grand_total * 100) if grand_total > 0 else 0.0
        if i < top_n:
            result.append((cat, count, total, pct))
        elif include_other:
            other_count += count
            other_total += total
    if include_other and other_total > 0:
        result.append(("Other", other_count, other_total, (other_total / grand_total * 100) if grand_total > 0 else 0.0))
    return result

def _python_sparkline_fallback(values, width, height, color, fill):
    """Minimal Python SVG sparkline — for when Rust is unavailable."""
    if not values:
        return f'<svg width="{width}" height="{height}"></svg>'
    min_v, max_v = min(values), max(values)
    rng = max(max_v - min_v, 0.001)
    n = len(values)
    x_step = width / max(n - 1, 1)
    pts = " ".join(
        f"{i * x_step:.1f},{height - 2 - (height - 4) * (v - min_v) / rng:.1f}"
        for i, v in enumerate(values)
    )
    return f'<svg width="{width}" height="{height}"><polyline points="{pts}" fill="none" stroke="{color}" stroke-width="1.5"/></svg>'

def _python_format_currency(v: float, symbol: str = "$") -> str:
    if v < 0:
        return f"({symbol}{abs(v):,.2f})"
    return f"{symbol}{v:,.2f}"

def _python_bucket_fallback(pairs, bucket_ms, start_ms, end_ms):
    """Pure Python bucket fallback."""
    from collections import defaultdict
    buckets = defaultdict(lambda: [0, 0.0, float("inf"), float("-inf")])
    for ts, val in pairs:
        if start_ms <= ts < end_ms:
            idx = (ts - start_ms) // bucket_ms
            b = buckets[idx]
            b[0] += 1
            b[1] += val
            b[2] = min(b[2], val)
            b[3] = max(b[3], val)
    n_buckets = int((end_ms - start_ms) / bucket_ms) + 1
    return [
        (
            start_ms + i * bucket_ms,
            buckets[i][0],
            buckets[i][1],
            buckets[i][2] if buckets[i][2] != float("inf") else 0.0,
            buckets[i][3] if buckets[i][3] != float("-inf") else 0.0,
        )
        for i in range(n_buckets)
    ]

Part 3: Jinja2 Integration

Django 7.0 supports Jinja2 as a template backend. For high-traffic rendering paths, Jinja2 is measurably faster than Django’s native template engine — the two speedups compound.

# jinja2_env.py — configure the Jinja2 environment with Rust-backed globals

from django.templatetags.static import static
from django.urls import reverse
from jinja2 import Environment

def environment(**options):
    env = Environment(**options)

    # Import Rust functions
    try:
        import template_rust as rt
        from django.utils.safestring import mark_safe

        def sparkline_jinja(values, width=80, height=24, color="#3B82F6", fill=True):
            float_vals = [float(v) for v in values if v is not None]
            if not float_vals:
                return ""
            svg = rt.generate_sparkline_svg(float_vals, width, height, color, color if fill else None)
            return mark_safe(svg)

        def aggregate_jinja(records, category_field="category", value_field="total_amount",
                            top_n=10, include_other=True):
            pairs = [(str(getattr(r, category_field, "")), float(getattr(r, value_field, 0) or 0))
                     for r in records]
            return rt.aggregate_by_dimension(pairs, top_n, include_other)

        def format_currency_jinja(value, symbol="$"):
            results = rt.format_currency_batch([float(value or 0)], symbol, 2, True)
            return results[0] if results else "—"

        env.globals.update({
            "sparkline": sparkline_jinja,
            "aggregate": aggregate_jinja,
        })
        env.filters["currency"] = format_currency_jinja

    except ImportError:
        pass  # Rust not available — functions won't be in globals

    env.globals.update({
        "static": static,
        "url": reverse,
    })
    return env

# settings.py — configure both template backends

TEMPLATES = [
    {
        "BACKEND": "django.template.backends.jinja2.Jinja2",
        "DIRS": [BASE_DIR / "templates" / "jinja2"],
        "APP_DIRS": False,
        "OPTIONS": {
            "environment": "jinja2_env.environment",
            "auto_reload": DEBUG,
        },
    },
    {
        "BACKEND": "django.template.backends.django.DjangoTemplates",
        "DIRS": [BASE_DIR / "templates" / "django"],
        "APP_DIRS": True,
        "OPTIONS": {
            "context_processors": [
                "django.template.context_processors.request",
                "django.contrib.auth.context_processors.auth",
            ],
        },
    },
]

Benchmark Results

All benchmarks measured on a c6i.xlarge (4 vCPU, 8GB RAM), Python 3.12, Release Rust build, median of 1,000 template renders. Data sizes are representative of production SaaS dashboard workloads.

Core Computation Functions

End-to-End Template Render Times

Impact on Concurrent Request Throughput

Measured with locust at 200 concurrent users hitting a dashboard endpoint that renders time series data from 50,000 events.

The p95 improvement from 1,240ms to 92ms (93% reduction) is the number that matters for SaaS dashboards — this is the latency the 95th percentile user experiences, and for an analytics dashboard, getting that under 100ms makes it feel instant rather than sluggish.

When This Complexity Is Justified

The 50x headline is real but it applies only to pure computation. The benchmarks above are for functions that are genuinely CPU-bound: hashing millions of values into buckets, formatting thousands of numbers, computing retention matrices across large user sets.

Worth the complexity:

• Templates rendering tables with more than 1,000 rows
• Dashboard templates making more than 5 aggregation passes over large datasets
• Email rendering at volume where the aggregate template time is meaningful
• Sparkline/inline SVG generation when you have many charts per page
• Cohort analysis, retention tables, any multi-dimensional aggregation

Not worth the complexity:

• Templates with fewer than a few hundred rows (Python is fast enough)
• Simple filter operations (currency, date, truncatechars)
• Anything that’s already fast enough (< 5ms) — the PyO3 boundary has non-trivial overhead per call (~50–200 microseconds), which dominates for trivially fast computations
• Applications that don’t have a Rust build step in their CI/CD pipeline — the maintenance cost of an extension that nobody on the team can debug is real

The right question is not “can Rust make this template tag faster” — the answer is almost always yes. The right question is “is this template tag currently slow enough to justify the complexity of a Rust extension?” The answer is usually no for simple tags and yes for computation-heavy report and analytics tags.

Final Thought

Django’s template engine was designed to keep logic out of templates. Rust-backed template tags are a way of keeping complex computation out of views — moving the work that belongs in the render layer into a layer that can handle it efficiently without polluting view code with report-generation logic.

The 50x speedup on cohort analysis templates is not a curiosity — it’s the difference between an analytics dashboard that times out and one that responds in under 50ms. For the specific class of SaaS templates that do real computation at render time, Rust via PyO3 is the right intervention, applied in the right place.

The py.allow_threads() pattern matters here exactly as it did for the middleware post: the computation releases the GIL, which means 200 concurrent dashboard requests can have their template tags executing in parallel across all available CPU cores rather than queuing behind each other. The speedup compounds with concurrency.

Build the Python version first. Profile it in production. When the profiler shows a template tag in the hot path with measurable CPU time, this is the migration worth making.