1use std::collections::BTreeMap;
14use std::fmt;
15use std::path::{Path, PathBuf};
16use std::sync::{Arc, Mutex};
17
18use chrono::{NaiveDate, Utc};
19
20use crate::dates::{Calendar, Date, DateTime};
21
22use crate::lookups::{self, Cell, Load, Lookup, Table};
23use crate::packages::{DispatchOrder, Package};
24use minijinja::value::{Enumerator, Kwargs, Object, ObjectRepr, Value};
25use minijinja::{Environment, Error, ErrorKind, UndefinedBehavior};
26use serde_json::{Map as JsonMap, Value as Json};
27
28const MACROS: &str = "__dre_macros__";
29
30#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
32pub enum Mode {
33 #[default]
35 Run,
36 Parse,
38}
39
40#[derive(Debug, Clone)]
42pub struct RunContext {
43 pub report: String,
44 pub set: Option<String>,
45 pub target: String,
47 pub schedule: Option<String>,
49 pub date: NaiveDate,
50 pub now: chrono::DateTime<Utc>,
52 pub scheduled_at: Option<chrono::DateTime<Utc>>,
54 pub calendar: Calendar,
56 pub locale: crate::numbers::Locale,
58}
59
60const MARKDOWN_SUFFIX: &str = "\u{0}md";
62
63#[derive(Debug, Clone, Default)]
65pub struct QueryRows {
66 pub columns: Vec<String>,
67 pub rows: Vec<Vec<Value>>,
68}
69
70#[derive(Debug, Clone, PartialEq, Eq)]
72pub struct Column {
73 pub name: String,
74 pub data_type: String,
76}
77
78pub trait QueryRunner: Send + Sync {
81 fn run_query(&self, sql: &str, max_rows: u64, profile: Option<&str>) -> Result<QueryRows, String>;
83 fn columns(&self, sql: &str, profile: Option<&str>) -> Result<Vec<Column>, String>;
85 fn load(&self, _name: &str, _table: &Table) -> Result<Option<(String, Option<String>)>, String> {
88 Ok(None)
89 }
90}
91
92#[derive(Debug, Clone, Default)]
95pub struct Connection {
96 pub profile: String,
98 pub target: String,
100 pub kind: String,
102 pub fields: JsonMap<String, Json>,
104 pub secrets: Vec<String>,
106}
107
108pub trait Connections: Send + Sync {
111 fn profile(&self, name: &str, role: Option<&str>) -> Result<Connection, String>;
112 fn identifier_quote(&self, kind: &str) -> Result<Option<String>, String>;
114 fn string_literal(&self, _kind: &str) -> Result<Option<String>, String> {
117 Ok(None)
118 }
119}
120
121#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Serialize)]
123pub struct Quoting {
124 pub database: bool,
125 pub schema: bool,
126 pub identifier: bool,
127}
128
129impl Quoting {
130 pub fn any(&self) -> bool {
131 self.database || self.schema || self.identifier
132 }
133}
134
135#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
137pub struct ResolvedSource {
138 pub source: String,
139 pub table: String,
140 pub profile: Option<String>,
142 pub database: Option<String>,
143 pub schema: String,
144 pub identifier: String,
145 pub quoting: Quoting,
146}
147
148impl ResolvedSource {
149 pub fn relation(&self, quote: Option<&str>) -> String {
152 let part = |s: &str, q: bool| match (q, quote) {
153 (true, Some(c)) => format!("{c}{}{c}", s.replace(c, &format!("{c}{c}"))),
154 _ => s.to_string(),
155 };
156 let mut parts = Vec::new();
157 if let Some(d) = &self.database {
158 parts.push(part(d, self.quoting.database));
159 }
160 parts.push(part(&self.schema, self.quoting.schema));
161 parts.push(part(&self.identifier, self.quoting.identifier));
162 parts.join(".")
163 }
164
165 pub fn key(&self) -> String {
167 format!("{}.{}", self.source, self.table)
168 }
169}
170
171pub type SourceResolver = Arc<dyn Fn(&str, &str) -> Result<ResolvedSource, String> + Send + Sync>;
173
174#[derive(Debug, Clone, PartialEq, Eq)]
175pub struct RenderError {
176 pub file: PathBuf,
177 pub line: Option<usize>,
178 pub message: String,
179}
180
181impl fmt::Display for RenderError {
182 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
183 match self.line {
184 Some(l) => write!(f, "{}:{l}: {}", self.file.display(), self.message),
185 None => write!(f, "{}: {}", self.file.display(), self.message),
186 }
187 }
188}
189
190type ColumnsCache = BTreeMap<(String, Option<String>), Value>;
192
193pub struct Renderer {
194 env: Environment<'static>,
195 raised: Arc<Mutex<Option<String>>>,
197 columns_cache: Arc<Mutex<ColumnsCache>>,
199 source_calls: Arc<Mutex<Vec<(String, String)>>>,
201 destination: Arc<Mutex<Option<Connection>>>,
203 warnings: Arc<Mutex<Vec<String>>>,
205 import: String,
206 macro_files: BTreeMap<String, Vec<(usize, PathBuf, usize)>>,
208}
209
210pub struct RendererConfig<'a> {
211 pub root: &'a Path,
212 pub macros: &'a [PathBuf],
213 pub context: RunContext,
214 pub vars: JsonMap<String, Json>,
216 pub cli_vars: BTreeMap<String, String>,
218 pub runner: Option<Arc<dyn QueryRunner>>,
219 pub connections: Option<Arc<dyn Connections>>,
221 pub mode: Mode,
222 pub connection: Option<String>,
225 pub source_type: String,
227 pub sources: Option<SourceResolver>,
229 pub run_query_max_rows: u64,
230 pub sql: BTreeMap<String, PathBuf>,
232 pub lookups: BTreeMap<String, Lookup>,
234 pub lookup_inline_max_rows: u64,
235 pub packages: Vec<Package>,
237 pub project_name: String,
239 pub dispatch: DispatchOrder,
240}
241
242impl Renderer {
243 pub fn new(cfg: RendererConfig<'_>) -> Result<Renderer, RenderError> {
244 let mut env = Environment::new();
245 let parse = cfg.mode == Mode::Parse;
246 env.set_undefined_behavior(if parse {
248 UndefinedBehavior::Chainable
249 } else {
250 UndefinedBehavior::Strict
251 });
252 env.set_syntax(
253 minijinja::syntax::SyntaxConfig::builder()
254 .keep_trailing_newline(true)
255 .build()
256 .expect("default delimiters are valid"),
257 );
258 env.set_auto_escape_callback(|name| {
261 if name.ends_with(MARKDOWN_SUFFIX) {
262 minijinja::AutoEscape::Custom("markdown".into())
263 } else {
264 minijinja::AutoEscape::None
265 }
266 });
267 env.set_formatter(|out, state, value| {
268 if *state.auto_escape() == minijinja::AutoEscape::Custom("markdown".into()) {
269 let text = value.to_string();
270 let text = if value.is_safe() {
271 text
272 } else {
273 dre_protocol::markdown::escape(&text)
274 };
275 return out.write_str(&text).map_err(Error::from);
276 }
277 minijinja::escape_formatter(out, state, value)
278 });
279 crate::mutable::register(&mut env);
280
281 add_vars(&mut env, cfg.vars, cfg.cli_vars);
282 let raised: Arc<Mutex<Option<String>>> = Arc::default();
286 let raised_c = raised.clone();
287 env.add_function("raise_error", move |message: String| -> Result<Value, Error> {
288 if parse {
289 raised_c.lock().unwrap().get_or_insert(message);
290 return Ok(Value::from(""));
291 }
292 Err(Error::new(ErrorKind::InvalidOperation, message))
293 });
294 let source_type = cfg.source_type.clone();
295 crate::dates::register(&mut env, cfg.context.calendar, cfg.context.date);
296 crate::numbers::register(&mut env, cfg.context.locale);
297 env.add_global("target", Value::from_object(Target(cfg.context.target.clone())));
298 env.add_global("run", Value::from_object(Run(cfg.context)));
299 let destination: Arc<Mutex<Option<Connection>>> = Arc::default();
300 if parse {
301 env.add_global("connection", Value::from_object(Lenient));
302 env.add_global("destination", Value::from_object(Lenient));
303 env.add_function(
304 "profile",
305 |_name: String, kwargs: Kwargs| -> Result<Value, Error> {
306 let role: Option<String> = kwargs.get("role")?;
307 kwargs.assert_all_used()?;
308 check_role(role.as_deref())?;
309 Ok(Value::from_object(Lenient))
310 },
311 );
312 } else {
313 env.add_global(
314 "connection",
315 Value::from_object(LazyConnection {
316 connections: cfg.connections.clone(),
317 name: cfg.connection.clone(),
318 resolved: std::sync::OnceLock::new(),
319 }),
320 );
321 env.add_global(
322 "destination",
323 Value::from_object(DestinationSlot(destination.clone())),
324 );
325 let c = cfg.connections.clone();
326 env.add_function(
327 "profile",
328 move |name: String, kwargs: Kwargs| -> Result<Value, Error> {
329 let role: Option<String> = kwargs.get("role")?;
330 kwargs.assert_all_used()?;
331 check_role(role.as_deref())?;
332 let Some(c) = &c else {
333 return Err(Error::new(
334 ErrorKind::InvalidOperation,
335 format!("`profile('{name}')`: no profiles.yml is loaded here"),
336 ));
337 };
338 c.profile(&name, role.as_deref())
339 .map(|t| Value::from_object(ConnectionValue(t)))
340 .map_err(|e| {
341 Error::new(ErrorKind::InvalidOperation, format!("`profile('{name}')`: {e}"))
342 })
343 },
344 );
345 }
346 let source_calls: Arc<Mutex<Vec<(String, String)>>> = Arc::default();
349 let relations: Arc<Mutex<Vec<(String, ResolvedSource)>>> = Arc::default();
350 {
351 let calls = source_calls.clone();
352 let relations = relations.clone();
353 let resolver = cfg.sources.clone();
354 let connections = cfg.connections.clone();
355 let kind = cfg.source_type.clone();
356 let quote: Arc<std::sync::OnceLock<Result<Option<String>, String>>> = Arc::default();
357 env.add_function(
358 "source",
359 move |source: String, table: String| -> Result<Value, Error> {
360 let fail = |e: String| {
361 Error::new(
362 ErrorKind::InvalidOperation,
363 format!("`source('{source}', '{table}')`: {e}"),
364 )
365 };
366 let Some(resolve) = &resolver else {
367 return Err(fail("the project declares no `sources:`".into()));
368 };
369 let r = resolve(&source, &table).map_err(fail)?;
370 calls.lock().unwrap().push((source.clone(), table.clone()));
371 let q = if parse || !r.quoting.any() {
372 None
373 } else {
374 let got = quote.get_or_init(|| match &connections {
375 Some(c) if !kind.is_empty() => c.identifier_quote(&kind),
376 _ => Err("no connection is known here to quote with".into()),
377 });
378 match got {
379 Ok(Some(q)) => Some(q.clone()),
380 Ok(None) => {
381 return Err(fail(format!(
382 "`quoting` needs the identifier quote character, but the `{kind}` plugin doesn't report one; update it (`dre plugin update {kind}`)"
383 )));
384 }
385 Err(e) => return Err(fail(format!("can't quote: {e}"))),
386 }
387 };
388 let text = r.relation(q.as_deref());
389 relations.lock().unwrap().push((text.clone(), r));
390 Ok(Value::from(text))
391 },
392 );
393 }
394 let runner = cfg.runner;
395 let warnings: Arc<Mutex<Vec<String>>> = Arc::default();
396 let lookups = Arc::new(Lookups {
397 connections: cfg.connections.clone(),
398 kind: cfg.source_type.clone(),
399 style: std::sync::OnceLock::new(),
400 root: cfg.root.to_path_buf(),
401 defs: cfg.lookups,
402 inline_max: cfg.lookup_inline_max_rows,
403 runner: runner.clone(),
404 tables: Mutex::default(),
405 loaded: Mutex::default(),
406 warnings: warnings.clone(),
407 });
408 let l = lookups.clone();
409 env.add_function("lookup", move |name: String| -> Result<Value, Error> {
410 let t = l.table(&name)?;
411 let rows = QueryRows {
412 columns: t.columns.clone(),
413 rows: t
414 .rows
415 .iter()
416 .map(|r| r.iter().map(cell_value).collect())
417 .collect(),
418 };
419 Ok(Value::from_object(QueryResult::new(rows)))
420 });
421 let default_max = cfg.run_query_max_rows;
422 let runner_c = runner.clone();
423 let rels = relations.clone();
424 env.add_function(
425 "run_query",
426 move |sql: String, kwargs: Kwargs| -> Result<Value, Error> {
427 let max_rows: Option<u64> = kwargs.get("max_rows")?;
428 let profile: Option<String> = kwargs.get("profile")?;
429 kwargs.assert_all_used()?;
430 if parse {
431 let mut empty = QueryResult::new(QueryRows::default());
432 empty.lenient = true;
433 return Ok(Value::from_object(empty));
434 }
435 let profile = choose_profile("run_query()", &sql, profile, &rels)?;
436 let Some(runner) = &runner else {
437 return Err(Error::new(
438 ErrorKind::InvalidOperation,
439 "`run_query()` has no connection here",
440 ));
441 };
442 let rows = runner
443 .run_query(&sql, max_rows.unwrap_or(default_max), profile.as_deref())
444 .map_err(|e| Error::new(ErrorKind::InvalidOperation, e))?;
445 Ok(Value::from_object(QueryResult::new(rows)))
446 },
447 );
448 let columns_cache: Arc<Mutex<ColumnsCache>> = Arc::default();
450 let cache = columns_cache.clone();
451 let rels = relations.clone();
452 env.add_function(
453 "columns",
454 move |rel: String, kwargs: Kwargs| -> Result<Value, Error> {
455 let profile: Option<String> = kwargs.get("profile")?;
456 kwargs.assert_all_used()?;
457 if parse {
458 return Ok(Value::from(Vec::<Value>::new()));
459 }
460 let what = format!("columns('{}')", crate::secrets::mask(&rel));
461 let profile = choose_profile(&what, &rel, profile, &rels)?;
462 let key = (rel.clone(), profile.clone());
463 if let Some(v) = cache.lock().unwrap().get(&key) {
464 return Ok(v.clone());
465 }
466 let Some(runner) = &runner_c else {
467 return Err(Error::new(
468 ErrorKind::InvalidOperation,
469 format!("`{what}` has no connection here"),
470 ));
471 };
472 let sql = format!("select * from {rel} as _dre_cols where 1=0");
473 let cols = runner
474 .columns(&sql, profile.as_deref())
475 .map_err(|e| Error::new(ErrorKind::InvalidOperation, format!("`{what}` failed: {e}")))?;
476 let v = Value::from(
477 cols.into_iter()
478 .map(|c| {
479 Value::from_pairs([
480 ("name", Value::from(c.name)),
481 ("type", Value::from(c.data_type)),
482 ])
483 })
484 .collect::<Vec<_>>(),
485 );
486 cache.lock().unwrap().insert(key, v.clone());
487 Ok(v)
488 },
489 );
490
491 let root = cfg.root.to_path_buf();
495 let display = |p: &Path| {
496 p.strip_prefix(&root)
497 .map(Path::to_path_buf)
498 .unwrap_or_else(|_| p.to_path_buf())
499 };
500 let own: Vec<(PathBuf, PathBuf)> = cfg.macros.iter().map(|m| (cfg.root.join(m), m.clone())).collect();
501 let (combined, names, spans) = combine(&own)?;
502 let mut import = if names.is_empty() {
503 String::new()
504 } else {
505 format!("{{% from \"{MACROS}\" import {} %}}", names.join(", "))
506 };
507 let mut templates = vec![(MACROS.to_string(), combined, spans, !names.is_empty())];
508 for p in &cfg.packages {
509 let files: Vec<(PathBuf, PathBuf)> = p.macros.iter().map(|m| (m.clone(), display(m))).collect();
510 let (src, _, spans) = combine(&files)?;
511 let t = package_template(&p.name);
512 import.push_str(&format!("{{% import \"{t}\" as {} %}}", p.name));
513 templates.push((t, src, spans, true));
514 }
515 add_ref(&mut env, cfg.root, cfg.sql, lookups, &import);
516 add_dispatch(
517 &mut env,
518 &cfg.project_name,
519 !names.is_empty(),
520 &cfg.packages,
521 cfg.dispatch,
522 source_type,
523 parse,
524 );
525 let mut r = Renderer {
526 env,
527 raised,
528 columns_cache,
529 source_calls,
530 destination,
531 warnings,
532 import,
533 macro_files: BTreeMap::new(),
534 };
535 for (name, src, spans, add) in templates {
536 r.macro_files.insert(name.clone(), spans);
537 if add {
538 r.env
539 .add_template_owned(name.clone(), src)
540 .map_err(|e| r.error(Path::new(&name), &e))?;
541 }
542 }
543 Ok(r)
544 }
545
546 pub fn take_warnings(&self) -> Vec<String> {
548 std::mem::take(&mut *self.warnings.lock().unwrap())
549 }
550
551 pub fn take_sources(&self) -> Vec<(String, String)> {
553 let mut out: Vec<(String, String)> = Vec::new();
554 for c in std::mem::take(&mut *self.source_calls.lock().unwrap()) {
555 if !out.contains(&c) {
556 out.push(c);
557 }
558 }
559 out
560 }
561
562 pub fn set_destination(&self, d: Option<Connection>) {
564 *self.destination.lock().unwrap() = d;
565 }
566
567 pub fn render(&self, file: &Path, src: &str) -> Result<String, RenderError> {
569 self.render_with(file, src, Value::UNDEFINED, false)
570 }
571
572 pub fn render_with(
575 &self,
576 file: &Path,
577 src: &str,
578 ctx: Value,
579 markdown: bool,
580 ) -> Result<String, RenderError> {
581 self.columns_cache.lock().unwrap().clear();
582 *self.raised.lock().unwrap() = None;
583 let full = format!("{}{src}", self.import);
584 let mut name = file.to_string_lossy().to_string();
585 if markdown {
586 name.push_str(MARKDOWN_SUFFIX);
587 }
588 let tmpl = self
589 .env
590 .template_from_named_str(&name, &full)
591 .map_err(|e| self.error(file, &e))?;
592 let rendered = if ctx.is_undefined() {
593 tmpl.render(())
594 } else {
595 tmpl.render(ctx)
596 };
597 rendered.map_err(|e| {
598 let mut err = self.error(file, &e);
599 if let Some(m) = self.raised.lock().unwrap().take() {
600 err.message = m;
601 }
602 err
603 })
604 }
605
606 fn error(&self, file: &Path, e: &Error) -> RenderError {
607 let mut deepest: &Error = e;
609 while let Some(src) = std::error::Error::source(deepest).and_then(|s| s.downcast_ref::<Error>()) {
610 deepest = src;
611 }
612 let (name, line) = match (deepest.name(), deepest.line()) {
613 (Some(n), l) => (n.to_string(), l),
614 _ => (e.name().unwrap_or_default().to_string(), e.line()),
615 };
616 let (file, line) = if let Some(spans) = self.macro_files.get(&name) {
617 match line.and_then(|l| {
618 spans
619 .iter()
620 .find(|(s, _, n)| l >= *s && l < s + n)
621 .map(|(s, f, _)| (f, l - s + 1))
622 }) {
623 Some((f, l)) => (f.clone(), Some(l)),
624 None => (file.to_path_buf(), line),
625 }
626 } else {
627 (file.to_path_buf(), line)
628 };
629 let message = deepest
630 .detail()
631 .map(str::to_string)
632 .unwrap_or_else(|| deepest.kind().to_string());
633 RenderError { file, line, message }
634 }
635}
636
637pub fn cli_var_value(s: &str) -> Json {
643 let t = s.trim();
644 let int = regex_lite(t, |c, i| c.is_ascii_digit() || (i == 0 && (c == '-' || c == '+')));
645 match t {
646 "" => return Json::String(String::new()),
647 "true" | "True" | "TRUE" => return Json::Bool(true),
648 "false" | "False" | "FALSE" => return Json::Bool(false),
649 "null" | "Null" | "NULL" | "~" => return Json::Null,
650 _ => {}
651 }
652 if int {
653 let digits = t.trim_start_matches(['-', '+']);
654 if !digits.is_empty()
655 && !(digits.len() > 1 && digits.starts_with('0'))
656 && let Ok(n) = t.parse::<i64>()
657 {
658 return Json::from(n);
659 }
660 return Json::String(s.to_string());
661 }
662 let float = t.trim_start_matches(['-', '+']);
663 let is_float = !float.is_empty()
664 && float
665 .chars()
666 .all(|c| c.is_ascii_digit() || matches!(c, '.' | 'e' | 'E' | '-' | '+'))
667 && float
668 .chars()
669 .next()
670 .is_some_and(|c| c.is_ascii_digit() || c == '.')
671 && float.chars().filter(|c| *c == '.').count() <= 1
672 && !float.starts_with("00")
673 && !(float.starts_with('0') && float.len() > 1 && float.as_bytes()[1].is_ascii_digit());
674 if is_float
675 && (float.contains('.') || float.contains(['e', 'E']))
676 && let Ok(f) = t.parse::<f64>()
677 && let Some(n) = serde_json::Number::from_f64(f)
678 {
679 return Json::Number(n);
680 }
681 let quoted = (t.starts_with('"') && t.ends_with('"') && t.len() >= 2)
682 || (t.starts_with('\'') && t.ends_with('\'') && t.len() >= 2);
683 let flow = (t.starts_with('[') && t.ends_with(']')) || (t.starts_with('{') && t.ends_with('}'));
684 if (quoted || flow)
685 && let Ok(v) = serde_saphyr::from_str::<Json>(t)
686 {
687 return v;
688 }
689 Json::String(s.to_string())
690}
691
692fn regex_lite(s: &str, ok: impl Fn(char, usize) -> bool) -> bool {
694 !s.is_empty() && s.chars().enumerate().all(|(i, c)| ok(c, i))
695}
696
697fn add_vars(env: &mut Environment<'static>, vars: JsonMap<String, Json>, cli: BTreeMap<String, String>) {
698 env.add_function(
699 "var",
700 move |name: String, default: Option<Value>| -> Result<Value, Error> {
701 if let Some(v) = cli.get(&name) {
702 return Ok(Value::from(minijinja::value::Serde(cli_var_value(v))));
703 }
704 if let Some(v) = vars.get(&name) {
705 return Ok(Value::from(minijinja::value::Serde(v)));
706 }
707 default.ok_or_else(|| {
708 Error::new(
709 ErrorKind::UndefinedError,
710 format!("`var('{name}')` has no value and no default"),
711 )
712 })
713 },
714 );
715 env.add_function("env_var", |name: String, default: Option<Value>| -> Result<Value, Error> {
716 match std::env::var(&name) {
717 Ok(v) => Ok(Value::from(v)),
718 Err(_) => default.ok_or_else(|| {
719 Error::new(
720 ErrorKind::UndefinedError,
721 format!("`env_var('{name}')`: environment variable `{name}` is not set and no default is given"),
722 )
723 }),
724 }
725 });
726}
727
728pub struct Limited {
733 env: Environment<'static>,
734}
735
736const LIMITED_FUNCTIONS: &[&str] = &[
738 "run_query",
739 "columns",
740 "source",
741 "ref",
742 "lookup",
743 "dispatch",
744 "profile",
745];
746const LIMITED_GLOBALS: &[&str] = &["connection", "destination"];
747
748impl Limited {
749 pub fn new(
750 context: RunContext,
751 vars: JsonMap<String, Json>,
752 cli_vars: BTreeMap<String, String>,
753 ) -> Limited {
754 let mut env = Environment::new();
755 env.set_undefined_behavior(UndefinedBehavior::Strict);
756 crate::mutable::register(&mut env);
757 add_vars(&mut env, vars, cli_vars);
758 env.add_function("raise_error", |message: String| -> Result<Value, Error> {
759 Err(Error::new(ErrorKind::InvalidOperation, message))
760 });
761 crate::dates::register(&mut env, context.calendar, context.date);
762 crate::numbers::register(&mut env, context.locale);
763 env.add_global("target", Value::from_object(Target(context.target.clone())));
764 env.add_global("run", Value::from_object(Run(context)));
765 for f in LIMITED_FUNCTIONS {
766 let name = *f;
767 env.add_function(name, move |_args: minijinja::value::Rest<Value>| -> Result<Value, Error> {
768 Err(Error::new(
769 ErrorKind::InvalidOperation,
770 format!("`{name}()` can't be used here: this value chooses a connection or a source, so it's rendered before any connection is open"),
771 ))
772 });
773 }
774 for g in LIMITED_GLOBALS {
775 env.add_global(*g, Value::from_object(Forbidden(g)));
776 }
777 Limited { env }
778 }
779
780 pub fn render(&self, key: &str, src: &str) -> Result<String, String> {
782 if !crate::preflight::is_templated(src) {
783 return Ok(src.to_string());
784 }
785 self.env
786 .render_str(src, ())
787 .map(|v| v.trim().to_string())
788 .map_err(|e| {
789 let mut deepest: &Error = &e;
790 while let Some(s) = std::error::Error::source(deepest).and_then(|s| s.downcast_ref::<Error>())
791 {
792 deepest = s;
793 }
794 let msg = deepest
795 .detail()
796 .map(str::to_string)
797 .unwrap_or_else(|| deepest.kind().to_string());
798 format!("{key} `{src}`: {msg}")
799 })
800 }
801}
802
803#[derive(Debug)]
805struct Forbidden(&'static str);
806
807impl Object for Forbidden {
808 fn get_value(self: &Arc<Self>, _key: &Value) -> Option<Value> {
809 Some(Value::from(Error::new(
810 ErrorKind::InvalidOperation,
811 format!(
812 "`{}.*` can't be used here: this value chooses a connection or a source, so it's rendered before any connection is open",
813 self.0
814 ),
815 )))
816 }
817
818 fn render(self: &Arc<Self>, _f: &mut fmt::Formatter<'_>) -> fmt::Result {
819 Err(fmt::Error)
820 }
821}
822
823fn package_template(name: &str) -> String {
824 format!("__dre_package_{name}__")
825}
826
827#[allow(clippy::type_complexity)]
830fn combine(
831 files: &[(PathBuf, PathBuf)],
832) -> Result<(String, Vec<String>, Vec<(usize, PathBuf, usize)>), RenderError> {
833 let re = regex::Regex::new(r"\{%-?\s*macro\s+([A-Za-z_]\w*)").unwrap();
834 let mut combined = String::new();
835 let mut names = Vec::new();
836 let mut spans = Vec::new();
837 for (read, shown) in files {
838 let src = std::fs::read_to_string(read).map_err(|e| RenderError {
839 file: shown.clone(),
840 line: None,
841 message: format!("can't read macro file: {e}"),
842 })?;
843 let start = combined.matches('\n').count() + 1;
844 names.extend(re.captures_iter(&src).map(|c| c[1].to_string()));
845 combined.push_str(&src);
846 if !combined.ends_with('\n') {
847 combined.push('\n');
848 }
849 spans.push((start, shown.clone(), src.matches('\n').count() + 1));
850 }
851 Ok((combined, names, spans))
852}
853
854fn add_dispatch(
859 env: &mut Environment<'static>,
860 project: &str,
861 project_has_macros: bool,
862 packages: &[Package],
863 order: DispatchOrder,
864 source_type: String,
865 parse: bool,
866) {
867 let project = project.to_string();
868 let templates: BTreeMap<String, String> = packages
869 .iter()
870 .map(|p| (p.name.clone(), package_template(&p.name)))
871 .collect();
872 env.add_function(
873 "dispatch",
874 move |state: &minijinja::State<'_, '_>,
875 name: String,
876 namespace: Option<String>|
877 -> Result<Value, Error> {
878 let ns = namespace.unwrap_or_else(|| project.clone());
879 let search = order.get(&ns).cloned().unwrap_or_else(|| {
880 if ns == project {
881 vec![ns.clone()]
882 } else {
883 vec![project.clone(), ns.clone()]
884 }
885 });
886 let candidates = [format!("{source_type}__{name}"), format!("default__{name}")];
887 for n in &search {
888 let template = if n == &project {
889 if !project_has_macros {
890 continue;
891 }
892 MACROS.to_string()
893 } else {
894 match templates.get(n) {
895 Some(t) => t.clone(),
896 None => continue,
897 }
898 };
899 let tmpl = state.env().get_template(&template)?;
900 let captured = tmpl.render_captured(())?;
901 let st = captured.state();
902 for c in &candidates {
903 if st.lookup(c).is_some_and(|v| !v.is_undefined()) {
904 return Ok(Value::from_object(Dispatched {
905 template,
906 name: c.clone(),
907 }));
908 }
909 }
910 }
911 if parse {
912 return Ok(Value::from_object(Dispatched {
915 template: String::new(),
916 name: String::new(),
917 }));
918 }
919 Err(Error::new(
920 ErrorKind::InvalidOperation,
921 format!(
922 "`dispatch('{name}', '{ns}')`: no `{}` or `{}` in {}",
923 candidates[0],
924 candidates[1],
925 search.join(", ")
926 ),
927 ))
928 },
929 );
930}
931
932#[derive(Debug)]
934struct Dispatched {
935 template: String,
936 name: String,
937}
938
939impl Object for Dispatched {
940 fn call(self: &Arc<Self>, state: &mut minijinja::State<'_, '_>, args: &[Value]) -> Result<Value, Error> {
941 if self.template.is_empty() {
942 return Ok(Value::from(""));
943 }
944 let tmpl = state.env().get_template(&self.template)?;
945 let out = tmpl
946 .render_captured(())?
947 .with_state_mut(|s| s.call_macro(&self.name, args))?;
948 Ok(Value::from(out))
949 }
950}
951
952fn add_ref(
956 env: &mut Environment<'static>,
957 root: &Path,
958 sql: BTreeMap<String, PathBuf>,
959 lookups: Arc<Lookups>,
960 import: &str,
961) {
962 let root = root.to_path_buf();
963 let import = import.to_string();
964 let stack: Arc<Mutex<Vec<String>>> = Arc::default();
966 env.add_function(
967 "ref",
968 move |state: &minijinja::State<'_, '_>, name: String| -> Result<Value, Error> {
969 if lookups.defs.contains_key(&name) {
970 return lookups.relation(&name).map(Value::from);
971 }
972 let Some(rel) = sql.get(&name) else {
973 return Err(Error::new(
974 ErrorKind::InvalidOperation,
975 format!("`ref('{name}')`: no `{name}.sql` and no lookup `{name}` in the project"),
976 ));
977 };
978 {
979 let mut chain = stack.lock().unwrap();
980 if chain.contains(&name) {
981 chain.push(name.clone());
982 let cycle = chain.join(" → ");
983 chain.clear();
984 return Err(Error::new(
985 ErrorKind::InvalidOperation,
986 format!("`ref()` cycle: {cycle}"),
987 ));
988 }
989 chain.push(name.clone());
990 }
991 let result = (|| {
992 let src = std::fs::read_to_string(root.join(rel)).map_err(|e| {
993 Error::new(
994 ErrorKind::InvalidOperation,
995 format!("can't read {}: {e}", rel.display()),
996 )
997 })?;
998 let file = rel.to_string_lossy().to_string();
999 let rendered = state
1000 .env()
1001 .template_from_named_str(&file, &format!("{import}{src}"))?
1002 .render(())?;
1003 let statements = crate::sqlsplit::split(&rendered);
1004 match statements.as_slice() {
1005 [one] => Ok(Value::from(format!("(\n{}\n)", one.text))),
1006 _ => Err(Error::new(
1007 ErrorKind::InvalidOperation,
1008 format!(
1009 "`ref('{name}')` needs {} to hold exactly one statement, but it has {}",
1010 rel.display(),
1011 statements.len()
1012 ),
1013 )),
1014 }
1015 })();
1016 let mut chain = stack.lock().unwrap();
1017 if chain.last() == Some(&name) {
1018 chain.pop();
1019 }
1020 result
1021 },
1022 );
1023}
1024
1025struct Lookups {
1027 connections: Option<Arc<dyn Connections>>,
1029 kind: String,
1030 style: std::sync::OnceLock<Result<lookups::StringLiteral, String>>,
1031 root: PathBuf,
1032 defs: BTreeMap<String, Lookup>,
1033 inline_max: u64,
1034 runner: Option<Arc<dyn QueryRunner>>,
1035 tables: Mutex<BTreeMap<String, Arc<Table>>>,
1036 loaded: Mutex<BTreeMap<String, String>>,
1038 warnings: Arc<Mutex<Vec<String>>>,
1039}
1040
1041impl Lookups {
1042 fn table(&self, name: &str) -> Result<Arc<Table>, Error> {
1043 let def = self.defs.get(name).ok_or_else(|| {
1044 Error::new(
1045 ErrorKind::InvalidOperation,
1046 format!("no lookup `{name}` under lookups/"),
1047 )
1048 })?;
1049 if let Some(t) = self.tables.lock().unwrap().get(name) {
1050 return Ok(t.clone());
1051 }
1052 let t = Arc::new(lookups::read(&self.root, def).map_err(|e| {
1053 Error::new(
1054 ErrorKind::InvalidOperation,
1055 format!("{}: {e}", def.file.display()),
1056 )
1057 })?);
1058 self.tables.lock().unwrap().insert(name.to_string(), t.clone());
1059 Ok(t)
1060 }
1061
1062 fn relation(&self, name: &str) -> Result<String, Error> {
1063 if let Some(r) = self.loaded.lock().unwrap().get(name) {
1064 return Ok(r.clone());
1065 }
1066 let t = self.table(name)?;
1067 let rows = t.rows.len() as u64;
1068 let load = match self.defs[name].load {
1069 Load::Inline => false,
1070 Load::TempTable => true,
1071 Load::Auto => rows > self.inline_max,
1072 };
1073 let loaded = match (&self.runner, load) {
1074 (Some(runner), true) => runner.load(name, &t).map_err(|e| {
1075 Error::new(
1076 ErrorKind::InvalidOperation,
1077 format!("loading lookup `{name}`: {e}"),
1078 )
1079 })?,
1080 _ => None,
1081 };
1082 let relation = match loaded {
1083 Some((relation, warning)) => {
1084 if let Some(w) = warning {
1085 self.warn(format!("lookup `{name}`: {w}"));
1086 }
1087 relation
1088 }
1089 None => {
1090 if load && self.runner.is_some() {
1091 self.warn(format!(
1092 "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"
1093 ));
1094 }
1095 t.inline_sql(self.style()?)
1096 }
1097 };
1098 self.loaded
1099 .lock()
1100 .unwrap()
1101 .insert(name.to_string(), relation.clone());
1102 Ok(relation)
1103 }
1104
1105 fn warn(&self, w: String) {
1106 self.warnings.lock().unwrap().push(w);
1107 }
1108
1109 fn style(&self) -> Result<lookups::StringLiteral, Error> {
1112 self.style
1113 .get_or_init(|| match &self.connections {
1114 Some(c) if !self.kind.is_empty() => {
1115 let asked = c.string_literal(&self.kind).unwrap_or(None);
1116 lookups::StringLiteral::from_describe(asked.as_deref())
1117 }
1118 _ => Ok(lookups::StringLiteral::Chr),
1119 })
1120 .clone()
1121 .map_err(|e| {
1122 Error::new(
1123 ErrorKind::InvalidOperation,
1124 format!("the `{}` source: {e}", self.kind),
1125 )
1126 })
1127 }
1128}
1129
1130fn cell_value(c: &Cell) -> Value {
1131 match c {
1132 Cell::Null => Value::from(()),
1133 Cell::Text(s) => Value::from(s.clone()),
1134 Cell::Int(n) => Value::from(*n),
1135 Cell::Num(n) => Value::from(*n),
1136 Cell::Bool(b) => Value::from(*b),
1137 Cell::Date(d) => Value::from(d.to_string()),
1138 }
1139}
1140
1141#[derive(Debug)]
1142struct Run(RunContext);
1143
1144impl Object for Run {
1145 fn get_value(self: &Arc<Self>, key: &Value) -> Option<Value> {
1146 let c = &self.0;
1147 Some(match key.as_str()? {
1148 "report" => Value::from(c.report.clone()),
1149 "set" => c.set.clone().map(Value::from).unwrap_or(Value::from(())),
1150 "target" => Value::from(c.target.clone()),
1151 "profile" => removed("run.profile", "`connection.name` (the query's connection)"),
1152 "source_type" => removed("run.source_type", "`connection.type` (the query's connection)"),
1153 "schedule" => c.schedule.clone().map(Value::from).unwrap_or(Value::from(())),
1154 "date" => Date::value(c.date, c.calendar),
1155 "now" => DateTime::now(c.now, c.calendar),
1156 "scheduled_at" => c
1157 .scheduled_at
1158 .map(|t| DateTime::now(t, c.calendar))
1159 .unwrap_or(Value::from(())),
1160 "timezone" => Value::from(c.calendar.tz.name()),
1161 _ => return None,
1162 })
1163 }
1164
1165 fn call_method(
1166 self: &Arc<Self>,
1167 _: &mut minijinja::State<'_, '_>,
1168 method: &str,
1169 args: &[Value],
1170 ) -> Result<Value, Error> {
1171 match method {
1172 "date_format" => {
1173 let fmt = args.first().and_then(|a| a.as_str()).ok_or_else(|| {
1174 Error::new(
1175 ErrorKind::InvalidOperation,
1176 "`run.date_format()` takes a strftime format string",
1177 )
1178 })?;
1179 let mut out = String::new();
1180 use std::fmt::Write;
1181 write!(out, "{}", self.0.date.format(fmt)).map_err(|_| {
1182 Error::new(
1183 ErrorKind::InvalidOperation,
1184 format!("invalid date format `{fmt}`"),
1185 )
1186 })?;
1187 Ok(Value::from(out))
1188 }
1189 _ => Err(Error::from(ErrorKind::UnknownMethod)),
1190 }
1191 }
1192}
1193
1194fn removed(name: &str, instead: &str) -> Value {
1196 Value::from(Error::new(
1197 ErrorKind::InvalidOperation,
1198 format!("`{name}` was removed in DRE 0.2: use {instead}"),
1199 ))
1200}
1201
1202fn check_role(role: Option<&str>) -> Result<(), Error> {
1204 match role {
1205 None | Some("connection") | Some("destination") => Ok(()),
1206 Some("source") => Err(Error::new(
1207 ErrorKind::InvalidOperation,
1208 "`role='source'` was renamed in DRE 0.2: use `role='connection'`",
1209 )),
1210 Some(r) => Err(Error::new(
1211 ErrorKind::InvalidOperation,
1212 format!("`role='{r}'`: role must be 'connection' or 'destination'"),
1213 )),
1214 }
1215}
1216
1217fn choose_profile(
1221 what: &str,
1222 sql: &str,
1223 explicit: Option<String>,
1224 relations: &Mutex<Vec<(String, ResolvedSource)>>,
1225) -> Result<Option<String>, Error> {
1226 let mut used: Vec<(String, String)> = Vec::new();
1227 for (text, r) in relations.lock().unwrap().iter() {
1228 if let Some(p) = &r.profile
1229 && mentions(sql, text)
1230 && !used.iter().any(|(k, _)| *k == r.key())
1231 {
1232 used.push((r.key(), p.clone()));
1233 }
1234 }
1235 let fail = |m: String| Error::new(ErrorKind::InvalidOperation, format!("`{what}`: {m}"));
1236 match explicit {
1237 Some(p) => {
1238 if let Some((k, q)) = used.iter().find(|(_, q)| *q != p) {
1239 return Err(fail(format!(
1240 "`profile='{p}'` disagrees with source `{k}`, which is on connection `{q}`"
1241 )));
1242 }
1243 Ok(Some(p))
1244 }
1245 None => match used.as_slice() {
1246 [] => Ok(None),
1247 [(_, p), rest @ ..] => {
1248 if let Some((k, q)) = rest.iter().find(|(_, q)| q != p) {
1249 return Err(fail(format!(
1250 "reads source `{}` on connection `{p}` and source `{k}` on `{q}`; one statement runs on one connection",
1251 used[0].0
1252 )));
1253 }
1254 Ok(Some(p.clone()))
1255 }
1256 },
1257 }
1258}
1259
1260fn mentions(sql: &str, text: &str) -> bool {
1262 let ident = |c: char| c.is_alphanumeric() || c == '_';
1263 sql.match_indices(text).any(|(i, _)| {
1264 let before = sql[..i].chars().next_back();
1265 let after = sql[i + text.len()..].chars().next();
1266 !before.is_some_and(|c| ident(c) || c == '.') && !after.is_some_and(|c| ident(c) || c == '.')
1267 })
1268}
1269
1270#[derive(Debug)]
1273struct Target(String);
1274
1275impl Object for Target {
1276 fn get_value(self: &Arc<Self>, key: &Value) -> Option<Value> {
1277 let k = key.as_str()?;
1278 Some(match k {
1279 "name" => Value::from(self.0.clone()),
1280 "type" => removed("target.type", "`connection.type` (the query's connection)"),
1281 "profile" => removed("target.profile", "`connection.name` (the query's connection)"),
1282 _ => removed(
1283 &format!("target.{k}"),
1284 &format!(
1285 "`connection.{k}` (the query's connection) or `profile('<name>').{k}`; `target` is now only the environment, `target.name`"
1286 ),
1287 ),
1288 })
1289 }
1290
1291 fn render(self: &Arc<Self>, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1292 f.write_str(&self.0)
1293 }
1294}
1295
1296#[derive(Debug)]
1299struct Lenient;
1300
1301impl Object for Lenient {
1302 fn get_value(self: &Arc<Self>, _key: &Value) -> Option<Value> {
1303 None
1304 }
1305
1306 fn render(self: &Arc<Self>, _f: &mut fmt::Formatter<'_>) -> fmt::Result {
1307 Ok(())
1308 }
1309}
1310
1311#[derive(Debug)]
1313struct ConnectionValue(Connection);
1314
1315impl Object for ConnectionValue {
1316 fn get_value(self: &Arc<Self>, key: &Value) -> Option<Value> {
1317 connection_field(&self.0, key)
1318 }
1319
1320 fn render(self: &Arc<Self>, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1321 write!(f, "{}", self.0.profile)
1322 }
1323}
1324
1325struct LazyConnection {
1328 connections: Option<Arc<dyn Connections>>,
1329 name: Option<String>,
1330 resolved: std::sync::OnceLock<Result<Connection, String>>,
1331}
1332
1333impl fmt::Debug for LazyConnection {
1334 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1335 f.write_str("connection")
1336 }
1337}
1338
1339impl LazyConnection {
1340 fn get(&self) -> &Result<Connection, String> {
1341 self.resolved.get_or_init(|| {
1342 let name = self.name.as_deref().ok_or(
1343 "no connection here: this isn't a query's SQL and the Binding inherits no `profile`",
1344 )?;
1345 match &self.connections {
1346 Some(c) => c.profile(name, Some("connection")),
1347 None => Err("no profiles.yml is loaded here".into()),
1348 }
1349 })
1350 }
1351}
1352
1353impl Object for LazyConnection {
1354 fn get_value(self: &Arc<Self>, key: &Value) -> Option<Value> {
1355 match self.get() {
1356 Ok(c) => connection_field(c, key),
1357 Err(e) => Some(Value::from(Error::new(
1358 ErrorKind::InvalidOperation,
1359 format!("`connection`: {e}"),
1360 ))),
1361 }
1362 }
1363
1364 fn render(self: &Arc<Self>, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1365 match &self.name {
1366 Some(n) => f.write_str(n),
1367 None => Err(fmt::Error),
1368 }
1369 }
1370}
1371
1372#[derive(Debug)]
1374struct DestinationSlot(Arc<Mutex<Option<Connection>>>);
1375
1376impl Object for DestinationSlot {
1377 fn get_value(self: &Arc<Self>, key: &Value) -> Option<Value> {
1378 match &*self.0.lock().unwrap() {
1379 Some(c) => connection_field(c, key),
1380 None => Some(Value::from(Error::new(
1381 ErrorKind::InvalidOperation,
1382 "`destination` only exists while rendering a destination's `path` and options",
1383 ))),
1384 }
1385 }
1386
1387 fn render(self: &Arc<Self>, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1388 match &*self.0.lock().unwrap() {
1389 Some(c) => f.write_str(&c.profile),
1390 None => Err(fmt::Error),
1391 }
1392 }
1393}
1394
1395fn connection_field(c: &Connection, key: &Value) -> Option<Value> {
1397 {
1398 let k = key.as_str()?;
1399 Some(match k {
1400 "name" | "profile" => Value::from(c.profile.clone()),
1401 "target" => Value::from(c.target.clone()),
1402 "type" => Value::from(c.kind.clone()),
1403 _ if c.secrets.iter().any(|s| s == k) => Value::from(Error::new(
1404 ErrorKind::InvalidOperation,
1405 format!(
1406 "`{k}` of profile `{}` holds a secret, so templates can't read it (it would end up in compiled SQL and logs)",
1407 c.profile
1408 ),
1409 )),
1410 _ => match c.fields.get(k) {
1411 Some(v) => Value::from(minijinja::value::Serde(v)),
1412 None => Value::from(Error::new(
1413 ErrorKind::UndefinedError,
1414 format!(
1415 "profile `{}` (target `{}`) has no field `{k}`; it has: {}",
1416 c.profile,
1417 c.target,
1418 c.fields
1419 .keys()
1420 .filter(|f| !c.secrets.contains(f))
1421 .cloned()
1422 .collect::<Vec<_>>()
1423 .join(", ")
1424 ),
1425 )),
1426 },
1427 })
1428 }
1429}
1430
1431#[derive(Debug)]
1432struct QueryResult {
1433 columns: Vec<String>,
1434 rows: Vec<Value>,
1435 lenient: bool,
1437}
1438
1439impl QueryResult {
1440 fn new(q: QueryRows) -> QueryResult {
1441 let cols: Arc<Vec<String>> = Arc::new(q.columns.clone());
1442 let rows = q
1443 .rows
1444 .into_iter()
1445 .map(|values| {
1446 Value::from_object(Row {
1447 columns: cols.clone(),
1448 values,
1449 })
1450 })
1451 .collect();
1452 QueryResult {
1453 columns: q.columns,
1454 rows,
1455 lenient: false,
1456 }
1457 }
1458}
1459
1460impl Object for QueryResult {
1461 fn repr(self: &Arc<Self>) -> ObjectRepr {
1462 ObjectRepr::Seq
1463 }
1464
1465 fn get_value(self: &Arc<Self>, key: &Value) -> Option<Value> {
1466 match key.as_str() {
1467 Some("rows") => Some(Value::from(self.rows.clone())),
1468 Some("columns") => Some(Value::from(self.columns.clone())),
1469 _ => self.rows.get(key.as_usize()?).cloned(),
1470 }
1471 }
1472
1473 fn enumerate(self: &Arc<Self>) -> Enumerator {
1474 Enumerator::Seq(self.rows.len())
1475 }
1476
1477 fn call_method(
1479 self: &Arc<Self>,
1480 _: &mut minijinja::State<'_, '_>,
1481 method: &str,
1482 args: &[Value],
1483 ) -> Result<Value, Error> {
1484 if method != "column" {
1485 return Err(Error::from(ErrorKind::UnknownMethod));
1486 }
1487 if self.lenient {
1488 return Ok(Value::from(Vec::<Value>::new()));
1489 }
1490 let [key] = args else {
1491 return Err(Error::new(
1492 ErrorKind::InvalidOperation,
1493 "`column()` takes one column name or index",
1494 ));
1495 };
1496 let i = match key.as_str() {
1497 Some(n) => self.columns.iter().position(|c| c == n).ok_or_else(|| {
1498 Error::new(
1499 ErrorKind::InvalidOperation,
1500 format!("no column `{n}` (columns: {})", self.columns.join(", ")),
1501 )
1502 })?,
1503 None => key
1504 .as_usize()
1505 .filter(|i| *i < self.columns.len())
1506 .ok_or_else(|| Error::new(ErrorKind::InvalidOperation, format!("no column {key}")))?,
1507 };
1508 Ok(Value::from(
1509 self.rows
1510 .iter()
1511 .map(|r| r.get_item_by_index(i).unwrap_or_default())
1512 .collect::<Vec<_>>(),
1513 ))
1514 }
1515}
1516
1517#[derive(Debug)]
1519pub(crate) struct Row {
1520 columns: Arc<Vec<String>>,
1521 values: Vec<Value>,
1522}
1523
1524impl Row {
1525 pub(crate) fn value(columns: Arc<Vec<String>>, values: Vec<Value>) -> Value {
1526 Value::from_object(Row { columns, values })
1527 }
1528}
1529
1530impl Object for Row {
1531 fn repr(self: &Arc<Self>) -> ObjectRepr {
1532 ObjectRepr::Seq
1533 }
1534
1535 fn get_value(self: &Arc<Self>, key: &Value) -> Option<Value> {
1536 if let Some(name) = key.as_str() {
1537 let i = self.columns.iter().position(|c| c == name)?;
1538 return self.values.get(i).cloned();
1539 }
1540 self.values.get(key.as_usize()?).cloned()
1541 }
1542
1543 fn enumerate(self: &Arc<Self>) -> Enumerator {
1544 Enumerator::Seq(self.values.len())
1545 }
1546}
1547
1548#[cfg(test)]
1549mod tests {
1550 use super::*;
1551
1552 fn renderer(vars: Json, cli: &[(&str, &str)], macros: &[(&str, &str)]) -> (tempfile::TempDir, Renderer) {
1553 let dir = tempfile::tempdir().unwrap();
1554 let mut files = Vec::new();
1555 for (name, src) in macros {
1556 let p = PathBuf::from("macros").join(name);
1557 std::fs::create_dir_all(dir.path().join("macros")).unwrap();
1558 std::fs::write(dir.path().join(&p), src).unwrap();
1559 files.push(p);
1560 }
1561 let r = Renderer::new(RendererConfig {
1562 root: dir.path(),
1563 macros: &files,
1564 context: RunContext {
1565 report: "monthly".into(),
1566 set: Some("client_a".into()),
1567 target: "prod".into(),
1568 schedule: None,
1569 date: NaiveDate::from_ymd_opt(2026, 1, 25).unwrap(),
1570 now: Utc::now(),
1571 scheduled_at: None,
1572 calendar: Calendar::default(),
1573 locale: Default::default(),
1574 },
1575 vars: vars.as_object().cloned().unwrap_or_default(),
1576 cli_vars: cli.iter().map(|(k, v)| (k.to_string(), v.to_string())).collect(),
1577 runner: None,
1578 connections: None,
1579 mode: Mode::Run,
1580 connection: None,
1581 source_type: "duckdb".into(),
1582 sources: None,
1583 run_query_max_rows: 10_000,
1584 sql: BTreeMap::new(),
1585 lookups: BTreeMap::new(),
1586 lookup_inline_max_rows: 200,
1587 packages: Vec::new(),
1588 project_name: "acme".into(),
1589 dispatch: DispatchOrder::new(),
1590 })
1591 .unwrap();
1592 (dir, r)
1593 }
1594
1595 fn render(r: &Renderer, src: &str) -> Result<String, RenderError> {
1596 r.render(Path::new("reports/q.sql"), src)
1597 }
1598
1599 fn ok(src: &str) -> String {
1600 let (_d, r) = renderer(serde_json::json!({"fixed": [3, 1, 2], "m": {"a": 1}}), &[], &[]);
1601 render(&r, src).unwrap_or_else(|e| panic!("{src}: {e}"))
1602 }
1603
1604 fn err(src: &str) -> String {
1605 let (_d, r) = renderer(serde_json::json!({"fixed": [3, 1, 2], "m": {"a": 1}}), &[], &[]);
1606 render(&r, src).unwrap_err().to_string()
1607 }
1608
1609 #[test]
1610 fn python_string_and_dict_methods_work() {
1611 assert_eq!(ok("{{ 'a,b,c'.split(',') | join('-') }}"), "a-b-c");
1612 assert_eq!(
1613 ok("{{ 'y' if 'abc'.startswith('a') }}{{ 'y' if 'abc'.endswith('x') }}"),
1614 "y"
1615 );
1616 assert_eq!(
1617 ok("{{ ' x '.strip() }}|{{ 'abc'.upper() }}|{{ 'a-b'.replace('-', '_') }}"),
1618 "x|ABC|a_b"
1619 );
1620 assert_eq!(
1621 ok("{{ var('m').keys() | list }} {{ var('m').items() | list }}"),
1622 "['a'] [('a', 1)]"
1623 );
1624 assert_eq!(
1625 ok("{{ var('m').get('a') }} {{ var('m').get('z', 'none') }}"),
1626 "1 none"
1627 );
1628 }
1629
1630 #[test]
1631 fn list_appends_in_a_loop_and_joins() {
1632 assert_eq!(
1633 ok(
1634 "{% set sq = list() %}{% for i in range(1, 5) %}{% set _ = sq.append(i * i) %}{% endfor %}{{ sq | join('+') }}"
1635 ),
1636 "1+4+9+16"
1637 );
1638 }
1639
1640 #[test]
1641 fn list_methods_follow_python() {
1642 assert_eq!(
1643 ok("{% set l = list([3, 1, 2]) %}{% set _ = l.sort() %}{{ l }}"),
1644 "[1, 2, 3]"
1645 );
1646 assert_eq!(
1647 ok("{% set l = list([3, 1, 2]) %}{% set _ = l.sort(reverse=true) %}{{ l }}"),
1648 "[3, 2, 1]"
1649 );
1650 assert_eq!(
1651 ok("{% set l = list([1, 2]) %}{% set _ = l.extend([3, 4]) %}{{ l }} {{ l | length }}"),
1652 "[1, 2, 3, 4] 4"
1653 );
1654 assert_eq!(
1655 ok("{% set l = list([1, 2, 3]) %}{{ l.pop() }} {{ l.pop(0) }} {{ l }}"),
1656 "3 1 [2]"
1657 );
1658 assert_eq!(
1659 ok("{% set l = list([1, 2]) %}{% set _ = l.insert(0, 9) %}{% set _ = l.insert(99, 7) %}{{ l }}"),
1660 "[9, 1, 2, 7]"
1661 );
1662 assert_eq!(
1663 ok(
1664 "{% set l = list([1, 2, 1]) %}{% set _ = l.remove(1) %}{{ l }} {{ l.index(1) }} {{ l.count(1) }}"
1665 ),
1666 "[2, 1] 1 1"
1667 );
1668 assert_eq!(
1669 ok("{% set l = list([1, 2]) %}{% set _ = l.reverse() %}{{ l }}"),
1670 "[2, 1]"
1671 );
1672 assert_eq!(
1673 ok("{% set l = list([1, 2]) %}{% set _ = l.clear() %}{{ l }} {{ 'empty' if not l }}"),
1674 "[] empty"
1675 );
1676 assert_eq!(
1677 ok("{% set l = list([1, 2, 3]) %}{{ l[0] }} {{ l[-1] }} {{ l[1:] }}"),
1678 "1 3 [2, 3]"
1679 );
1680 }
1681
1682 #[test]
1683 fn lists_share_by_reference_and_copy_apart() {
1684 assert_eq!(
1685 ok("{% set a = list() %}{% set b = a %}{% set _ = b.append(1) %}{{ a }}"),
1686 "[1]"
1687 );
1688 assert_eq!(
1689 ok("{% set a = list([1]) %}{% set b = a.copy() %}{% set _ = b.append(2) %}{{ a }} {{ b }}"),
1690 "[1] [1, 2]"
1691 );
1692 }
1693
1694 #[test]
1695 fn a_list_changed_inside_its_own_loop_does_not_skip_or_repeat() {
1696 assert_eq!(
1697 ok(
1698 "{% set l = list([1, 2, 3]) %}{% for x in l %}{% set _ = l.append(x * 10) %}{{ x }} {% endfor %}{{ l }}"
1699 ),
1700 "1 2 3 [1, 2, 3, 10, 20, 30]"
1701 );
1702 assert_eq!(
1703 ok("{% set l = list([1, 2]) %}{% set _ = l.extend(l) %}{{ l }}"),
1704 "[1, 2, 1, 2]"
1705 );
1706 }
1707
1708 #[test]
1709 fn a_list_starts_from_a_var_without_changing_it() {
1710 assert_eq!(
1711 ok("{% set l = list(var('fixed')) %}{% set _ = l.append(4) %}{{ l }} {{ var('fixed') }}"),
1712 "[3, 1, 2, 4] [3, 1, 2]"
1713 );
1714 }
1715
1716 #[test]
1717 fn dict_methods_follow_python() {
1718 assert_eq!(
1719 ok("{% set d = dict(a=1) %}{% set _ = d.update(b=2) %}{% set _ = d.update({'c': 3}) %}{{ d }}"),
1720 "{'a': 1, 'b': 2, 'c': 3}"
1721 );
1722 let make = "{% set d = dict() %}{% set _ = d.setdefault('b', 1) %}{% set _ = d.setdefault('a', 2) %}";
1725 assert_eq!(
1726 ok(&(make.to_string() + "{% for k, v in d.items() %}{{ k }}={{ v }} {% endfor %}")),
1727 "b=1 a=2 "
1728 );
1729 assert_eq!(
1730 ok(&(make.to_string() + "{% for k in d %}{{ k }} {% endfor %}{{ d.values() | list }}")),
1731 "b a [1, 2]"
1732 );
1733 assert_eq!(
1734 ok("{% set d = dict(a=1) %}{{ d.setdefault('a', 9) }} {{ d.setdefault('b', 2) }} {{ d }}"),
1735 "1 2 {'a': 1, 'b': 2}"
1736 );
1737 assert_eq!(
1738 ok("{% set d = dict(a=1) %}{{ d.pop('a') }} {{ d.pop('a', 'gone') }} {{ d }}"),
1739 "1 gone {}"
1740 );
1741 assert_eq!(
1742 ok(
1743 "{% set d = dict(a=1) %}{{ d.get('a') }} {{ d['a'] }} {{ d.a }} {{ 'in' if 'a' in d }} {{ d.keys() | list }}"
1744 ),
1745 "1 1 1 in ['a']"
1746 );
1747 assert_eq!(
1748 ok("{% set d = dict(a=1) %}{% set e = d.copy() %}{% set _ = e.update(a=2) %}{{ d.a }} {{ e.a }}"),
1749 "1 2"
1750 );
1751 assert_eq!(
1752 ok("{% set d = dict(a=1) %}{% set _ = d.update(a=5, b=6) %}{{ d | length }} {{ d.a }}"),
1753 "2 5"
1754 );
1755 assert_eq!(
1756 ok("{% set d = dict(a=1) %}{% set _ = d.update(d) %}{{ d }}"),
1757 "{'a': 1}"
1758 );
1759 }
1760
1761 #[test]
1762 fn mutable_values_print_and_serialise_like_plain_ones() {
1763 assert_eq!(
1764 ok("{{ list([1, 'a']) }} {{ dict(k='v') }}"),
1765 "[1, 'a'] {'k': 'v'}"
1766 );
1767 assert_eq!(
1768 ok("{{ list(dict(k='v')) }} {{ list(list([2, 1])) | sort }}"),
1769 "['k'] [1, 2]"
1770 );
1771 assert_eq!(ok("{% set l = list() %}{{ 'yes' if l else 'no' }}"), "no");
1772 }
1773
1774 #[test]
1776 fn expressions_follow_jinja2() {
1777 assert_eq!(
1778 ok("{{ ['a', 1] }} {{ var('m') }} {{ var('m').items() | list }}"),
1779 "['a', 1] {'a': 1} [('a', 1)]"
1780 );
1781 assert_eq!(ok("{{ 2.5 | round }} {{ 3.5 | round }}"), "2.0 4.0");
1782 assert_eq!(ok("{{ -7 // 2 }} {{ -7 % 3 }}"), "-4 2");
1783 assert_eq!(
1784 ok("{{ ['b', 'P', 'n'] | max }} {{ \"don't\" | title }}"),
1785 "P Don't"
1786 );
1787 assert!(err("{{ 1 / 0 }}").contains("unable to calculate 1 / 0"));
1788 }
1789
1790 #[test]
1791 fn mutating_an_immutable_value_says_what_to_write() {
1792 let e = err("{% set l = [] %}{% set _ = l.append(1) %}");
1793 assert!(e.contains("append()") && e.contains("list()"), "{e}");
1794 let e = err("{% set _ = var('fixed').append(1) %}");
1795 assert!(e.contains("list()"), "{e}");
1796 let e = err("{% set d = {} %}{% set _ = d.update(a=1) %}");
1797 assert!(e.contains("dict()"), "{e}");
1798 }
1799
1800 #[test]
1801 fn bad_arguments_to_mutable_methods_are_clear_errors() {
1802 assert!(err("{% set l = list() %}{% set _ = l.append() %}").contains("append()"));
1803 assert!(err("{% set l = list() %}{% set _ = l.pop() %}").contains("pop()"));
1804 assert!(err("{% set l = list([1]) %}{% set _ = l.remove(2) %}").contains("not in the list"));
1805 assert!(err("{% set d = dict() %}{% set _ = d.pop('x') %}").contains("no key"));
1806 assert!(err("{% set d = dict('nope') %}").contains("mapping"));
1807 }
1808
1809 #[test]
1810 fn run_context_and_date_formats() {
1811 let (_d, r) = renderer(Json::Null, &[], &[]);
1812 let out = render(
1813 &r,
1814 "{{ run.report }}/{{ run.set }}/{{ run.target }}/{{ target.name }}/{{ target }} {{ run.date }} {{ run.date.yyyymmdd }} \
1815 {{ run.date.ddmmyyyy }} {{ run.date.yyyy }}-{{ run.date.mm }}-{{ run.date.dd }} {{ run.date_format('%Y-W%V') }}",
1816 )
1817 .unwrap();
1818 assert_eq!(
1819 out,
1820 "monthly/client_a/prod/prod/prod 2026-01-25 20260125 25012026 2026-01-25 2026-W04"
1821 );
1822 }
1823
1824 #[test]
1825 fn var_precedence_is_cli_then_binding_then_default() {
1826 let (_d, r) = renderer(
1827 serde_json::json!({"a": "binding", "b": "binding", "n": 5}),
1828 &[("a", "cli")],
1829 &[],
1830 );
1831 assert_eq!(
1832 render(
1833 &r,
1834 "{{ var('a') }} {{ var('b') }} {{ var('c', 'dflt') }} {{ var('n') + 1 }}"
1835 )
1836 .unwrap(),
1837 "cli binding dflt 6"
1838 );
1839 let e = render(&r, "select\n{{ var('missing') }}").unwrap_err();
1840 assert_eq!(
1841 (e.line, e.message.contains("`var('missing')` has no value")),
1842 (Some(2), true),
1843 "{e}"
1844 );
1845 }
1846
1847 #[test]
1848 fn env_var_reads_the_environment_or_errors() {
1849 let (_d, r) = renderer(Json::Null, &[], &[]);
1850 assert_eq!(
1851 render(&r, "{{ env_var('DRE_TEST_UNSET_VAR', 'x') }}").unwrap(),
1852 "x"
1853 );
1854 let home = std::env::var("PATH").unwrap();
1855 assert_eq!(render(&r, "{{ env_var('PATH') }}").unwrap(), home);
1856 assert!(
1857 render(&r, "{{ env_var('DRE_TEST_UNSET_VAR') }}")
1858 .unwrap_err()
1859 .message
1860 .contains("is not set")
1861 );
1862 }
1863
1864 #[test]
1865 fn macros_from_every_file_are_callable_and_keep_line_numbers() {
1866 let (_d, r) = renderer(
1867 Json::Null,
1868 &[],
1869 &[
1870 ("a.sql", "{% macro double(x) %}{{ x * 2 }}{% endmacro %}\n"),
1871 (
1872 "b.sql",
1873 "{% macro broken() %}\n{{ var('nope') }}\n{% endmacro %}\n{% macro quad(x) %}{{ x * 4 }}{% endmacro %}\n",
1874 ),
1875 ],
1876 );
1877 assert_eq!(
1878 render(&r, "select {{ double(2) }}, {{ quad(1) }}\n").unwrap(),
1879 "select 4, 4\n"
1880 );
1881 let e = render(&r, "line one\nselect {{ broken() }}").unwrap_err();
1882 assert_eq!(e.file, PathBuf::from("macros/b.sql"), "{e}");
1883 assert_eq!(e.line, Some(2), "{e}");
1884 let e = render(&r, "one\ntwo\n{{ nope }}").unwrap_err();
1885 assert_eq!(
1886 (e.file.clone(), e.line),
1887 (PathBuf::from("reports/q.sql"), Some(3)),
1888 "{e}"
1889 );
1890 }
1891
1892 struct Fake;
1893 impl QueryRunner for Fake {
1894 fn run_query(&self, sql: &str, max_rows: u64, _profile: Option<&str>) -> Result<QueryRows, String> {
1895 if max_rows < 2 {
1896 return Err(format!("`{sql}` returned more than {max_rows} rows"));
1897 }
1898 Ok(QueryRows {
1899 columns: vec!["region".into(), "n".into()],
1900 rows: vec![
1901 vec![Value::from("apac"), Value::from(1)],
1902 vec![Value::from("emea"), Value::from(2)],
1903 ],
1904 })
1905 }
1906 fn columns(&self, _: &str, _: Option<&str>) -> Result<Vec<Column>, String> {
1907 Ok(Vec::new())
1908 }
1909 }
1910
1911 #[test]
1912 fn run_query_rows_are_accessible_by_name_and_index() {
1913 let dir = tempfile::tempdir().unwrap();
1914 let r = Renderer::new(RendererConfig {
1915 root: dir.path(),
1916 macros: &[],
1917 context: RunContext {
1918 report: "r".into(),
1919 set: None,
1920 target: "t".into(),
1921 schedule: None,
1922 date: NaiveDate::from_ymd_opt(2026, 1, 1).unwrap(),
1923 now: Utc::now(),
1924 scheduled_at: None,
1925 calendar: Calendar::default(),
1926 locale: Default::default(),
1927 },
1928 vars: JsonMap::new(),
1929 cli_vars: BTreeMap::new(),
1930 runner: Some(Arc::new(Fake)),
1931 connections: None,
1932 mode: Mode::Run,
1933 connection: None,
1934 source_type: "duckdb".into(),
1935 sources: None,
1936 run_query_max_rows: 10_000,
1937 sql: BTreeMap::new(),
1938 lookups: BTreeMap::new(),
1939 lookup_inline_max_rows: 200,
1940 packages: Vec::new(),
1941 project_name: "acme".into(),
1942 dispatch: DispatchOrder::new(),
1943 })
1944 .unwrap();
1945 let src = "{% set res = run_query('select') %}{{ res.columns | join(',') }}|\
1946 {% for row in res.rows %}{{ row.region }}={{ row[1] }}{{ ',' if not loop.last }}{% endfor %}|{{ res | length }}|{{ res[1]['region'] }}";
1947 assert_eq!(render(&r, src).unwrap(), "region,n|apac=1,emea=2|2|emea");
1948 let e = render(&r, "{{ run_query('select', max_rows=1) }}").unwrap_err();
1949 assert!(e.message.contains("more than 1 rows"), "{e}");
1950 assert!(render(&r, "{% if run.set is none %}none{% endif %}").unwrap() == "none");
1951 }
1952
1953 #[test]
1954 fn removed_names_say_what_to_write_instead() {
1955 let e = err("{{ run.profile }}");
1956 assert!(
1957 e.contains("`run.profile` was removed") && e.contains("connection.name"),
1958 "{e}"
1959 );
1960 let e = err("{{ target.schema }}");
1961 assert!(
1962 e.contains("`target.schema` was removed") && e.contains("connection.schema"),
1963 "{e}"
1964 );
1965 }
1966
1967 #[test]
1968 fn a_source_relation_quotes_the_parts_asked_for() {
1969 let r = ResolvedSource {
1970 source: "sales".into(),
1971 table: "orders".into(),
1972 profile: None,
1973 database: None,
1974 schema: "Sales".into(),
1975 identifier: "Or\"ders".into(),
1976 quoting: Quoting {
1977 database: true,
1978 schema: false,
1979 identifier: true,
1980 },
1981 };
1982 assert_eq!(r.relation(Some("\"")), "Sales.\"Or\"\"ders\"");
1983 assert_eq!(r.relation(None), "Sales.Or\"ders");
1984 let r = ResolvedSource {
1985 database: Some("main".into()),
1986 ..r
1987 };
1988 assert_eq!(r.relation(Some("`")), "`main`.Sales.`Or\"ders`");
1989 }
1990
1991 #[test]
1992 fn a_relation_is_mentioned_only_as_a_whole_name() {
1993 assert!(mentions("select * from sales.orders where 1", "sales.orders"));
1994 assert!(!mentions("select * from sales.orders_old", "sales.orders"));
1995 assert!(!mentions("select * from x.sales.orders", "sales.orders"));
1996 }
1997}
1998
1999#[cfg(test)]
2000mod cli_var_tests {
2001 use super::cli_var_value as v;
2002 use serde_json::json;
2003
2004 #[test]
2005 fn yaml_1_2_core_rules() {
2006 assert_eq!(v("false"), json!(false));
2007 assert_eq!(v("TRUE"), json!(true));
2008 assert_eq!(v("~"), json!(null));
2009 assert_eq!(v("5"), json!(5));
2010 assert_eq!(v("-12"), json!(-12));
2011 assert_eq!(v("2.5"), json!(2.5));
2012 assert_eq!(v("[\"NAM\", EMEA]"), json!(["NAM", "EMEA"]));
2013 assert_eq!(v("{k: 1}"), json!({"k": 1}));
2014 for s in [
2015 "yes",
2016 "NO",
2017 "2026-01-31",
2018 "010",
2019 "1.0.0",
2020 "abc",
2021 "[unclosed",
2022 "0x1F",
2023 ] {
2024 assert_eq!(v(s), json!(s), "{s}");
2025 }
2026 assert_eq!(v("\"false\""), json!("false"));
2027 assert_eq!(v("'5'"), json!("5"));
2028 assert_eq!(v(""), json!(""));
2029 }
2030}