use std::collections::BTreeMap;
use std::fmt;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use chrono::{NaiveDate, Utc};
use crate::dates::{Calendar, Date, DateTime};
use crate::lookups::{self, Cell, Load, Lookup, Table};
use crate::packages::{DispatchOrder, Package};
use minijinja::value::{Enumerator, Kwargs, Object, ObjectRepr, Value};
use minijinja::{Environment, Error, ErrorKind, UndefinedBehavior};
use serde_json::{Map as JsonMap, Value as Json};
const MACROS: &str = "__dre_macros__";
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum Mode {
#[default]
Run,
Parse,
}
#[derive(Debug, Clone)]
pub struct RunContext {
pub report: String,
pub set: Option<String>,
pub target: String,
pub schedule: Option<String>,
pub date: NaiveDate,
pub now: chrono::DateTime<Utc>,
pub scheduled_at: Option<chrono::DateTime<Utc>>,
pub calendar: Calendar,
pub locale: crate::numbers::Locale,
}
const MARKDOWN_SUFFIX: &str = "\u{0}md";
#[derive(Debug, Clone, Default)]
pub struct QueryRows {
pub columns: Vec<String>,
pub rows: Vec<Vec<Value>>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Column {
pub name: String,
pub data_type: String,
}
pub trait QueryRunner: Send + Sync {
fn run_query(&self, sql: &str, max_rows: u64, profile: Option<&str>) -> Result<QueryRows, String>;
fn columns(&self, sql: &str, profile: Option<&str>) -> Result<Vec<Column>, String>;
fn load(&self, _name: &str, _table: &Table) -> Result<Option<(String, Option<String>)>, String> {
Ok(None)
}
}
#[derive(Debug, Clone, Default)]
pub struct Connection {
pub profile: String,
pub target: String,
pub kind: String,
pub fields: JsonMap<String, Json>,
pub secrets: Vec<String>,
}
pub trait Connections: Send + Sync {
fn profile(&self, name: &str, role: Option<&str>) -> Result<Connection, String>;
fn identifier_quote(&self, kind: &str) -> Result<Option<String>, String>;
fn string_literal(&self, _kind: &str) -> Result<Option<String>, String> {
Ok(None)
}
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Serialize)]
pub struct Quoting {
pub database: bool,
pub schema: bool,
pub identifier: bool,
}
impl Quoting {
pub fn any(&self) -> bool {
self.database || self.schema || self.identifier
}
}
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
pub struct ResolvedSource {
pub source: String,
pub table: String,
pub profile: Option<String>,
pub database: Option<String>,
pub schema: String,
pub identifier: String,
pub quoting: Quoting,
}
impl ResolvedSource {
pub fn relation(&self, quote: Option<&str>) -> String {
let part = |s: &str, q: bool| match (q, quote) {
(true, Some(c)) => format!("{c}{}{c}", s.replace(c, &format!("{c}{c}"))),
_ => s.to_string(),
};
let mut parts = Vec::new();
if let Some(d) = &self.database {
parts.push(part(d, self.quoting.database));
}
parts.push(part(&self.schema, self.quoting.schema));
parts.push(part(&self.identifier, self.quoting.identifier));
parts.join(".")
}
pub fn key(&self) -> String {
format!("{}.{}", self.source, self.table)
}
}
pub type SourceResolver = Arc<dyn Fn(&str, &str) -> Result<ResolvedSource, String> + Send + Sync>;
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RenderError {
pub file: PathBuf,
pub line: Option<usize>,
pub message: String,
}
impl fmt::Display for RenderError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self.line {
Some(l) => write!(f, "{}:{l}: {}", self.file.display(), self.message),
None => write!(f, "{}: {}", self.file.display(), self.message),
}
}
}
type ColumnsCache = BTreeMap<(String, Option<String>), Value>;
pub struct Renderer {
env: Environment<'static>,
raised: Arc<Mutex<Option<String>>>,
columns_cache: Arc<Mutex<ColumnsCache>>,
source_calls: Arc<Mutex<Vec<(String, String)>>>,
destination: Arc<Mutex<Option<Connection>>>,
warnings: Arc<Mutex<Vec<String>>>,
import: String,
macro_files: BTreeMap<String, Vec<(usize, PathBuf, usize)>>,
}
pub struct RendererConfig<'a> {
pub root: &'a Path,
pub macros: &'a [PathBuf],
pub context: RunContext,
pub vars: JsonMap<String, Json>,
pub cli_vars: BTreeMap<String, String>,
pub runner: Option<Arc<dyn QueryRunner>>,
pub connections: Option<Arc<dyn Connections>>,
pub mode: Mode,
pub connection: Option<String>,
pub source_type: String,
pub sources: Option<SourceResolver>,
pub run_query_max_rows: u64,
pub sql: BTreeMap<String, PathBuf>,
pub lookups: BTreeMap<String, Lookup>,
pub lookup_inline_max_rows: u64,
pub packages: Vec<Package>,
pub project_name: String,
pub dispatch: DispatchOrder,
}
impl Renderer {
pub fn new(cfg: RendererConfig<'_>) -> Result<Renderer, RenderError> {
let mut env = Environment::new();
let parse = cfg.mode == Mode::Parse;
env.set_undefined_behavior(if parse {
UndefinedBehavior::Chainable
} else {
UndefinedBehavior::Strict
});
env.set_syntax(
minijinja::syntax::SyntaxConfig::builder()
.keep_trailing_newline(true)
.build()
.expect("default delimiters are valid"),
);
env.set_auto_escape_callback(|name| {
if name.ends_with(MARKDOWN_SUFFIX) {
minijinja::AutoEscape::Custom("markdown".into())
} else {
minijinja::AutoEscape::None
}
});
env.set_formatter(|out, state, value| {
if *state.auto_escape() == minijinja::AutoEscape::Custom("markdown".into()) {
let text = value.to_string();
let text = if value.is_safe() {
text
} else {
dre_protocol::markdown::escape(&text)
};
return out.write_str(&text).map_err(Error::from);
}
minijinja::escape_formatter(out, state, value)
});
crate::mutable::register(&mut env);
add_vars(&mut env, cfg.vars, cfg.cli_vars);
let raised: Arc<Mutex<Option<String>>> = Arc::default();
let raised_c = raised.clone();
env.add_function("raise_error", move |message: String| -> Result<Value, Error> {
if parse {
raised_c.lock().unwrap().get_or_insert(message);
return Ok(Value::from(""));
}
Err(Error::new(ErrorKind::InvalidOperation, message))
});
let source_type = cfg.source_type.clone();
crate::dates::register(&mut env, cfg.context.calendar, cfg.context.date);
crate::numbers::register(&mut env, cfg.context.locale);
env.add_global("target", Value::from_object(Target(cfg.context.target.clone())));
env.add_global("run", Value::from_object(Run(cfg.context)));
let destination: Arc<Mutex<Option<Connection>>> = Arc::default();
if parse {
env.add_global("connection", Value::from_object(Lenient));
env.add_global("destination", Value::from_object(Lenient));
env.add_function(
"profile",
|_name: String, kwargs: Kwargs| -> Result<Value, Error> {
let role: Option<String> = kwargs.get("role")?;
kwargs.assert_all_used()?;
check_role(role.as_deref())?;
Ok(Value::from_object(Lenient))
},
);
} else {
env.add_global(
"connection",
Value::from_object(LazyConnection {
connections: cfg.connections.clone(),
name: cfg.connection.clone(),
resolved: std::sync::OnceLock::new(),
}),
);
env.add_global(
"destination",
Value::from_object(DestinationSlot(destination.clone())),
);
let c = cfg.connections.clone();
env.add_function(
"profile",
move |name: String, kwargs: Kwargs| -> Result<Value, Error> {
let role: Option<String> = kwargs.get("role")?;
kwargs.assert_all_used()?;
check_role(role.as_deref())?;
let Some(c) = &c else {
return Err(Error::new(
ErrorKind::InvalidOperation,
format!("`profile('{name}')`: no profiles.yml is loaded here"),
));
};
c.profile(&name, role.as_deref())
.map(|t| Value::from_object(ConnectionValue(t)))
.map_err(|e| {
Error::new(ErrorKind::InvalidOperation, format!("`profile('{name}')`: {e}"))
})
},
);
}
let source_calls: Arc<Mutex<Vec<(String, String)>>> = Arc::default();
let relations: Arc<Mutex<Vec<(String, ResolvedSource)>>> = Arc::default();
{
let calls = source_calls.clone();
let relations = relations.clone();
let resolver = cfg.sources.clone();
let connections = cfg.connections.clone();
let kind = cfg.source_type.clone();
let quote: Arc<std::sync::OnceLock<Result<Option<String>, String>>> = Arc::default();
env.add_function(
"source",
move |source: String, table: String| -> Result<Value, Error> {
let fail = |e: String| {
Error::new(
ErrorKind::InvalidOperation,
format!("`source('{source}', '{table}')`: {e}"),
)
};
let Some(resolve) = &resolver else {
return Err(fail("the project declares no `sources:`".into()));
};
let r = resolve(&source, &table).map_err(fail)?;
calls.lock().unwrap().push((source.clone(), table.clone()));
let q = if parse || !r.quoting.any() {
None
} else {
let got = quote.get_or_init(|| match &connections {
Some(c) if !kind.is_empty() => c.identifier_quote(&kind),
_ => Err("no connection is known here to quote with".into()),
});
match got {
Ok(Some(q)) => Some(q.clone()),
Ok(None) => {
return Err(fail(format!(
"`quoting` needs the identifier quote character, but the `{kind}` plugin doesn't report one; update it (`dre plugin update {kind}`)"
)));
}
Err(e) => return Err(fail(format!("can't quote: {e}"))),
}
};
let text = r.relation(q.as_deref());
relations.lock().unwrap().push((text.clone(), r));
Ok(Value::from(text))
},
);
}
let runner = cfg.runner;
let warnings: Arc<Mutex<Vec<String>>> = Arc::default();
let lookups = Arc::new(Lookups {
connections: cfg.connections.clone(),
kind: cfg.source_type.clone(),
style: std::sync::OnceLock::new(),
root: cfg.root.to_path_buf(),
defs: cfg.lookups,
inline_max: cfg.lookup_inline_max_rows,
runner: runner.clone(),
tables: Mutex::default(),
loaded: Mutex::default(),
warnings: warnings.clone(),
});
let l = lookups.clone();
env.add_function("lookup", move |name: String| -> Result<Value, Error> {
let t = l.table(&name)?;
let rows = QueryRows {
columns: t.columns.clone(),
rows: t
.rows
.iter()
.map(|r| r.iter().map(cell_value).collect())
.collect(),
};
Ok(Value::from_object(QueryResult::new(rows)))
});
let default_max = cfg.run_query_max_rows;
let runner_c = runner.clone();
let rels = relations.clone();
env.add_function(
"run_query",
move |sql: String, kwargs: Kwargs| -> Result<Value, Error> {
let max_rows: Option<u64> = kwargs.get("max_rows")?;
let profile: Option<String> = kwargs.get("profile")?;
kwargs.assert_all_used()?;
if parse {
let mut empty = QueryResult::new(QueryRows::default());
empty.lenient = true;
return Ok(Value::from_object(empty));
}
let profile = choose_profile("run_query()", &sql, profile, &rels)?;
let Some(runner) = &runner else {
return Err(Error::new(
ErrorKind::InvalidOperation,
"`run_query()` has no connection here",
));
};
let rows = runner
.run_query(&sql, max_rows.unwrap_or(default_max), profile.as_deref())
.map_err(|e| Error::new(ErrorKind::InvalidOperation, e))?;
Ok(Value::from_object(QueryResult::new(rows)))
},
);
let columns_cache: Arc<Mutex<ColumnsCache>> = Arc::default();
let cache = columns_cache.clone();
let rels = relations.clone();
env.add_function(
"columns",
move |rel: String, kwargs: Kwargs| -> Result<Value, Error> {
let profile: Option<String> = kwargs.get("profile")?;
kwargs.assert_all_used()?;
if parse {
return Ok(Value::from(Vec::<Value>::new()));
}
let what = format!("columns('{}')", crate::secrets::mask(&rel));
let profile = choose_profile(&what, &rel, profile, &rels)?;
let key = (rel.clone(), profile.clone());
if let Some(v) = cache.lock().unwrap().get(&key) {
return Ok(v.clone());
}
let Some(runner) = &runner_c else {
return Err(Error::new(
ErrorKind::InvalidOperation,
format!("`{what}` has no connection here"),
));
};
let sql = format!("select * from {rel} as _dre_cols where 1=0");
let cols = runner
.columns(&sql, profile.as_deref())
.map_err(|e| Error::new(ErrorKind::InvalidOperation, format!("`{what}` failed: {e}")))?;
let v = Value::from(
cols.into_iter()
.map(|c| {
Value::from_pairs([
("name", Value::from(c.name)),
("type", Value::from(c.data_type)),
])
})
.collect::<Vec<_>>(),
);
cache.lock().unwrap().insert(key, v.clone());
Ok(v)
},
);
let root = cfg.root.to_path_buf();
let display = |p: &Path| {
p.strip_prefix(&root)
.map(Path::to_path_buf)
.unwrap_or_else(|_| p.to_path_buf())
};
let own: Vec<(PathBuf, PathBuf)> = cfg.macros.iter().map(|m| (cfg.root.join(m), m.clone())).collect();
let (combined, names, spans) = combine(&own)?;
let mut import = if names.is_empty() {
String::new()
} else {
format!("{{% from \"{MACROS}\" import {} %}}", names.join(", "))
};
let mut templates = vec![(MACROS.to_string(), combined, spans, !names.is_empty())];
for p in &cfg.packages {
let files: Vec<(PathBuf, PathBuf)> = p.macros.iter().map(|m| (m.clone(), display(m))).collect();
let (src, _, spans) = combine(&files)?;
let t = package_template(&p.name);
import.push_str(&format!("{{% import \"{t}\" as {} %}}", p.name));
templates.push((t, src, spans, true));
}
add_ref(&mut env, cfg.root, cfg.sql, lookups, &import);
add_dispatch(
&mut env,
&cfg.project_name,
!names.is_empty(),
&cfg.packages,
cfg.dispatch,
source_type,
parse,
);
let mut r = Renderer {
env,
raised,
columns_cache,
source_calls,
destination,
warnings,
import,
macro_files: BTreeMap::new(),
};
for (name, src, spans, add) in templates {
r.macro_files.insert(name.clone(), spans);
if add {
r.env
.add_template_owned(name.clone(), src)
.map_err(|e| r.error(Path::new(&name), &e))?;
}
}
Ok(r)
}
pub fn take_warnings(&self) -> Vec<String> {
std::mem::take(&mut *self.warnings.lock().unwrap())
}
pub fn take_sources(&self) -> Vec<(String, String)> {
let mut out: Vec<(String, String)> = Vec::new();
for c in std::mem::take(&mut *self.source_calls.lock().unwrap()) {
if !out.contains(&c) {
out.push(c);
}
}
out
}
pub fn set_destination(&self, d: Option<Connection>) {
*self.destination.lock().unwrap() = d;
}
pub fn render(&self, file: &Path, src: &str) -> Result<String, RenderError> {
self.render_with(file, src, Value::UNDEFINED, false)
}
pub fn render_with(
&self,
file: &Path,
src: &str,
ctx: Value,
markdown: bool,
) -> Result<String, RenderError> {
self.columns_cache.lock().unwrap().clear();
*self.raised.lock().unwrap() = None;
let full = format!("{}{src}", self.import);
let mut name = file.to_string_lossy().to_string();
if markdown {
name.push_str(MARKDOWN_SUFFIX);
}
let tmpl = self
.env
.template_from_named_str(&name, &full)
.map_err(|e| self.error(file, &e))?;
let rendered = if ctx.is_undefined() {
tmpl.render(())
} else {
tmpl.render(ctx)
};
rendered.map_err(|e| {
let mut err = self.error(file, &e);
if let Some(m) = self.raised.lock().unwrap().take() {
err.message = m;
}
err
})
}
fn error(&self, file: &Path, e: &Error) -> RenderError {
let mut deepest: &Error = e;
while let Some(src) = std::error::Error::source(deepest).and_then(|s| s.downcast_ref::<Error>()) {
deepest = src;
}
let (name, line) = match (deepest.name(), deepest.line()) {
(Some(n), l) => (n.to_string(), l),
_ => (e.name().unwrap_or_default().to_string(), e.line()),
};
let (file, line) = if let Some(spans) = self.macro_files.get(&name) {
match line.and_then(|l| {
spans
.iter()
.find(|(s, _, n)| l >= *s && l < s + n)
.map(|(s, f, _)| (f, l - s + 1))
}) {
Some((f, l)) => (f.clone(), Some(l)),
None => (file.to_path_buf(), line),
}
} else {
(file.to_path_buf(), line)
};
let message = deepest
.detail()
.map(str::to_string)
.unwrap_or_else(|| deepest.kind().to_string());
RenderError { file, line, message }
}
}
pub fn cli_var_value(s: &str) -> Json {
let t = s.trim();
let int = regex_lite(t, |c, i| c.is_ascii_digit() || (i == 0 && (c == '-' || c == '+')));
match t {
"" => return Json::String(String::new()),
"true" | "True" | "TRUE" => return Json::Bool(true),
"false" | "False" | "FALSE" => return Json::Bool(false),
"null" | "Null" | "NULL" | "~" => return Json::Null,
_ => {}
}
if int {
let digits = t.trim_start_matches(['-', '+']);
if !digits.is_empty()
&& !(digits.len() > 1 && digits.starts_with('0'))
&& let Ok(n) = t.parse::<i64>()
{
return Json::from(n);
}
return Json::String(s.to_string());
}
let float = t.trim_start_matches(['-', '+']);
let is_float = !float.is_empty()
&& float
.chars()
.all(|c| c.is_ascii_digit() || matches!(c, '.' | 'e' | 'E' | '-' | '+'))
&& float
.chars()
.next()
.is_some_and(|c| c.is_ascii_digit() || c == '.')
&& float.chars().filter(|c| *c == '.').count() <= 1
&& !float.starts_with("00")
&& !(float.starts_with('0') && float.len() > 1 && float.as_bytes()[1].is_ascii_digit());
if is_float
&& (float.contains('.') || float.contains(['e', 'E']))
&& let Ok(f) = t.parse::<f64>()
&& let Some(n) = serde_json::Number::from_f64(f)
{
return Json::Number(n);
}
let quoted = (t.starts_with('"') && t.ends_with('"') && t.len() >= 2)
|| (t.starts_with('\'') && t.ends_with('\'') && t.len() >= 2);
let flow = (t.starts_with('[') && t.ends_with(']')) || (t.starts_with('{') && t.ends_with('}'));
if (quoted || flow)
&& let Ok(v) = serde_saphyr::from_str::<Json>(t)
{
return v;
}
Json::String(s.to_string())
}
fn regex_lite(s: &str, ok: impl Fn(char, usize) -> bool) -> bool {
!s.is_empty() && s.chars().enumerate().all(|(i, c)| ok(c, i))
}
fn add_vars(env: &mut Environment<'static>, vars: JsonMap<String, Json>, cli: BTreeMap<String, String>) {
env.add_function(
"var",
move |name: String, default: Option<Value>| -> Result<Value, Error> {
if let Some(v) = cli.get(&name) {
return Ok(Value::from(minijinja::value::Serde(cli_var_value(v))));
}
if let Some(v) = vars.get(&name) {
return Ok(Value::from(minijinja::value::Serde(v)));
}
default.ok_or_else(|| {
Error::new(
ErrorKind::UndefinedError,
format!("`var('{name}')` has no value and no default"),
)
})
},
);
env.add_function("env_var", |name: String, default: Option<Value>| -> Result<Value, Error> {
match std::env::var(&name) {
Ok(v) => Ok(Value::from(v)),
Err(_) => default.ok_or_else(|| {
Error::new(
ErrorKind::UndefinedError,
format!("`env_var('{name}')`: environment variable `{name}` is not set and no default is given"),
)
}),
}
});
}
pub fn add_common(
env: &mut Environment<'static>,
context: RunContext,
vars: JsonMap<String, Json>,
cli_vars: BTreeMap<String, String>,
) {
crate::mutable::register(env);
add_vars(env, vars, cli_vars);
env.add_function(
"env_var",
|name: String, _default: Option<Value>| -> Result<Value, Error> {
Err(Error::new(
ErrorKind::InvalidOperation,
format!("`env_var('{name}')` can't be used in a document template; pass the value as a var"),
))
},
);
env.add_function("raise_error", |message: String| -> Result<Value, Error> {
Err(Error::new(ErrorKind::InvalidOperation, message))
});
crate::dates::register(env, context.calendar, context.date);
crate::numbers::register(env, context.locale);
env.add_global("target", Value::from_object(Target(context.target.clone())));
env.add_global("run", Value::from_object(Run(context)));
}
pub struct Limited {
env: Environment<'static>,
}
const LIMITED_FUNCTIONS: &[&str] = &[
"run_query",
"columns",
"source",
"ref",
"lookup",
"dispatch",
"profile",
];
const LIMITED_GLOBALS: &[&str] = &["connection", "destination"];
impl Limited {
pub fn new(
context: RunContext,
vars: JsonMap<String, Json>,
cli_vars: BTreeMap<String, String>,
) -> Limited {
let mut env = Environment::new();
env.set_undefined_behavior(UndefinedBehavior::Strict);
crate::mutable::register(&mut env);
add_vars(&mut env, vars, cli_vars);
env.add_function("raise_error", |message: String| -> Result<Value, Error> {
Err(Error::new(ErrorKind::InvalidOperation, message))
});
crate::dates::register(&mut env, context.calendar, context.date);
crate::numbers::register(&mut env, context.locale);
env.add_global("target", Value::from_object(Target(context.target.clone())));
env.add_global("run", Value::from_object(Run(context)));
for f in LIMITED_FUNCTIONS {
let name = *f;
env.add_function(name, move |_args: minijinja::value::Rest<Value>| -> Result<Value, Error> {
Err(Error::new(
ErrorKind::InvalidOperation,
format!("`{name}()` can't be used here: this value chooses a connection or a source, so it's rendered before any connection is open"),
))
});
}
for g in LIMITED_GLOBALS {
env.add_global(*g, Value::from_object(Forbidden(g)));
}
Limited { env }
}
pub fn render(&self, key: &str, src: &str) -> Result<String, String> {
if !crate::preflight::is_templated(src) {
return Ok(src.to_string());
}
self.env
.render_str(src, ())
.map(|v| v.trim().to_string())
.map_err(|e| {
let mut deepest: &Error = &e;
while let Some(s) = std::error::Error::source(deepest).and_then(|s| s.downcast_ref::<Error>())
{
deepest = s;
}
let msg = deepest
.detail()
.map(str::to_string)
.unwrap_or_else(|| deepest.kind().to_string());
format!("{key} `{src}`: {msg}")
})
}
}
#[derive(Debug)]
struct Forbidden(&'static str);
impl Object for Forbidden {
fn get_value(self: &Arc<Self>, _key: &Value) -> Option<Value> {
Some(Value::from(Error::new(
ErrorKind::InvalidOperation,
format!(
"`{}.*` can't be used here: this value chooses a connection or a source, so it's rendered before any connection is open",
self.0
),
)))
}
fn render(self: &Arc<Self>, _f: &mut fmt::Formatter<'_>) -> fmt::Result {
Err(fmt::Error)
}
}
fn package_template(name: &str) -> String {
format!("__dre_package_{name}__")
}
#[allow(clippy::type_complexity)]
fn combine(
files: &[(PathBuf, PathBuf)],
) -> Result<(String, Vec<String>, Vec<(usize, PathBuf, usize)>), RenderError> {
let re = regex::Regex::new(r"\{%-?\s*macro\s+([A-Za-z_]\w*)").unwrap();
let mut combined = String::new();
let mut names = Vec::new();
let mut spans = Vec::new();
for (read, shown) in files {
let src = std::fs::read_to_string(read).map_err(|e| RenderError {
file: shown.clone(),
line: None,
message: format!("can't read macro file: {e}"),
})?;
let start = combined.matches('\n').count() + 1;
names.extend(re.captures_iter(&src).map(|c| c[1].to_string()));
combined.push_str(&src);
if !combined.ends_with('\n') {
combined.push('\n');
}
spans.push((start, shown.clone(), src.matches('\n').count() + 1));
}
Ok((combined, names, spans))
}
fn add_dispatch(
env: &mut Environment<'static>,
project: &str,
project_has_macros: bool,
packages: &[Package],
order: DispatchOrder,
source_type: String,
parse: bool,
) {
let project = project.to_string();
let templates: BTreeMap<String, String> = packages
.iter()
.map(|p| (p.name.clone(), package_template(&p.name)))
.collect();
env.add_function(
"dispatch",
move |state: &minijinja::State<'_, '_>,
name: String,
namespace: Option<String>|
-> Result<Value, Error> {
let ns = namespace.unwrap_or_else(|| project.clone());
let search = order.get(&ns).cloned().unwrap_or_else(|| {
if ns == project {
vec![ns.clone()]
} else {
vec![project.clone(), ns.clone()]
}
});
let candidates = [format!("{source_type}__{name}"), format!("default__{name}")];
for n in &search {
let template = if n == &project {
if !project_has_macros {
continue;
}
MACROS.to_string()
} else {
match templates.get(n) {
Some(t) => t.clone(),
None => continue,
}
};
let tmpl = state.env().get_template(&template)?;
let captured = tmpl.render_captured(())?;
let st = captured.state();
for c in &candidates {
if st.lookup(c).is_some_and(|v| !v.is_undefined()) {
return Ok(Value::from_object(Dispatched {
template,
name: c.clone(),
}));
}
}
}
if parse {
return Ok(Value::from_object(Dispatched {
template: String::new(),
name: String::new(),
}));
}
Err(Error::new(
ErrorKind::InvalidOperation,
format!(
"`dispatch('{name}', '{ns}')`: no `{}` or `{}` in {}",
candidates[0],
candidates[1],
search.join(", ")
),
))
},
);
}
#[derive(Debug)]
struct Dispatched {
template: String,
name: String,
}
impl Object for Dispatched {
fn call(self: &Arc<Self>, state: &mut minijinja::State<'_, '_>, args: &[Value]) -> Result<Value, Error> {
if self.template.is_empty() {
return Ok(Value::from(""));
}
let tmpl = state.env().get_template(&self.template)?;
let out = tmpl
.render_captured(())?
.with_state_mut(|s| s.call_macro(&self.name, args))?;
Ok(Value::from(out))
}
}
fn add_ref(
env: &mut Environment<'static>,
root: &Path,
sql: BTreeMap<String, PathBuf>,
lookups: Arc<Lookups>,
import: &str,
) {
let root = root.to_path_buf();
let import = import.to_string();
let stack: Arc<Mutex<Vec<String>>> = Arc::default();
env.add_function(
"ref",
move |state: &minijinja::State<'_, '_>, name: String| -> Result<Value, Error> {
if lookups.defs.contains_key(&name) {
return lookups.relation(&name).map(Value::from);
}
let Some(rel) = sql.get(&name) else {
return Err(Error::new(
ErrorKind::InvalidOperation,
format!("`ref('{name}')`: no `{name}.sql` and no lookup `{name}` in the project"),
));
};
{
let mut chain = stack.lock().unwrap();
if chain.contains(&name) {
chain.push(name.clone());
let cycle = chain.join(" → ");
chain.clear();
return Err(Error::new(
ErrorKind::InvalidOperation,
format!("`ref()` cycle: {cycle}"),
));
}
chain.push(name.clone());
}
let result = (|| {
let src = std::fs::read_to_string(root.join(rel)).map_err(|e| {
Error::new(
ErrorKind::InvalidOperation,
format!("can't read {}: {e}", rel.display()),
)
})?;
let file = rel.to_string_lossy().to_string();
let rendered = state
.env()
.template_from_named_str(&file, &format!("{import}{src}"))?
.render(())?;
let statements = crate::sqlsplit::split(&rendered);
match statements.as_slice() {
[one] => Ok(Value::from(format!("(\n{}\n)", one.text))),
_ => Err(Error::new(
ErrorKind::InvalidOperation,
format!(
"`ref('{name}')` needs {} to hold exactly one statement, but it has {}",
rel.display(),
statements.len()
),
)),
}
})();
let mut chain = stack.lock().unwrap();
if chain.last() == Some(&name) {
chain.pop();
}
result
},
);
}
struct Lookups {
connections: Option<Arc<dyn Connections>>,
kind: String,
style: std::sync::OnceLock<Result<lookups::StringLiteral, String>>,
root: PathBuf,
defs: BTreeMap<String, Lookup>,
inline_max: u64,
runner: Option<Arc<dyn QueryRunner>>,
tables: Mutex<BTreeMap<String, Arc<Table>>>,
loaded: Mutex<BTreeMap<String, String>>,
warnings: Arc<Mutex<Vec<String>>>,
}
impl Lookups {
fn table(&self, name: &str) -> Result<Arc<Table>, Error> {
let def = self.defs.get(name).ok_or_else(|| {
Error::new(
ErrorKind::InvalidOperation,
format!("no lookup `{name}` under lookups/"),
)
})?;
if let Some(t) = self.tables.lock().unwrap().get(name) {
return Ok(t.clone());
}
let t = Arc::new(lookups::read(&self.root, def).map_err(|e| {
Error::new(
ErrorKind::InvalidOperation,
format!("{}: {e}", def.file.display()),
)
})?);
self.tables.lock().unwrap().insert(name.to_string(), t.clone());
Ok(t)
}
fn relation(&self, name: &str) -> Result<String, Error> {
if let Some(r) = self.loaded.lock().unwrap().get(name) {
return Ok(r.clone());
}
let t = self.table(name)?;
let rows = t.rows.len() as u64;
let load = match self.defs[name].load {
Load::Inline => false,
Load::TempTable => true,
Load::Auto => rows > self.inline_max,
};
let loaded = match (&self.runner, load) {
(Some(runner), true) => runner.load(name, &t).map_err(|e| {
Error::new(
ErrorKind::InvalidOperation,
format!("loading lookup `{name}`: {e}"),
)
})?,
_ => None,
};
let relation = match loaded {
Some((relation, warning)) => {
if let Some(w) = warning {
self.warn(format!("lookup `{name}`: {w}"));
}
relation
}
None => {
if load && self.runner.is_some() {
self.warn(format!(
"lookup `{name}` has {rows} rows; this source can't load it into a temp table, so it's inlined in the SQL. Data this size probably belongs in a table in the database"
));
}
t.inline_sql(self.style()?)
}
};
self.loaded
.lock()
.unwrap()
.insert(name.to_string(), relation.clone());
Ok(relation)
}
fn warn(&self, w: String) {
self.warnings.lock().unwrap().push(w);
}
fn style(&self) -> Result<lookups::StringLiteral, Error> {
self.style
.get_or_init(|| match &self.connections {
Some(c) if !self.kind.is_empty() => {
let asked = c.string_literal(&self.kind).unwrap_or(None);
lookups::StringLiteral::from_describe(asked.as_deref())
}
_ => Ok(lookups::StringLiteral::Chr),
})
.clone()
.map_err(|e| {
Error::new(
ErrorKind::InvalidOperation,
format!("the `{}` source: {e}", self.kind),
)
})
}
}
fn cell_value(c: &Cell) -> Value {
match c {
Cell::Null => Value::from(()),
Cell::Text(s) => Value::from(s.clone()),
Cell::Int(n) => Value::from(*n),
Cell::Num(n) => Value::from(*n),
Cell::Bool(b) => Value::from(*b),
Cell::Date(d) => Value::from(d.to_string()),
}
}
#[derive(Debug)]
struct Run(RunContext);
impl Object for Run {
fn get_value(self: &Arc<Self>, key: &Value) -> Option<Value> {
let c = &self.0;
Some(match key.as_str()? {
"report" => Value::from(c.report.clone()),
"set" => c.set.clone().map(Value::from).unwrap_or(Value::from(())),
"target" => Value::from(c.target.clone()),
"profile" => removed("run.profile", "`connection.name` (the query's connection)"),
"source_type" => removed("run.source_type", "`connection.type` (the query's connection)"),
"schedule" => c.schedule.clone().map(Value::from).unwrap_or(Value::from(())),
"date" => Date::value(c.date, c.calendar),
"now" => DateTime::now(c.now, c.calendar),
"scheduled_at" => c
.scheduled_at
.map(|t| DateTime::now(t, c.calendar))
.unwrap_or(Value::from(())),
"timezone" => Value::from(c.calendar.tz.name()),
_ => return None,
})
}
fn call_method(
self: &Arc<Self>,
_: &mut minijinja::State<'_, '_>,
method: &str,
args: &[Value],
) -> Result<Value, Error> {
match method {
"date_format" => {
let fmt = args.first().and_then(|a| a.as_str()).ok_or_else(|| {
Error::new(
ErrorKind::InvalidOperation,
"`run.date_format()` takes a strftime format string",
)
})?;
let mut out = String::new();
use std::fmt::Write;
write!(out, "{}", self.0.date.format(fmt)).map_err(|_| {
Error::new(
ErrorKind::InvalidOperation,
format!("invalid date format `{fmt}`"),
)
})?;
Ok(Value::from(out))
}
_ => Err(Error::from(ErrorKind::UnknownMethod)),
}
}
}
fn removed(name: &str, instead: &str) -> Value {
Value::from(Error::new(
ErrorKind::InvalidOperation,
format!("`{name}` was removed in DRE 0.2: use {instead}"),
))
}
fn check_role(role: Option<&str>) -> Result<(), Error> {
match role {
None | Some("connection") | Some("destination") => Ok(()),
Some("source") => Err(Error::new(
ErrorKind::InvalidOperation,
"`role='source'` was renamed in DRE 0.2: use `role='connection'`",
)),
Some(r) => Err(Error::new(
ErrorKind::InvalidOperation,
format!("`role='{r}'`: role must be 'connection' or 'destination'"),
)),
}
}
fn choose_profile(
what: &str,
sql: &str,
explicit: Option<String>,
relations: &Mutex<Vec<(String, ResolvedSource)>>,
) -> Result<Option<String>, Error> {
let mut used: Vec<(String, String)> = Vec::new();
for (text, r) in relations.lock().unwrap().iter() {
if let Some(p) = &r.profile
&& mentions(sql, text)
&& !used.iter().any(|(k, _)| *k == r.key())
{
used.push((r.key(), p.clone()));
}
}
let fail = |m: String| Error::new(ErrorKind::InvalidOperation, format!("`{what}`: {m}"));
match explicit {
Some(p) => {
if let Some((k, q)) = used.iter().find(|(_, q)| *q != p) {
return Err(fail(format!(
"`profile='{p}'` disagrees with source `{k}`, which is on connection `{q}`"
)));
}
Ok(Some(p))
}
None => match used.as_slice() {
[] => Ok(None),
[(_, p), rest @ ..] => {
if let Some((k, q)) = rest.iter().find(|(_, q)| q != p) {
return Err(fail(format!(
"reads source `{}` on connection `{p}` and source `{k}` on `{q}`; one statement runs on one connection",
used[0].0
)));
}
Ok(Some(p.clone()))
}
},
}
}
fn mentions(sql: &str, text: &str) -> bool {
let ident = |c: char| c.is_alphanumeric() || c == '_';
sql.match_indices(text).any(|(i, _)| {
let before = sql[..i].chars().next_back();
let after = sql[i + text.len()..].chars().next();
!before.is_some_and(|c| ident(c) || c == '.') && !after.is_some_and(|c| ident(c) || c == '.')
})
}
#[derive(Debug)]
struct Target(String);
impl Object for Target {
fn get_value(self: &Arc<Self>, key: &Value) -> Option<Value> {
let k = key.as_str()?;
Some(match k {
"name" => Value::from(self.0.clone()),
"type" => removed("target.type", "`connection.type` (the query's connection)"),
"profile" => removed("target.profile", "`connection.name` (the query's connection)"),
_ => removed(
&format!("target.{k}"),
&format!(
"`connection.{k}` (the query's connection) or `profile('<name>').{k}`; `target` is now only the environment, `target.name`"
),
),
})
}
fn render(self: &Arc<Self>, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(&self.0)
}
}
#[derive(Debug)]
struct Lenient;
impl Object for Lenient {
fn get_value(self: &Arc<Self>, _key: &Value) -> Option<Value> {
None
}
fn render(self: &Arc<Self>, _f: &mut fmt::Formatter<'_>) -> fmt::Result {
Ok(())
}
}
#[derive(Debug)]
struct ConnectionValue(Connection);
impl Object for ConnectionValue {
fn get_value(self: &Arc<Self>, key: &Value) -> Option<Value> {
connection_field(&self.0, key)
}
fn render(self: &Arc<Self>, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "{}", self.0.profile)
}
}
struct LazyConnection {
connections: Option<Arc<dyn Connections>>,
name: Option<String>,
resolved: std::sync::OnceLock<Result<Connection, String>>,
}
impl fmt::Debug for LazyConnection {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str("connection")
}
}
impl LazyConnection {
fn get(&self) -> &Result<Connection, String> {
self.resolved.get_or_init(|| {
let name = self.name.as_deref().ok_or(
"no connection here: this isn't a query's SQL and the Binding inherits no `profile`",
)?;
match &self.connections {
Some(c) => c.profile(name, Some("connection")),
None => Err("no profiles.yml is loaded here".into()),
}
})
}
}
impl Object for LazyConnection {
fn get_value(self: &Arc<Self>, key: &Value) -> Option<Value> {
match self.get() {
Ok(c) => connection_field(c, key),
Err(e) => Some(Value::from(Error::new(
ErrorKind::InvalidOperation,
format!("`connection`: {e}"),
))),
}
}
fn render(self: &Arc<Self>, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match &self.name {
Some(n) => f.write_str(n),
None => Err(fmt::Error),
}
}
}
#[derive(Debug)]
struct DestinationSlot(Arc<Mutex<Option<Connection>>>);
impl Object for DestinationSlot {
fn get_value(self: &Arc<Self>, key: &Value) -> Option<Value> {
match &*self.0.lock().unwrap() {
Some(c) => connection_field(c, key),
None => Some(Value::from(Error::new(
ErrorKind::InvalidOperation,
"`destination` only exists while rendering a destination's `path` and options",
))),
}
}
fn render(self: &Arc<Self>, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match &*self.0.lock().unwrap() {
Some(c) => f.write_str(&c.profile),
None => Err(fmt::Error),
}
}
}
fn connection_field(c: &Connection, key: &Value) -> Option<Value> {
{
let k = key.as_str()?;
Some(match k {
"name" | "profile" => Value::from(c.profile.clone()),
"target" => Value::from(c.target.clone()),
"type" => Value::from(c.kind.clone()),
_ if c.secrets.iter().any(|s| s == k) => Value::from(Error::new(
ErrorKind::InvalidOperation,
format!(
"`{k}` of profile `{}` holds a secret, so templates can't read it (it would end up in compiled SQL and logs)",
c.profile
),
)),
_ => match c.fields.get(k) {
Some(v) => Value::from(minijinja::value::Serde(v)),
None => Value::from(Error::new(
ErrorKind::UndefinedError,
format!(
"profile `{}` (target `{}`) has no field `{k}`; it has: {}",
c.profile,
c.target,
c.fields
.keys()
.filter(|f| !c.secrets.contains(f))
.cloned()
.collect::<Vec<_>>()
.join(", ")
),
)),
},
})
}
}
#[derive(Debug)]
struct QueryResult {
columns: Vec<String>,
rows: Vec<Value>,
lenient: bool,
}
impl QueryResult {
fn new(q: QueryRows) -> QueryResult {
let cols: Arc<Vec<String>> = Arc::new(q.columns.clone());
let rows = q
.rows
.into_iter()
.map(|values| {
Value::from_object(Row {
columns: cols.clone(),
values,
})
})
.collect();
QueryResult {
columns: q.columns,
rows,
lenient: false,
}
}
}
impl Object for QueryResult {
fn repr(self: &Arc<Self>) -> ObjectRepr {
ObjectRepr::Seq
}
fn get_value(self: &Arc<Self>, key: &Value) -> Option<Value> {
match key.as_str() {
Some("rows") => Some(Value::from(self.rows.clone())),
Some("columns") => Some(Value::from(self.columns.clone())),
_ => self.rows.get(key.as_usize()?).cloned(),
}
}
fn enumerate(self: &Arc<Self>) -> Enumerator {
Enumerator::Seq(self.rows.len())
}
fn call_method(
self: &Arc<Self>,
_: &mut minijinja::State<'_, '_>,
method: &str,
args: &[Value],
) -> Result<Value, Error> {
if method != "column" {
return Err(Error::from(ErrorKind::UnknownMethod));
}
if self.lenient {
return Ok(Value::from(Vec::<Value>::new()));
}
let [key] = args else {
return Err(Error::new(
ErrorKind::InvalidOperation,
"`column()` takes one column name or index",
));
};
let i = match key.as_str() {
Some(n) => self.columns.iter().position(|c| c == n).ok_or_else(|| {
Error::new(
ErrorKind::InvalidOperation,
format!("no column `{n}` (columns: {})", self.columns.join(", ")),
)
})?,
None => key
.as_usize()
.filter(|i| *i < self.columns.len())
.ok_or_else(|| Error::new(ErrorKind::InvalidOperation, format!("no column {key}")))?,
};
Ok(Value::from(
self.rows
.iter()
.map(|r| r.get_item_by_index(i).unwrap_or_default())
.collect::<Vec<_>>(),
))
}
}
#[derive(Debug)]
pub(crate) struct Row {
columns: Arc<Vec<String>>,
values: Vec<Value>,
}
impl Row {
pub(crate) fn value(columns: Arc<Vec<String>>, values: Vec<Value>) -> Value {
Value::from_object(Row { columns, values })
}
}
impl Object for Row {
fn repr(self: &Arc<Self>) -> ObjectRepr {
ObjectRepr::Seq
}
fn get_value(self: &Arc<Self>, key: &Value) -> Option<Value> {
if let Some(name) = key.as_str() {
let i = self.columns.iter().position(|c| c == name)?;
return self.values.get(i).cloned();
}
self.values.get(key.as_usize()?).cloned()
}
fn enumerate(self: &Arc<Self>) -> Enumerator {
Enumerator::Seq(self.values.len())
}
}
#[cfg(test)]
mod tests {
use super::*;
fn renderer(vars: Json, cli: &[(&str, &str)], macros: &[(&str, &str)]) -> (tempfile::TempDir, Renderer) {
let dir = tempfile::tempdir().unwrap();
let mut files = Vec::new();
for (name, src) in macros {
let p = PathBuf::from("macros").join(name);
std::fs::create_dir_all(dir.path().join("macros")).unwrap();
std::fs::write(dir.path().join(&p), src).unwrap();
files.push(p);
}
let r = Renderer::new(RendererConfig {
root: dir.path(),
macros: &files,
context: RunContext {
report: "monthly".into(),
set: Some("client_a".into()),
target: "prod".into(),
schedule: None,
date: NaiveDate::from_ymd_opt(2026, 1, 25).unwrap(),
now: Utc::now(),
scheduled_at: None,
calendar: Calendar::default(),
locale: Default::default(),
},
vars: vars.as_object().cloned().unwrap_or_default(),
cli_vars: cli.iter().map(|(k, v)| (k.to_string(), v.to_string())).collect(),
runner: None,
connections: None,
mode: Mode::Run,
connection: None,
source_type: "duckdb".into(),
sources: None,
run_query_max_rows: 10_000,
sql: BTreeMap::new(),
lookups: BTreeMap::new(),
lookup_inline_max_rows: 200,
packages: Vec::new(),
project_name: "acme".into(),
dispatch: DispatchOrder::new(),
})
.unwrap();
(dir, r)
}
fn render(r: &Renderer, src: &str) -> Result<String, RenderError> {
r.render(Path::new("reports/q.sql"), src)
}
fn ok(src: &str) -> String {
let (_d, r) = renderer(serde_json::json!({"fixed": [3, 1, 2], "m": {"a": 1}}), &[], &[]);
render(&r, src).unwrap_or_else(|e| panic!("{src}: {e}"))
}
fn err(src: &str) -> String {
let (_d, r) = renderer(serde_json::json!({"fixed": [3, 1, 2], "m": {"a": 1}}), &[], &[]);
render(&r, src).unwrap_err().to_string()
}
#[test]
fn python_string_and_dict_methods_work() {
assert_eq!(ok("{{ 'a,b,c'.split(',') | join('-') }}"), "a-b-c");
assert_eq!(
ok("{{ 'y' if 'abc'.startswith('a') }}{{ 'y' if 'abc'.endswith('x') }}"),
"y"
);
assert_eq!(
ok("{{ ' x '.strip() }}|{{ 'abc'.upper() }}|{{ 'a-b'.replace('-', '_') }}"),
"x|ABC|a_b"
);
assert_eq!(
ok("{{ var('m').keys() | list }} {{ var('m').items() | list }}"),
"['a'] [('a', 1)]"
);
assert_eq!(
ok("{{ var('m').get('a') }} {{ var('m').get('z', 'none') }}"),
"1 none"
);
}
#[test]
fn list_appends_in_a_loop_and_joins() {
assert_eq!(
ok(
"{% set sq = list() %}{% for i in range(1, 5) %}{% set _ = sq.append(i * i) %}{% endfor %}{{ sq | join('+') }}"
),
"1+4+9+16"
);
}
#[test]
fn list_methods_follow_python() {
assert_eq!(
ok("{% set l = list([3, 1, 2]) %}{% set _ = l.sort() %}{{ l }}"),
"[1, 2, 3]"
);
assert_eq!(
ok("{% set l = list([3, 1, 2]) %}{% set _ = l.sort(reverse=true) %}{{ l }}"),
"[3, 2, 1]"
);
assert_eq!(
ok("{% set l = list([1, 2]) %}{% set _ = l.extend([3, 4]) %}{{ l }} {{ l | length }}"),
"[1, 2, 3, 4] 4"
);
assert_eq!(
ok("{% set l = list([1, 2, 3]) %}{{ l.pop() }} {{ l.pop(0) }} {{ l }}"),
"3 1 [2]"
);
assert_eq!(
ok("{% set l = list([1, 2]) %}{% set _ = l.insert(0, 9) %}{% set _ = l.insert(99, 7) %}{{ l }}"),
"[9, 1, 2, 7]"
);
assert_eq!(
ok(
"{% set l = list([1, 2, 1]) %}{% set _ = l.remove(1) %}{{ l }} {{ l.index(1) }} {{ l.count(1) }}"
),
"[2, 1] 1 1"
);
assert_eq!(
ok("{% set l = list([1, 2]) %}{% set _ = l.reverse() %}{{ l }}"),
"[2, 1]"
);
assert_eq!(
ok("{% set l = list([1, 2]) %}{% set _ = l.clear() %}{{ l }} {{ 'empty' if not l }}"),
"[] empty"
);
assert_eq!(
ok("{% set l = list([1, 2, 3]) %}{{ l[0] }} {{ l[-1] }} {{ l[1:] }}"),
"1 3 [2, 3]"
);
}
#[test]
fn lists_share_by_reference_and_copy_apart() {
assert_eq!(
ok("{% set a = list() %}{% set b = a %}{% set _ = b.append(1) %}{{ a }}"),
"[1]"
);
assert_eq!(
ok("{% set a = list([1]) %}{% set b = a.copy() %}{% set _ = b.append(2) %}{{ a }} {{ b }}"),
"[1] [1, 2]"
);
}
#[test]
fn a_list_changed_inside_its_own_loop_does_not_skip_or_repeat() {
assert_eq!(
ok(
"{% set l = list([1, 2, 3]) %}{% for x in l %}{% set _ = l.append(x * 10) %}{{ x }} {% endfor %}{{ l }}"
),
"1 2 3 [1, 2, 3, 10, 20, 30]"
);
assert_eq!(
ok("{% set l = list([1, 2]) %}{% set _ = l.extend(l) %}{{ l }}"),
"[1, 2, 1, 2]"
);
}
#[test]
fn a_list_starts_from_a_var_without_changing_it() {
assert_eq!(
ok("{% set l = list(var('fixed')) %}{% set _ = l.append(4) %}{{ l }} {{ var('fixed') }}"),
"[3, 1, 2, 4] [3, 1, 2]"
);
}
#[test]
fn dict_methods_follow_python() {
assert_eq!(
ok("{% set d = dict(a=1) %}{% set _ = d.update(b=2) %}{% set _ = d.update({'c': 3}) %}{{ d }}"),
"{'a': 1, 'b': 2, 'c': 3}"
);
let make = "{% set d = dict() %}{% set _ = d.setdefault('b', 1) %}{% set _ = d.setdefault('a', 2) %}";
assert_eq!(
ok(&(make.to_string() + "{% for k, v in d.items() %}{{ k }}={{ v }} {% endfor %}")),
"b=1 a=2 "
);
assert_eq!(
ok(&(make.to_string() + "{% for k in d %}{{ k }} {% endfor %}{{ d.values() | list }}")),
"b a [1, 2]"
);
assert_eq!(
ok("{% set d = dict(a=1) %}{{ d.setdefault('a', 9) }} {{ d.setdefault('b', 2) }} {{ d }}"),
"1 2 {'a': 1, 'b': 2}"
);
assert_eq!(
ok("{% set d = dict(a=1) %}{{ d.pop('a') }} {{ d.pop('a', 'gone') }} {{ d }}"),
"1 gone {}"
);
assert_eq!(
ok(
"{% set d = dict(a=1) %}{{ d.get('a') }} {{ d['a'] }} {{ d.a }} {{ 'in' if 'a' in d }} {{ d.keys() | list }}"
),
"1 1 1 in ['a']"
);
assert_eq!(
ok("{% set d = dict(a=1) %}{% set e = d.copy() %}{% set _ = e.update(a=2) %}{{ d.a }} {{ e.a }}"),
"1 2"
);
assert_eq!(
ok("{% set d = dict(a=1) %}{% set _ = d.update(a=5, b=6) %}{{ d | length }} {{ d.a }}"),
"2 5"
);
assert_eq!(
ok("{% set d = dict(a=1) %}{% set _ = d.update(d) %}{{ d }}"),
"{'a': 1}"
);
}
#[test]
fn mutable_values_print_and_serialise_like_plain_ones() {
assert_eq!(
ok("{{ list([1, 'a']) }} {{ dict(k='v') }}"),
"[1, 'a'] {'k': 'v'}"
);
assert_eq!(
ok("{{ list(dict(k='v')) }} {{ list(list([2, 1])) | sort }}"),
"['k'] [1, 2]"
);
assert_eq!(ok("{% set l = list() %}{{ 'yes' if l else 'no' }}"), "no");
}
#[test]
fn expressions_follow_jinja2() {
assert_eq!(
ok("{{ ['a', 1] }} {{ var('m') }} {{ var('m').items() | list }}"),
"['a', 1] {'a': 1} [('a', 1)]"
);
assert_eq!(ok("{{ 2.5 | round }} {{ 3.5 | round }}"), "2.0 4.0");
assert_eq!(ok("{{ -7 // 2 }} {{ -7 % 3 }}"), "-4 2");
assert_eq!(
ok("{{ ['b', 'P', 'n'] | max }} {{ \"don't\" | title }}"),
"P Don't"
);
assert!(err("{{ 1 / 0 }}").contains("unable to calculate 1 / 0"));
}
#[test]
fn mutating_an_immutable_value_says_what_to_write() {
let e = err("{% set l = [] %}{% set _ = l.append(1) %}");
assert!(e.contains("append()") && e.contains("list()"), "{e}");
let e = err("{% set _ = var('fixed').append(1) %}");
assert!(e.contains("list()"), "{e}");
let e = err("{% set d = {} %}{% set _ = d.update(a=1) %}");
assert!(e.contains("dict()"), "{e}");
}
#[test]
fn bad_arguments_to_mutable_methods_are_clear_errors() {
assert!(err("{% set l = list() %}{% set _ = l.append() %}").contains("append()"));
assert!(err("{% set l = list() %}{% set _ = l.pop() %}").contains("pop()"));
assert!(err("{% set l = list([1]) %}{% set _ = l.remove(2) %}").contains("not in the list"));
assert!(err("{% set d = dict() %}{% set _ = d.pop('x') %}").contains("no key"));
assert!(err("{% set d = dict('nope') %}").contains("mapping"));
}
#[test]
fn run_context_and_date_formats() {
let (_d, r) = renderer(Json::Null, &[], &[]);
let out = render(
&r,
"{{ run.report }}/{{ run.set }}/{{ run.target }}/{{ target.name }}/{{ target }} {{ run.date }} {{ run.date.yyyymmdd }} \
{{ run.date.ddmmyyyy }} {{ run.date.yyyy }}-{{ run.date.mm }}-{{ run.date.dd }} {{ run.date_format('%Y-W%V') }}",
)
.unwrap();
assert_eq!(
out,
"monthly/client_a/prod/prod/prod 2026-01-25 20260125 25012026 2026-01-25 2026-W04"
);
}
#[test]
fn var_precedence_is_cli_then_binding_then_default() {
let (_d, r) = renderer(
serde_json::json!({"a": "binding", "b": "binding", "n": 5}),
&[("a", "cli")],
&[],
);
assert_eq!(
render(
&r,
"{{ var('a') }} {{ var('b') }} {{ var('c', 'dflt') }} {{ var('n') + 1 }}"
)
.unwrap(),
"cli binding dflt 6"
);
let e = render(&r, "select\n{{ var('missing') }}").unwrap_err();
assert_eq!(
(e.line, e.message.contains("`var('missing')` has no value")),
(Some(2), true),
"{e}"
);
}
#[test]
fn env_var_reads_the_environment_or_errors() {
let (_d, r) = renderer(Json::Null, &[], &[]);
assert_eq!(
render(&r, "{{ env_var('DRE_TEST_UNSET_VAR', 'x') }}").unwrap(),
"x"
);
let home = std::env::var("PATH").unwrap();
assert_eq!(render(&r, "{{ env_var('PATH') }}").unwrap(), home);
assert!(
render(&r, "{{ env_var('DRE_TEST_UNSET_VAR') }}")
.unwrap_err()
.message
.contains("is not set")
);
}
#[test]
fn macros_from_every_file_are_callable_and_keep_line_numbers() {
let (_d, r) = renderer(
Json::Null,
&[],
&[
("a.sql", "{% macro double(x) %}{{ x * 2 }}{% endmacro %}\n"),
(
"b.sql",
"{% macro broken() %}\n{{ var('nope') }}\n{% endmacro %}\n{% macro quad(x) %}{{ x * 4 }}{% endmacro %}\n",
),
],
);
assert_eq!(
render(&r, "select {{ double(2) }}, {{ quad(1) }}\n").unwrap(),
"select 4, 4\n"
);
let e = render(&r, "line one\nselect {{ broken() }}").unwrap_err();
assert_eq!(e.file, PathBuf::from("macros/b.sql"), "{e}");
assert_eq!(e.line, Some(2), "{e}");
let e = render(&r, "one\ntwo\n{{ nope }}").unwrap_err();
assert_eq!(
(e.file.clone(), e.line),
(PathBuf::from("reports/q.sql"), Some(3)),
"{e}"
);
}
struct Fake;
impl QueryRunner for Fake {
fn run_query(&self, sql: &str, max_rows: u64, _profile: Option<&str>) -> Result<QueryRows, String> {
if max_rows < 2 {
return Err(format!("`{sql}` returned more than {max_rows} rows"));
}
Ok(QueryRows {
columns: vec!["region".into(), "n".into()],
rows: vec![
vec![Value::from("apac"), Value::from(1)],
vec![Value::from("emea"), Value::from(2)],
],
})
}
fn columns(&self, _: &str, _: Option<&str>) -> Result<Vec<Column>, String> {
Ok(Vec::new())
}
}
#[test]
fn run_query_rows_are_accessible_by_name_and_index() {
let dir = tempfile::tempdir().unwrap();
let r = Renderer::new(RendererConfig {
root: dir.path(),
macros: &[],
context: RunContext {
report: "r".into(),
set: None,
target: "t".into(),
schedule: None,
date: NaiveDate::from_ymd_opt(2026, 1, 1).unwrap(),
now: Utc::now(),
scheduled_at: None,
calendar: Calendar::default(),
locale: Default::default(),
},
vars: JsonMap::new(),
cli_vars: BTreeMap::new(),
runner: Some(Arc::new(Fake)),
connections: None,
mode: Mode::Run,
connection: None,
source_type: "duckdb".into(),
sources: None,
run_query_max_rows: 10_000,
sql: BTreeMap::new(),
lookups: BTreeMap::new(),
lookup_inline_max_rows: 200,
packages: Vec::new(),
project_name: "acme".into(),
dispatch: DispatchOrder::new(),
})
.unwrap();
let src = "{% set res = run_query('select') %}{{ res.columns | join(',') }}|\
{% for row in res.rows %}{{ row.region }}={{ row[1] }}{{ ',' if not loop.last }}{% endfor %}|{{ res | length }}|{{ res[1]['region'] }}";
assert_eq!(render(&r, src).unwrap(), "region,n|apac=1,emea=2|2|emea");
let e = render(&r, "{{ run_query('select', max_rows=1) }}").unwrap_err();
assert!(e.message.contains("more than 1 rows"), "{e}");
assert!(render(&r, "{% if run.set is none %}none{% endif %}").unwrap() == "none");
}
#[test]
fn removed_names_say_what_to_write_instead() {
let e = err("{{ run.profile }}");
assert!(
e.contains("`run.profile` was removed") && e.contains("connection.name"),
"{e}"
);
let e = err("{{ target.schema }}");
assert!(
e.contains("`target.schema` was removed") && e.contains("connection.schema"),
"{e}"
);
}
#[test]
fn a_source_relation_quotes_the_parts_asked_for() {
let r = ResolvedSource {
source: "sales".into(),
table: "orders".into(),
profile: None,
database: None,
schema: "Sales".into(),
identifier: "Or\"ders".into(),
quoting: Quoting {
database: true,
schema: false,
identifier: true,
},
};
assert_eq!(r.relation(Some("\"")), "Sales.\"Or\"\"ders\"");
assert_eq!(r.relation(None), "Sales.Or\"ders");
let r = ResolvedSource {
database: Some("main".into()),
..r
};
assert_eq!(r.relation(Some("`")), "`main`.Sales.`Or\"ders`");
}
#[test]
fn a_relation_is_mentioned_only_as_a_whole_name() {
assert!(mentions("select * from sales.orders where 1", "sales.orders"));
assert!(!mentions("select * from sales.orders_old", "sales.orders"));
assert!(!mentions("select * from x.sales.orders", "sales.orders"));
}
}
#[cfg(test)]
mod cli_var_tests {
use super::cli_var_value as v;
use serde_json::json;
#[test]
fn yaml_1_2_core_rules() {
assert_eq!(v("false"), json!(false));
assert_eq!(v("TRUE"), json!(true));
assert_eq!(v("~"), json!(null));
assert_eq!(v("5"), json!(5));
assert_eq!(v("-12"), json!(-12));
assert_eq!(v("2.5"), json!(2.5));
assert_eq!(v("[\"NAM\", EMEA]"), json!(["NAM", "EMEA"]));
assert_eq!(v("{k: 1}"), json!({"k": 1}));
for s in [
"yes",
"NO",
"2026-01-31",
"010",
"1.0.0",
"abc",
"[unclosed",
"0x1F",
] {
assert_eq!(v(s), json!(s), "{s}");
}
assert_eq!(v("\"false\""), json!("false"));
assert_eq!(v("'5'"), json!("5"));
assert_eq!(v(""), json!(""));
}
}