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 fn add_common(
733 env: &mut Environment<'static>,
734 context: RunContext,
735 vars: JsonMap<String, Json>,
736 cli_vars: BTreeMap<String, String>,
737) {
738 crate::mutable::register(env);
739 add_vars(env, vars, cli_vars);
740 env.add_function(
741 "env_var",
742 |name: String, _default: Option<Value>| -> Result<Value, Error> {
743 Err(Error::new(
744 ErrorKind::InvalidOperation,
745 format!("`env_var('{name}')` can't be used in a document template; pass the value as a var"),
746 ))
747 },
748 );
749 env.add_function("raise_error", |message: String| -> Result<Value, Error> {
750 Err(Error::new(ErrorKind::InvalidOperation, message))
751 });
752 crate::dates::register(env, context.calendar, context.date);
753 crate::numbers::register(env, context.locale);
754 env.add_global("target", Value::from_object(Target(context.target.clone())));
755 env.add_global("run", Value::from_object(Run(context)));
756}
757
758pub struct Limited {
763 env: Environment<'static>,
764}
765
766const LIMITED_FUNCTIONS: &[&str] = &[
768 "run_query",
769 "columns",
770 "source",
771 "ref",
772 "lookup",
773 "dispatch",
774 "profile",
775];
776const LIMITED_GLOBALS: &[&str] = &["connection", "destination"];
777
778impl Limited {
779 pub fn new(
780 context: RunContext,
781 vars: JsonMap<String, Json>,
782 cli_vars: BTreeMap<String, String>,
783 ) -> Limited {
784 let mut env = Environment::new();
785 env.set_undefined_behavior(UndefinedBehavior::Strict);
786 crate::mutable::register(&mut env);
787 add_vars(&mut env, vars, cli_vars);
788 env.add_function("raise_error", |message: String| -> Result<Value, Error> {
789 Err(Error::new(ErrorKind::InvalidOperation, message))
790 });
791 crate::dates::register(&mut env, context.calendar, context.date);
792 crate::numbers::register(&mut env, context.locale);
793 env.add_global("target", Value::from_object(Target(context.target.clone())));
794 env.add_global("run", Value::from_object(Run(context)));
795 for f in LIMITED_FUNCTIONS {
796 let name = *f;
797 env.add_function(name, move |_args: minijinja::value::Rest<Value>| -> Result<Value, Error> {
798 Err(Error::new(
799 ErrorKind::InvalidOperation,
800 format!("`{name}()` can't be used here: this value chooses a connection or a source, so it's rendered before any connection is open"),
801 ))
802 });
803 }
804 for g in LIMITED_GLOBALS {
805 env.add_global(*g, Value::from_object(Forbidden(g)));
806 }
807 Limited { env }
808 }
809
810 pub fn render(&self, key: &str, src: &str) -> Result<String, String> {
812 if !crate::preflight::is_templated(src) {
813 return Ok(src.to_string());
814 }
815 self.env
816 .render_str(src, ())
817 .map(|v| v.trim().to_string())
818 .map_err(|e| {
819 let mut deepest: &Error = &e;
820 while let Some(s) = std::error::Error::source(deepest).and_then(|s| s.downcast_ref::<Error>())
821 {
822 deepest = s;
823 }
824 let msg = deepest
825 .detail()
826 .map(str::to_string)
827 .unwrap_or_else(|| deepest.kind().to_string());
828 format!("{key} `{src}`: {msg}")
829 })
830 }
831}
832
833#[derive(Debug)]
835struct Forbidden(&'static str);
836
837impl Object for Forbidden {
838 fn get_value(self: &Arc<Self>, _key: &Value) -> Option<Value> {
839 Some(Value::from(Error::new(
840 ErrorKind::InvalidOperation,
841 format!(
842 "`{}.*` can't be used here: this value chooses a connection or a source, so it's rendered before any connection is open",
843 self.0
844 ),
845 )))
846 }
847
848 fn render(self: &Arc<Self>, _f: &mut fmt::Formatter<'_>) -> fmt::Result {
849 Err(fmt::Error)
850 }
851}
852
853fn package_template(name: &str) -> String {
854 format!("__dre_package_{name}__")
855}
856
857#[allow(clippy::type_complexity)]
860fn combine(
861 files: &[(PathBuf, PathBuf)],
862) -> Result<(String, Vec<String>, Vec<(usize, PathBuf, usize)>), RenderError> {
863 let re = regex::Regex::new(r"\{%-?\s*macro\s+([A-Za-z_]\w*)").unwrap();
864 let mut combined = String::new();
865 let mut names = Vec::new();
866 let mut spans = Vec::new();
867 for (read, shown) in files {
868 let src = std::fs::read_to_string(read).map_err(|e| RenderError {
869 file: shown.clone(),
870 line: None,
871 message: format!("can't read macro file: {e}"),
872 })?;
873 let start = combined.matches('\n').count() + 1;
874 names.extend(re.captures_iter(&src).map(|c| c[1].to_string()));
875 combined.push_str(&src);
876 if !combined.ends_with('\n') {
877 combined.push('\n');
878 }
879 spans.push((start, shown.clone(), src.matches('\n').count() + 1));
880 }
881 Ok((combined, names, spans))
882}
883
884fn add_dispatch(
889 env: &mut Environment<'static>,
890 project: &str,
891 project_has_macros: bool,
892 packages: &[Package],
893 order: DispatchOrder,
894 source_type: String,
895 parse: bool,
896) {
897 let project = project.to_string();
898 let templates: BTreeMap<String, String> = packages
899 .iter()
900 .map(|p| (p.name.clone(), package_template(&p.name)))
901 .collect();
902 env.add_function(
903 "dispatch",
904 move |state: &minijinja::State<'_, '_>,
905 name: String,
906 namespace: Option<String>|
907 -> Result<Value, Error> {
908 let ns = namespace.unwrap_or_else(|| project.clone());
909 let search = order.get(&ns).cloned().unwrap_or_else(|| {
910 if ns == project {
911 vec![ns.clone()]
912 } else {
913 vec![project.clone(), ns.clone()]
914 }
915 });
916 let candidates = [format!("{source_type}__{name}"), format!("default__{name}")];
917 for n in &search {
918 let template = if n == &project {
919 if !project_has_macros {
920 continue;
921 }
922 MACROS.to_string()
923 } else {
924 match templates.get(n) {
925 Some(t) => t.clone(),
926 None => continue,
927 }
928 };
929 let tmpl = state.env().get_template(&template)?;
930 let captured = tmpl.render_captured(())?;
931 let st = captured.state();
932 for c in &candidates {
933 if st.lookup(c).is_some_and(|v| !v.is_undefined()) {
934 return Ok(Value::from_object(Dispatched {
935 template,
936 name: c.clone(),
937 }));
938 }
939 }
940 }
941 if parse {
942 return Ok(Value::from_object(Dispatched {
945 template: String::new(),
946 name: String::new(),
947 }));
948 }
949 Err(Error::new(
950 ErrorKind::InvalidOperation,
951 format!(
952 "`dispatch('{name}', '{ns}')`: no `{}` or `{}` in {}",
953 candidates[0],
954 candidates[1],
955 search.join(", ")
956 ),
957 ))
958 },
959 );
960}
961
962#[derive(Debug)]
964struct Dispatched {
965 template: String,
966 name: String,
967}
968
969impl Object for Dispatched {
970 fn call(self: &Arc<Self>, state: &mut minijinja::State<'_, '_>, args: &[Value]) -> Result<Value, Error> {
971 if self.template.is_empty() {
972 return Ok(Value::from(""));
973 }
974 let tmpl = state.env().get_template(&self.template)?;
975 let out = tmpl
976 .render_captured(())?
977 .with_state_mut(|s| s.call_macro(&self.name, args))?;
978 Ok(Value::from(out))
979 }
980}
981
982fn add_ref(
986 env: &mut Environment<'static>,
987 root: &Path,
988 sql: BTreeMap<String, PathBuf>,
989 lookups: Arc<Lookups>,
990 import: &str,
991) {
992 let root = root.to_path_buf();
993 let import = import.to_string();
994 let stack: Arc<Mutex<Vec<String>>> = Arc::default();
996 env.add_function(
997 "ref",
998 move |state: &minijinja::State<'_, '_>, name: String| -> Result<Value, Error> {
999 if lookups.defs.contains_key(&name) {
1000 return lookups.relation(&name).map(Value::from);
1001 }
1002 let Some(rel) = sql.get(&name) else {
1003 return Err(Error::new(
1004 ErrorKind::InvalidOperation,
1005 format!("`ref('{name}')`: no `{name}.sql` and no lookup `{name}` in the project"),
1006 ));
1007 };
1008 {
1009 let mut chain = stack.lock().unwrap();
1010 if chain.contains(&name) {
1011 chain.push(name.clone());
1012 let cycle = chain.join(" → ");
1013 chain.clear();
1014 return Err(Error::new(
1015 ErrorKind::InvalidOperation,
1016 format!("`ref()` cycle: {cycle}"),
1017 ));
1018 }
1019 chain.push(name.clone());
1020 }
1021 let result = (|| {
1022 let src = std::fs::read_to_string(root.join(rel)).map_err(|e| {
1023 Error::new(
1024 ErrorKind::InvalidOperation,
1025 format!("can't read {}: {e}", rel.display()),
1026 )
1027 })?;
1028 let file = rel.to_string_lossy().to_string();
1029 let rendered = state
1030 .env()
1031 .template_from_named_str(&file, &format!("{import}{src}"))?
1032 .render(())?;
1033 let statements = crate::sqlsplit::split(&rendered);
1034 match statements.as_slice() {
1035 [one] => Ok(Value::from(format!("(\n{}\n)", one.text))),
1036 _ => Err(Error::new(
1037 ErrorKind::InvalidOperation,
1038 format!(
1039 "`ref('{name}')` needs {} to hold exactly one statement, but it has {}",
1040 rel.display(),
1041 statements.len()
1042 ),
1043 )),
1044 }
1045 })();
1046 let mut chain = stack.lock().unwrap();
1047 if chain.last() == Some(&name) {
1048 chain.pop();
1049 }
1050 result
1051 },
1052 );
1053}
1054
1055struct Lookups {
1057 connections: Option<Arc<dyn Connections>>,
1059 kind: String,
1060 style: std::sync::OnceLock<Result<lookups::StringLiteral, String>>,
1061 root: PathBuf,
1062 defs: BTreeMap<String, Lookup>,
1063 inline_max: u64,
1064 runner: Option<Arc<dyn QueryRunner>>,
1065 tables: Mutex<BTreeMap<String, Arc<Table>>>,
1066 loaded: Mutex<BTreeMap<String, String>>,
1068 warnings: Arc<Mutex<Vec<String>>>,
1069}
1070
1071impl Lookups {
1072 fn table(&self, name: &str) -> Result<Arc<Table>, Error> {
1073 let def = self.defs.get(name).ok_or_else(|| {
1074 Error::new(
1075 ErrorKind::InvalidOperation,
1076 format!("no lookup `{name}` under lookups/"),
1077 )
1078 })?;
1079 if let Some(t) = self.tables.lock().unwrap().get(name) {
1080 return Ok(t.clone());
1081 }
1082 let t = Arc::new(lookups::read(&self.root, def).map_err(|e| {
1083 Error::new(
1084 ErrorKind::InvalidOperation,
1085 format!("{}: {e}", def.file.display()),
1086 )
1087 })?);
1088 self.tables.lock().unwrap().insert(name.to_string(), t.clone());
1089 Ok(t)
1090 }
1091
1092 fn relation(&self, name: &str) -> Result<String, Error> {
1093 if let Some(r) = self.loaded.lock().unwrap().get(name) {
1094 return Ok(r.clone());
1095 }
1096 let t = self.table(name)?;
1097 let rows = t.rows.len() as u64;
1098 let load = match self.defs[name].load {
1099 Load::Inline => false,
1100 Load::TempTable => true,
1101 Load::Auto => rows > self.inline_max,
1102 };
1103 let loaded = match (&self.runner, load) {
1104 (Some(runner), true) => runner.load(name, &t).map_err(|e| {
1105 Error::new(
1106 ErrorKind::InvalidOperation,
1107 format!("loading lookup `{name}`: {e}"),
1108 )
1109 })?,
1110 _ => None,
1111 };
1112 let relation = match loaded {
1113 Some((relation, warning)) => {
1114 if let Some(w) = warning {
1115 self.warn(format!("lookup `{name}`: {w}"));
1116 }
1117 relation
1118 }
1119 None => {
1120 if load && self.runner.is_some() {
1121 self.warn(format!(
1122 "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"
1123 ));
1124 }
1125 t.inline_sql(self.style()?)
1126 }
1127 };
1128 self.loaded
1129 .lock()
1130 .unwrap()
1131 .insert(name.to_string(), relation.clone());
1132 Ok(relation)
1133 }
1134
1135 fn warn(&self, w: String) {
1136 self.warnings.lock().unwrap().push(w);
1137 }
1138
1139 fn style(&self) -> Result<lookups::StringLiteral, Error> {
1142 self.style
1143 .get_or_init(|| match &self.connections {
1144 Some(c) if !self.kind.is_empty() => {
1145 let asked = c.string_literal(&self.kind).unwrap_or(None);
1146 lookups::StringLiteral::from_describe(asked.as_deref())
1147 }
1148 _ => Ok(lookups::StringLiteral::Chr),
1149 })
1150 .clone()
1151 .map_err(|e| {
1152 Error::new(
1153 ErrorKind::InvalidOperation,
1154 format!("the `{}` source: {e}", self.kind),
1155 )
1156 })
1157 }
1158}
1159
1160fn cell_value(c: &Cell) -> Value {
1161 match c {
1162 Cell::Null => Value::from(()),
1163 Cell::Text(s) => Value::from(s.clone()),
1164 Cell::Int(n) => Value::from(*n),
1165 Cell::Num(n) => Value::from(*n),
1166 Cell::Bool(b) => Value::from(*b),
1167 Cell::Date(d) => Value::from(d.to_string()),
1168 }
1169}
1170
1171#[derive(Debug)]
1172struct Run(RunContext);
1173
1174impl Object for Run {
1175 fn get_value(self: &Arc<Self>, key: &Value) -> Option<Value> {
1176 let c = &self.0;
1177 Some(match key.as_str()? {
1178 "report" => Value::from(c.report.clone()),
1179 "set" => c.set.clone().map(Value::from).unwrap_or(Value::from(())),
1180 "target" => Value::from(c.target.clone()),
1181 "profile" => removed("run.profile", "`connection.name` (the query's connection)"),
1182 "source_type" => removed("run.source_type", "`connection.type` (the query's connection)"),
1183 "schedule" => c.schedule.clone().map(Value::from).unwrap_or(Value::from(())),
1184 "date" => Date::value(c.date, c.calendar),
1185 "now" => DateTime::now(c.now, c.calendar),
1186 "scheduled_at" => c
1187 .scheduled_at
1188 .map(|t| DateTime::now(t, c.calendar))
1189 .unwrap_or(Value::from(())),
1190 "timezone" => Value::from(c.calendar.tz.name()),
1191 _ => return None,
1192 })
1193 }
1194
1195 fn call_method(
1196 self: &Arc<Self>,
1197 _: &mut minijinja::State<'_, '_>,
1198 method: &str,
1199 args: &[Value],
1200 ) -> Result<Value, Error> {
1201 match method {
1202 "date_format" => {
1203 let fmt = args.first().and_then(|a| a.as_str()).ok_or_else(|| {
1204 Error::new(
1205 ErrorKind::InvalidOperation,
1206 "`run.date_format()` takes a strftime format string",
1207 )
1208 })?;
1209 let mut out = String::new();
1210 use std::fmt::Write;
1211 write!(out, "{}", self.0.date.format(fmt)).map_err(|_| {
1212 Error::new(
1213 ErrorKind::InvalidOperation,
1214 format!("invalid date format `{fmt}`"),
1215 )
1216 })?;
1217 Ok(Value::from(out))
1218 }
1219 _ => Err(Error::from(ErrorKind::UnknownMethod)),
1220 }
1221 }
1222}
1223
1224fn removed(name: &str, instead: &str) -> Value {
1226 Value::from(Error::new(
1227 ErrorKind::InvalidOperation,
1228 format!("`{name}` was removed in DRE 0.2: use {instead}"),
1229 ))
1230}
1231
1232fn check_role(role: Option<&str>) -> Result<(), Error> {
1234 match role {
1235 None | Some("connection") | Some("destination") => Ok(()),
1236 Some("source") => Err(Error::new(
1237 ErrorKind::InvalidOperation,
1238 "`role='source'` was renamed in DRE 0.2: use `role='connection'`",
1239 )),
1240 Some(r) => Err(Error::new(
1241 ErrorKind::InvalidOperation,
1242 format!("`role='{r}'`: role must be 'connection' or 'destination'"),
1243 )),
1244 }
1245}
1246
1247fn choose_profile(
1251 what: &str,
1252 sql: &str,
1253 explicit: Option<String>,
1254 relations: &Mutex<Vec<(String, ResolvedSource)>>,
1255) -> Result<Option<String>, Error> {
1256 let mut used: Vec<(String, String)> = Vec::new();
1257 for (text, r) in relations.lock().unwrap().iter() {
1258 if let Some(p) = &r.profile
1259 && mentions(sql, text)
1260 && !used.iter().any(|(k, _)| *k == r.key())
1261 {
1262 used.push((r.key(), p.clone()));
1263 }
1264 }
1265 let fail = |m: String| Error::new(ErrorKind::InvalidOperation, format!("`{what}`: {m}"));
1266 match explicit {
1267 Some(p) => {
1268 if let Some((k, q)) = used.iter().find(|(_, q)| *q != p) {
1269 return Err(fail(format!(
1270 "`profile='{p}'` disagrees with source `{k}`, which is on connection `{q}`"
1271 )));
1272 }
1273 Ok(Some(p))
1274 }
1275 None => match used.as_slice() {
1276 [] => Ok(None),
1277 [(_, p), rest @ ..] => {
1278 if let Some((k, q)) = rest.iter().find(|(_, q)| q != p) {
1279 return Err(fail(format!(
1280 "reads source `{}` on connection `{p}` and source `{k}` on `{q}`; one statement runs on one connection",
1281 used[0].0
1282 )));
1283 }
1284 Ok(Some(p.clone()))
1285 }
1286 },
1287 }
1288}
1289
1290fn mentions(sql: &str, text: &str) -> bool {
1292 let ident = |c: char| c.is_alphanumeric() || c == '_';
1293 sql.match_indices(text).any(|(i, _)| {
1294 let before = sql[..i].chars().next_back();
1295 let after = sql[i + text.len()..].chars().next();
1296 !before.is_some_and(|c| ident(c) || c == '.') && !after.is_some_and(|c| ident(c) || c == '.')
1297 })
1298}
1299
1300#[derive(Debug)]
1303struct Target(String);
1304
1305impl Object for Target {
1306 fn get_value(self: &Arc<Self>, key: &Value) -> Option<Value> {
1307 let k = key.as_str()?;
1308 Some(match k {
1309 "name" => Value::from(self.0.clone()),
1310 "type" => removed("target.type", "`connection.type` (the query's connection)"),
1311 "profile" => removed("target.profile", "`connection.name` (the query's connection)"),
1312 _ => removed(
1313 &format!("target.{k}"),
1314 &format!(
1315 "`connection.{k}` (the query's connection) or `profile('<name>').{k}`; `target` is now only the environment, `target.name`"
1316 ),
1317 ),
1318 })
1319 }
1320
1321 fn render(self: &Arc<Self>, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1322 f.write_str(&self.0)
1323 }
1324}
1325
1326#[derive(Debug)]
1329struct Lenient;
1330
1331impl Object for Lenient {
1332 fn get_value(self: &Arc<Self>, _key: &Value) -> Option<Value> {
1333 None
1334 }
1335
1336 fn render(self: &Arc<Self>, _f: &mut fmt::Formatter<'_>) -> fmt::Result {
1337 Ok(())
1338 }
1339}
1340
1341#[derive(Debug)]
1343struct ConnectionValue(Connection);
1344
1345impl Object for ConnectionValue {
1346 fn get_value(self: &Arc<Self>, key: &Value) -> Option<Value> {
1347 connection_field(&self.0, key)
1348 }
1349
1350 fn render(self: &Arc<Self>, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1351 write!(f, "{}", self.0.profile)
1352 }
1353}
1354
1355struct LazyConnection {
1358 connections: Option<Arc<dyn Connections>>,
1359 name: Option<String>,
1360 resolved: std::sync::OnceLock<Result<Connection, String>>,
1361}
1362
1363impl fmt::Debug for LazyConnection {
1364 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1365 f.write_str("connection")
1366 }
1367}
1368
1369impl LazyConnection {
1370 fn get(&self) -> &Result<Connection, String> {
1371 self.resolved.get_or_init(|| {
1372 let name = self.name.as_deref().ok_or(
1373 "no connection here: this isn't a query's SQL and the Binding inherits no `profile`",
1374 )?;
1375 match &self.connections {
1376 Some(c) => c.profile(name, Some("connection")),
1377 None => Err("no profiles.yml is loaded here".into()),
1378 }
1379 })
1380 }
1381}
1382
1383impl Object for LazyConnection {
1384 fn get_value(self: &Arc<Self>, key: &Value) -> Option<Value> {
1385 match self.get() {
1386 Ok(c) => connection_field(c, key),
1387 Err(e) => Some(Value::from(Error::new(
1388 ErrorKind::InvalidOperation,
1389 format!("`connection`: {e}"),
1390 ))),
1391 }
1392 }
1393
1394 fn render(self: &Arc<Self>, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1395 match &self.name {
1396 Some(n) => f.write_str(n),
1397 None => Err(fmt::Error),
1398 }
1399 }
1400}
1401
1402#[derive(Debug)]
1404struct DestinationSlot(Arc<Mutex<Option<Connection>>>);
1405
1406impl Object for DestinationSlot {
1407 fn get_value(self: &Arc<Self>, key: &Value) -> Option<Value> {
1408 match &*self.0.lock().unwrap() {
1409 Some(c) => connection_field(c, key),
1410 None => Some(Value::from(Error::new(
1411 ErrorKind::InvalidOperation,
1412 "`destination` only exists while rendering a destination's `path` and options",
1413 ))),
1414 }
1415 }
1416
1417 fn render(self: &Arc<Self>, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1418 match &*self.0.lock().unwrap() {
1419 Some(c) => f.write_str(&c.profile),
1420 None => Err(fmt::Error),
1421 }
1422 }
1423}
1424
1425fn connection_field(c: &Connection, key: &Value) -> Option<Value> {
1427 {
1428 let k = key.as_str()?;
1429 Some(match k {
1430 "name" | "profile" => Value::from(c.profile.clone()),
1431 "target" => Value::from(c.target.clone()),
1432 "type" => Value::from(c.kind.clone()),
1433 _ if c.secrets.iter().any(|s| s == k) => Value::from(Error::new(
1434 ErrorKind::InvalidOperation,
1435 format!(
1436 "`{k}` of profile `{}` holds a secret, so templates can't read it (it would end up in compiled SQL and logs)",
1437 c.profile
1438 ),
1439 )),
1440 _ => match c.fields.get(k) {
1441 Some(v) => Value::from(minijinja::value::Serde(v)),
1442 None => Value::from(Error::new(
1443 ErrorKind::UndefinedError,
1444 format!(
1445 "profile `{}` (target `{}`) has no field `{k}`; it has: {}",
1446 c.profile,
1447 c.target,
1448 c.fields
1449 .keys()
1450 .filter(|f| !c.secrets.contains(f))
1451 .cloned()
1452 .collect::<Vec<_>>()
1453 .join(", ")
1454 ),
1455 )),
1456 },
1457 })
1458 }
1459}
1460
1461#[derive(Debug)]
1462struct QueryResult {
1463 columns: Vec<String>,
1464 rows: Vec<Value>,
1465 lenient: bool,
1467}
1468
1469impl QueryResult {
1470 fn new(q: QueryRows) -> QueryResult {
1471 let cols: Arc<Vec<String>> = Arc::new(q.columns.clone());
1472 let rows = q
1473 .rows
1474 .into_iter()
1475 .map(|values| {
1476 Value::from_object(Row {
1477 columns: cols.clone(),
1478 values,
1479 })
1480 })
1481 .collect();
1482 QueryResult {
1483 columns: q.columns,
1484 rows,
1485 lenient: false,
1486 }
1487 }
1488}
1489
1490impl Object for QueryResult {
1491 fn repr(self: &Arc<Self>) -> ObjectRepr {
1492 ObjectRepr::Seq
1493 }
1494
1495 fn get_value(self: &Arc<Self>, key: &Value) -> Option<Value> {
1496 match key.as_str() {
1497 Some("rows") => Some(Value::from(self.rows.clone())),
1498 Some("columns") => Some(Value::from(self.columns.clone())),
1499 _ => self.rows.get(key.as_usize()?).cloned(),
1500 }
1501 }
1502
1503 fn enumerate(self: &Arc<Self>) -> Enumerator {
1504 Enumerator::Seq(self.rows.len())
1505 }
1506
1507 fn call_method(
1509 self: &Arc<Self>,
1510 _: &mut minijinja::State<'_, '_>,
1511 method: &str,
1512 args: &[Value],
1513 ) -> Result<Value, Error> {
1514 if method != "column" {
1515 return Err(Error::from(ErrorKind::UnknownMethod));
1516 }
1517 if self.lenient {
1518 return Ok(Value::from(Vec::<Value>::new()));
1519 }
1520 let [key] = args else {
1521 return Err(Error::new(
1522 ErrorKind::InvalidOperation,
1523 "`column()` takes one column name or index",
1524 ));
1525 };
1526 let i = match key.as_str() {
1527 Some(n) => self.columns.iter().position(|c| c == n).ok_or_else(|| {
1528 Error::new(
1529 ErrorKind::InvalidOperation,
1530 format!("no column `{n}` (columns: {})", self.columns.join(", ")),
1531 )
1532 })?,
1533 None => key
1534 .as_usize()
1535 .filter(|i| *i < self.columns.len())
1536 .ok_or_else(|| Error::new(ErrorKind::InvalidOperation, format!("no column {key}")))?,
1537 };
1538 Ok(Value::from(
1539 self.rows
1540 .iter()
1541 .map(|r| r.get_item_by_index(i).unwrap_or_default())
1542 .collect::<Vec<_>>(),
1543 ))
1544 }
1545}
1546
1547#[derive(Debug)]
1549pub(crate) struct Row {
1550 columns: Arc<Vec<String>>,
1551 values: Vec<Value>,
1552}
1553
1554impl Row {
1555 pub(crate) fn value(columns: Arc<Vec<String>>, values: Vec<Value>) -> Value {
1556 Value::from_object(Row { columns, values })
1557 }
1558}
1559
1560impl Object for Row {
1561 fn repr(self: &Arc<Self>) -> ObjectRepr {
1562 ObjectRepr::Seq
1563 }
1564
1565 fn get_value(self: &Arc<Self>, key: &Value) -> Option<Value> {
1566 if let Some(name) = key.as_str() {
1567 let i = self.columns.iter().position(|c| c == name)?;
1568 return self.values.get(i).cloned();
1569 }
1570 self.values.get(key.as_usize()?).cloned()
1571 }
1572
1573 fn enumerate(self: &Arc<Self>) -> Enumerator {
1574 Enumerator::Seq(self.values.len())
1575 }
1576}
1577
1578#[cfg(test)]
1579mod tests {
1580 use super::*;
1581
1582 fn renderer(vars: Json, cli: &[(&str, &str)], macros: &[(&str, &str)]) -> (tempfile::TempDir, Renderer) {
1583 let dir = tempfile::tempdir().unwrap();
1584 let mut files = Vec::new();
1585 for (name, src) in macros {
1586 let p = PathBuf::from("macros").join(name);
1587 std::fs::create_dir_all(dir.path().join("macros")).unwrap();
1588 std::fs::write(dir.path().join(&p), src).unwrap();
1589 files.push(p);
1590 }
1591 let r = Renderer::new(RendererConfig {
1592 root: dir.path(),
1593 macros: &files,
1594 context: RunContext {
1595 report: "monthly".into(),
1596 set: Some("client_a".into()),
1597 target: "prod".into(),
1598 schedule: None,
1599 date: NaiveDate::from_ymd_opt(2026, 1, 25).unwrap(),
1600 now: Utc::now(),
1601 scheduled_at: None,
1602 calendar: Calendar::default(),
1603 locale: Default::default(),
1604 },
1605 vars: vars.as_object().cloned().unwrap_or_default(),
1606 cli_vars: cli.iter().map(|(k, v)| (k.to_string(), v.to_string())).collect(),
1607 runner: None,
1608 connections: None,
1609 mode: Mode::Run,
1610 connection: None,
1611 source_type: "duckdb".into(),
1612 sources: None,
1613 run_query_max_rows: 10_000,
1614 sql: BTreeMap::new(),
1615 lookups: BTreeMap::new(),
1616 lookup_inline_max_rows: 200,
1617 packages: Vec::new(),
1618 project_name: "acme".into(),
1619 dispatch: DispatchOrder::new(),
1620 })
1621 .unwrap();
1622 (dir, r)
1623 }
1624
1625 fn render(r: &Renderer, src: &str) -> Result<String, RenderError> {
1626 r.render(Path::new("reports/q.sql"), src)
1627 }
1628
1629 fn ok(src: &str) -> String {
1630 let (_d, r) = renderer(serde_json::json!({"fixed": [3, 1, 2], "m": {"a": 1}}), &[], &[]);
1631 render(&r, src).unwrap_or_else(|e| panic!("{src}: {e}"))
1632 }
1633
1634 fn err(src: &str) -> String {
1635 let (_d, r) = renderer(serde_json::json!({"fixed": [3, 1, 2], "m": {"a": 1}}), &[], &[]);
1636 render(&r, src).unwrap_err().to_string()
1637 }
1638
1639 #[test]
1640 fn python_string_and_dict_methods_work() {
1641 assert_eq!(ok("{{ 'a,b,c'.split(',') | join('-') }}"), "a-b-c");
1642 assert_eq!(
1643 ok("{{ 'y' if 'abc'.startswith('a') }}{{ 'y' if 'abc'.endswith('x') }}"),
1644 "y"
1645 );
1646 assert_eq!(
1647 ok("{{ ' x '.strip() }}|{{ 'abc'.upper() }}|{{ 'a-b'.replace('-', '_') }}"),
1648 "x|ABC|a_b"
1649 );
1650 assert_eq!(
1651 ok("{{ var('m').keys() | list }} {{ var('m').items() | list }}"),
1652 "['a'] [('a', 1)]"
1653 );
1654 assert_eq!(
1655 ok("{{ var('m').get('a') }} {{ var('m').get('z', 'none') }}"),
1656 "1 none"
1657 );
1658 }
1659
1660 #[test]
1661 fn list_appends_in_a_loop_and_joins() {
1662 assert_eq!(
1663 ok(
1664 "{% set sq = list() %}{% for i in range(1, 5) %}{% set _ = sq.append(i * i) %}{% endfor %}{{ sq | join('+') }}"
1665 ),
1666 "1+4+9+16"
1667 );
1668 }
1669
1670 #[test]
1671 fn list_methods_follow_python() {
1672 assert_eq!(
1673 ok("{% set l = list([3, 1, 2]) %}{% set _ = l.sort() %}{{ l }}"),
1674 "[1, 2, 3]"
1675 );
1676 assert_eq!(
1677 ok("{% set l = list([3, 1, 2]) %}{% set _ = l.sort(reverse=true) %}{{ l }}"),
1678 "[3, 2, 1]"
1679 );
1680 assert_eq!(
1681 ok("{% set l = list([1, 2]) %}{% set _ = l.extend([3, 4]) %}{{ l }} {{ l | length }}"),
1682 "[1, 2, 3, 4] 4"
1683 );
1684 assert_eq!(
1685 ok("{% set l = list([1, 2, 3]) %}{{ l.pop() }} {{ l.pop(0) }} {{ l }}"),
1686 "3 1 [2]"
1687 );
1688 assert_eq!(
1689 ok("{% set l = list([1, 2]) %}{% set _ = l.insert(0, 9) %}{% set _ = l.insert(99, 7) %}{{ l }}"),
1690 "[9, 1, 2, 7]"
1691 );
1692 assert_eq!(
1693 ok(
1694 "{% set l = list([1, 2, 1]) %}{% set _ = l.remove(1) %}{{ l }} {{ l.index(1) }} {{ l.count(1) }}"
1695 ),
1696 "[2, 1] 1 1"
1697 );
1698 assert_eq!(
1699 ok("{% set l = list([1, 2]) %}{% set _ = l.reverse() %}{{ l }}"),
1700 "[2, 1]"
1701 );
1702 assert_eq!(
1703 ok("{% set l = list([1, 2]) %}{% set _ = l.clear() %}{{ l }} {{ 'empty' if not l }}"),
1704 "[] empty"
1705 );
1706 assert_eq!(
1707 ok("{% set l = list([1, 2, 3]) %}{{ l[0] }} {{ l[-1] }} {{ l[1:] }}"),
1708 "1 3 [2, 3]"
1709 );
1710 }
1711
1712 #[test]
1713 fn lists_share_by_reference_and_copy_apart() {
1714 assert_eq!(
1715 ok("{% set a = list() %}{% set b = a %}{% set _ = b.append(1) %}{{ a }}"),
1716 "[1]"
1717 );
1718 assert_eq!(
1719 ok("{% set a = list([1]) %}{% set b = a.copy() %}{% set _ = b.append(2) %}{{ a }} {{ b }}"),
1720 "[1] [1, 2]"
1721 );
1722 }
1723
1724 #[test]
1725 fn a_list_changed_inside_its_own_loop_does_not_skip_or_repeat() {
1726 assert_eq!(
1727 ok(
1728 "{% set l = list([1, 2, 3]) %}{% for x in l %}{% set _ = l.append(x * 10) %}{{ x }} {% endfor %}{{ l }}"
1729 ),
1730 "1 2 3 [1, 2, 3, 10, 20, 30]"
1731 );
1732 assert_eq!(
1733 ok("{% set l = list([1, 2]) %}{% set _ = l.extend(l) %}{{ l }}"),
1734 "[1, 2, 1, 2]"
1735 );
1736 }
1737
1738 #[test]
1739 fn a_list_starts_from_a_var_without_changing_it() {
1740 assert_eq!(
1741 ok("{% set l = list(var('fixed')) %}{% set _ = l.append(4) %}{{ l }} {{ var('fixed') }}"),
1742 "[3, 1, 2, 4] [3, 1, 2]"
1743 );
1744 }
1745
1746 #[test]
1747 fn dict_methods_follow_python() {
1748 assert_eq!(
1749 ok("{% set d = dict(a=1) %}{% set _ = d.update(b=2) %}{% set _ = d.update({'c': 3}) %}{{ d }}"),
1750 "{'a': 1, 'b': 2, 'c': 3}"
1751 );
1752 let make = "{% set d = dict() %}{% set _ = d.setdefault('b', 1) %}{% set _ = d.setdefault('a', 2) %}";
1755 assert_eq!(
1756 ok(&(make.to_string() + "{% for k, v in d.items() %}{{ k }}={{ v }} {% endfor %}")),
1757 "b=1 a=2 "
1758 );
1759 assert_eq!(
1760 ok(&(make.to_string() + "{% for k in d %}{{ k }} {% endfor %}{{ d.values() | list }}")),
1761 "b a [1, 2]"
1762 );
1763 assert_eq!(
1764 ok("{% set d = dict(a=1) %}{{ d.setdefault('a', 9) }} {{ d.setdefault('b', 2) }} {{ d }}"),
1765 "1 2 {'a': 1, 'b': 2}"
1766 );
1767 assert_eq!(
1768 ok("{% set d = dict(a=1) %}{{ d.pop('a') }} {{ d.pop('a', 'gone') }} {{ d }}"),
1769 "1 gone {}"
1770 );
1771 assert_eq!(
1772 ok(
1773 "{% set d = dict(a=1) %}{{ d.get('a') }} {{ d['a'] }} {{ d.a }} {{ 'in' if 'a' in d }} {{ d.keys() | list }}"
1774 ),
1775 "1 1 1 in ['a']"
1776 );
1777 assert_eq!(
1778 ok("{% set d = dict(a=1) %}{% set e = d.copy() %}{% set _ = e.update(a=2) %}{{ d.a }} {{ e.a }}"),
1779 "1 2"
1780 );
1781 assert_eq!(
1782 ok("{% set d = dict(a=1) %}{% set _ = d.update(a=5, b=6) %}{{ d | length }} {{ d.a }}"),
1783 "2 5"
1784 );
1785 assert_eq!(
1786 ok("{% set d = dict(a=1) %}{% set _ = d.update(d) %}{{ d }}"),
1787 "{'a': 1}"
1788 );
1789 }
1790
1791 #[test]
1792 fn mutable_values_print_and_serialise_like_plain_ones() {
1793 assert_eq!(
1794 ok("{{ list([1, 'a']) }} {{ dict(k='v') }}"),
1795 "[1, 'a'] {'k': 'v'}"
1796 );
1797 assert_eq!(
1798 ok("{{ list(dict(k='v')) }} {{ list(list([2, 1])) | sort }}"),
1799 "['k'] [1, 2]"
1800 );
1801 assert_eq!(ok("{% set l = list() %}{{ 'yes' if l else 'no' }}"), "no");
1802 }
1803
1804 #[test]
1806 fn expressions_follow_jinja2() {
1807 assert_eq!(
1808 ok("{{ ['a', 1] }} {{ var('m') }} {{ var('m').items() | list }}"),
1809 "['a', 1] {'a': 1} [('a', 1)]"
1810 );
1811 assert_eq!(ok("{{ 2.5 | round }} {{ 3.5 | round }}"), "2.0 4.0");
1812 assert_eq!(ok("{{ -7 // 2 }} {{ -7 % 3 }}"), "-4 2");
1813 assert_eq!(
1814 ok("{{ ['b', 'P', 'n'] | max }} {{ \"don't\" | title }}"),
1815 "P Don't"
1816 );
1817 assert!(err("{{ 1 / 0 }}").contains("unable to calculate 1 / 0"));
1818 }
1819
1820 #[test]
1821 fn mutating_an_immutable_value_says_what_to_write() {
1822 let e = err("{% set l = [] %}{% set _ = l.append(1) %}");
1823 assert!(e.contains("append()") && e.contains("list()"), "{e}");
1824 let e = err("{% set _ = var('fixed').append(1) %}");
1825 assert!(e.contains("list()"), "{e}");
1826 let e = err("{% set d = {} %}{% set _ = d.update(a=1) %}");
1827 assert!(e.contains("dict()"), "{e}");
1828 }
1829
1830 #[test]
1831 fn bad_arguments_to_mutable_methods_are_clear_errors() {
1832 assert!(err("{% set l = list() %}{% set _ = l.append() %}").contains("append()"));
1833 assert!(err("{% set l = list() %}{% set _ = l.pop() %}").contains("pop()"));
1834 assert!(err("{% set l = list([1]) %}{% set _ = l.remove(2) %}").contains("not in the list"));
1835 assert!(err("{% set d = dict() %}{% set _ = d.pop('x') %}").contains("no key"));
1836 assert!(err("{% set d = dict('nope') %}").contains("mapping"));
1837 }
1838
1839 #[test]
1840 fn run_context_and_date_formats() {
1841 let (_d, r) = renderer(Json::Null, &[], &[]);
1842 let out = render(
1843 &r,
1844 "{{ run.report }}/{{ run.set }}/{{ run.target }}/{{ target.name }}/{{ target }} {{ run.date }} {{ run.date.yyyymmdd }} \
1845 {{ run.date.ddmmyyyy }} {{ run.date.yyyy }}-{{ run.date.mm }}-{{ run.date.dd }} {{ run.date_format('%Y-W%V') }}",
1846 )
1847 .unwrap();
1848 assert_eq!(
1849 out,
1850 "monthly/client_a/prod/prod/prod 2026-01-25 20260125 25012026 2026-01-25 2026-W04"
1851 );
1852 }
1853
1854 #[test]
1855 fn var_precedence_is_cli_then_binding_then_default() {
1856 let (_d, r) = renderer(
1857 serde_json::json!({"a": "binding", "b": "binding", "n": 5}),
1858 &[("a", "cli")],
1859 &[],
1860 );
1861 assert_eq!(
1862 render(
1863 &r,
1864 "{{ var('a') }} {{ var('b') }} {{ var('c', 'dflt') }} {{ var('n') + 1 }}"
1865 )
1866 .unwrap(),
1867 "cli binding dflt 6"
1868 );
1869 let e = render(&r, "select\n{{ var('missing') }}").unwrap_err();
1870 assert_eq!(
1871 (e.line, e.message.contains("`var('missing')` has no value")),
1872 (Some(2), true),
1873 "{e}"
1874 );
1875 }
1876
1877 #[test]
1878 fn env_var_reads_the_environment_or_errors() {
1879 let (_d, r) = renderer(Json::Null, &[], &[]);
1880 assert_eq!(
1881 render(&r, "{{ env_var('DRE_TEST_UNSET_VAR', 'x') }}").unwrap(),
1882 "x"
1883 );
1884 let home = std::env::var("PATH").unwrap();
1885 assert_eq!(render(&r, "{{ env_var('PATH') }}").unwrap(), home);
1886 assert!(
1887 render(&r, "{{ env_var('DRE_TEST_UNSET_VAR') }}")
1888 .unwrap_err()
1889 .message
1890 .contains("is not set")
1891 );
1892 }
1893
1894 #[test]
1895 fn macros_from_every_file_are_callable_and_keep_line_numbers() {
1896 let (_d, r) = renderer(
1897 Json::Null,
1898 &[],
1899 &[
1900 ("a.sql", "{% macro double(x) %}{{ x * 2 }}{% endmacro %}\n"),
1901 (
1902 "b.sql",
1903 "{% macro broken() %}\n{{ var('nope') }}\n{% endmacro %}\n{% macro quad(x) %}{{ x * 4 }}{% endmacro %}\n",
1904 ),
1905 ],
1906 );
1907 assert_eq!(
1908 render(&r, "select {{ double(2) }}, {{ quad(1) }}\n").unwrap(),
1909 "select 4, 4\n"
1910 );
1911 let e = render(&r, "line one\nselect {{ broken() }}").unwrap_err();
1912 assert_eq!(e.file, PathBuf::from("macros/b.sql"), "{e}");
1913 assert_eq!(e.line, Some(2), "{e}");
1914 let e = render(&r, "one\ntwo\n{{ nope }}").unwrap_err();
1915 assert_eq!(
1916 (e.file.clone(), e.line),
1917 (PathBuf::from("reports/q.sql"), Some(3)),
1918 "{e}"
1919 );
1920 }
1921
1922 struct Fake;
1923 impl QueryRunner for Fake {
1924 fn run_query(&self, sql: &str, max_rows: u64, _profile: Option<&str>) -> Result<QueryRows, String> {
1925 if max_rows < 2 {
1926 return Err(format!("`{sql}` returned more than {max_rows} rows"));
1927 }
1928 Ok(QueryRows {
1929 columns: vec!["region".into(), "n".into()],
1930 rows: vec![
1931 vec![Value::from("apac"), Value::from(1)],
1932 vec![Value::from("emea"), Value::from(2)],
1933 ],
1934 })
1935 }
1936 fn columns(&self, _: &str, _: Option<&str>) -> Result<Vec<Column>, String> {
1937 Ok(Vec::new())
1938 }
1939 }
1940
1941 #[test]
1942 fn run_query_rows_are_accessible_by_name_and_index() {
1943 let dir = tempfile::tempdir().unwrap();
1944 let r = Renderer::new(RendererConfig {
1945 root: dir.path(),
1946 macros: &[],
1947 context: RunContext {
1948 report: "r".into(),
1949 set: None,
1950 target: "t".into(),
1951 schedule: None,
1952 date: NaiveDate::from_ymd_opt(2026, 1, 1).unwrap(),
1953 now: Utc::now(),
1954 scheduled_at: None,
1955 calendar: Calendar::default(),
1956 locale: Default::default(),
1957 },
1958 vars: JsonMap::new(),
1959 cli_vars: BTreeMap::new(),
1960 runner: Some(Arc::new(Fake)),
1961 connections: None,
1962 mode: Mode::Run,
1963 connection: None,
1964 source_type: "duckdb".into(),
1965 sources: None,
1966 run_query_max_rows: 10_000,
1967 sql: BTreeMap::new(),
1968 lookups: BTreeMap::new(),
1969 lookup_inline_max_rows: 200,
1970 packages: Vec::new(),
1971 project_name: "acme".into(),
1972 dispatch: DispatchOrder::new(),
1973 })
1974 .unwrap();
1975 let src = "{% set res = run_query('select') %}{{ res.columns | join(',') }}|\
1976 {% for row in res.rows %}{{ row.region }}={{ row[1] }}{{ ',' if not loop.last }}{% endfor %}|{{ res | length }}|{{ res[1]['region'] }}";
1977 assert_eq!(render(&r, src).unwrap(), "region,n|apac=1,emea=2|2|emea");
1978 let e = render(&r, "{{ run_query('select', max_rows=1) }}").unwrap_err();
1979 assert!(e.message.contains("more than 1 rows"), "{e}");
1980 assert!(render(&r, "{% if run.set is none %}none{% endif %}").unwrap() == "none");
1981 }
1982
1983 #[test]
1984 fn removed_names_say_what_to_write_instead() {
1985 let e = err("{{ run.profile }}");
1986 assert!(
1987 e.contains("`run.profile` was removed") && e.contains("connection.name"),
1988 "{e}"
1989 );
1990 let e = err("{{ target.schema }}");
1991 assert!(
1992 e.contains("`target.schema` was removed") && e.contains("connection.schema"),
1993 "{e}"
1994 );
1995 }
1996
1997 #[test]
1998 fn a_source_relation_quotes_the_parts_asked_for() {
1999 let r = ResolvedSource {
2000 source: "sales".into(),
2001 table: "orders".into(),
2002 profile: None,
2003 database: None,
2004 schema: "Sales".into(),
2005 identifier: "Or\"ders".into(),
2006 quoting: Quoting {
2007 database: true,
2008 schema: false,
2009 identifier: true,
2010 },
2011 };
2012 assert_eq!(r.relation(Some("\"")), "Sales.\"Or\"\"ders\"");
2013 assert_eq!(r.relation(None), "Sales.Or\"ders");
2014 let r = ResolvedSource {
2015 database: Some("main".into()),
2016 ..r
2017 };
2018 assert_eq!(r.relation(Some("`")), "`main`.Sales.`Or\"ders`");
2019 }
2020
2021 #[test]
2022 fn a_relation_is_mentioned_only_as_a_whole_name() {
2023 assert!(mentions("select * from sales.orders where 1", "sales.orders"));
2024 assert!(!mentions("select * from sales.orders_old", "sales.orders"));
2025 assert!(!mentions("select * from x.sales.orders", "sales.orders"));
2026 }
2027}
2028
2029#[cfg(test)]
2030mod cli_var_tests {
2031 use super::cli_var_value as v;
2032 use serde_json::json;
2033
2034 #[test]
2035 fn yaml_1_2_core_rules() {
2036 assert_eq!(v("false"), json!(false));
2037 assert_eq!(v("TRUE"), json!(true));
2038 assert_eq!(v("~"), json!(null));
2039 assert_eq!(v("5"), json!(5));
2040 assert_eq!(v("-12"), json!(-12));
2041 assert_eq!(v("2.5"), json!(2.5));
2042 assert_eq!(v("[\"NAM\", EMEA]"), json!(["NAM", "EMEA"]));
2043 assert_eq!(v("{k: 1}"), json!({"k": 1}));
2044 for s in [
2045 "yes",
2046 "NO",
2047 "2026-01-31",
2048 "010",
2049 "1.0.0",
2050 "abc",
2051 "[unclosed",
2052 "0x1F",
2053 ] {
2054 assert_eq!(v(s), json!(s), "{s}");
2055 }
2056 assert_eq!(v("\"false\""), json!("false"));
2057 assert_eq!(v("'5'"), json!("5"));
2058 assert_eq!(v(""), json!(""));
2059 }
2060}