lean_ctx/core/patterns/
alembic.rs

1//! Alembic (SQLAlchemy migrations) output compression.
2//!
3//! `alembic upgrade`/`downgrade` log each step as
4//! `INFO  [alembic.runtime.migration] Running upgrade <from> -> <to>, <msg>`
5//! plus boilerplate (`Context impl ...`, `Will assume transactional DDL`).
6//! We keep one compact line per applied revision and any error, dropping the
7//! logger prefix and boilerplate.
8
9use crate::core::compressor::strip_ansi;
10
11pub fn compress(_cmd: &str, output: &str) -> Option<String> {
12    let trimmed = output.trim();
13    if trimmed.is_empty() {
14        return Some("alembic: ok".to_string());
15    }
16
17    let mut upgrades: Vec<String> = Vec::new();
18    let mut downgrades: Vec<String> = Vec::new();
19    let mut errors: Vec<String> = Vec::new();
20
21    for raw in trimmed.lines() {
22        let stripped = strip_ansi(raw);
23        let line = stripped.trim();
24        if line.is_empty() {
25            continue;
26        }
27        let body = strip_log_prefix(line);
28
29        if let Some(rest) = body.strip_prefix("Running upgrade ") {
30            upgrades.push(parse_target(rest));
31        } else if let Some(rest) = body.strip_prefix("Running downgrade ") {
32            downgrades.push(parse_target(rest));
33        } else if is_error(line) {
34            errors.push(body.to_string());
35        }
36    }
37
38    if upgrades.is_empty() && downgrades.is_empty() && errors.is_empty() {
39        return Some(fallback(trimmed));
40    }
41
42    let mut parts: Vec<String> = Vec::new();
43    let mut header = String::from("alembic:");
44    if !upgrades.is_empty() {
45        header.push_str(&format!(" {} upgrade(s)", upgrades.len()));
46    }
47    if !downgrades.is_empty() {
48        header.push_str(&format!(" {} downgrade(s)", downgrades.len()));
49    }
50    if upgrades.is_empty() && downgrades.is_empty() {
51        header.push_str(" FAILED");
52    }
53    parts.push(header);
54    for u in &upgrades {
55        parts.push(format!("  ↑ {u}"));
56    }
57    for d in &downgrades {
58        parts.push(format!("  ↓ {d}"));
59    }
60    for e in errors.iter().take(5) {
61        parts.push(format!("  {e}"));
62    }
63    Some(parts.join("\n"))
64}
65
66/// Drop a leading `LEVEL  [alembic...] ` python-logging prefix.
67fn strip_log_prefix(line: &str) -> &str {
68    let is_log = line.starts_with("INFO")
69        || line.starts_with("WARNING")
70        || line.starts_with("ERROR")
71        || line.starts_with("DEBUG");
72    if is_log
73        && line.contains("[alembic")
74        && let Some(idx) = line.find("] ")
75    {
76        return line[idx + 2..].trim_start();
77    }
78    line
79}
80
81/// Parse `<from> -> <to>, <msg>` into `<to> <msg>` (msg optional).
82fn parse_target(rest: &str) -> String {
83    let after = rest.split("-> ").nth(1).unwrap_or(rest).trim();
84    match after.split_once(", ") {
85        Some((rev, msg)) => format!("{} {}", rev.trim(), msg.trim()),
86        None => after.to_string(),
87    }
88}
89
90fn is_error(line: &str) -> bool {
91    line.starts_with("ERROR")
92        || line.starts_with("FAILED")
93        || line.contains("Error:")
94        || line.contains("alembic.util.exc")
95        || line.contains("Can't locate revision")
96        || line.contains("Target database is not up to date")
97}
98
99fn fallback(text: &str) -> String {
100    let lines: Vec<&str> = text.lines().filter(|l| !l.trim().is_empty()).collect();
101    let n = lines.len().min(8);
102    let mut s = lines[..n].join("\n");
103    if lines.len() > n {
104        s.push_str(&format!("\n... (+{} lines)", lines.len() - n));
105    }
106    s
107}
108
109#[cfg(test)]
110mod tests {
111    use super::*;
112
113    const UPGRADE: &str = "INFO  [alembic.runtime.migration] Context impl PostgresqlImpl.\nINFO  [alembic.runtime.migration] Will assume transactional DDL.\nINFO  [alembic.runtime.migration] Running upgrade  -> a1b2c3, create users table\nINFO  [alembic.runtime.migration] Running upgrade a1b2c3 -> d4e5f6, add email index\n";
114
115    #[test]
116    fn keeps_revisions_drops_boilerplate() {
117        let r = compress("alembic upgrade head", UPGRADE).unwrap();
118        assert!(r.contains("2 upgrade(s)"), "counts upgrades: {r}");
119        assert!(r.contains("a1b2c3 create users table"), "{r}");
120        assert!(r.contains("d4e5f6 add email index"), "{r}");
121        assert!(!r.contains("transactional DDL"), "drops boilerplate: {r}");
122        assert!(!r.contains("Context impl"), "drops boilerplate: {r}");
123    }
124
125    #[test]
126    fn shorter_than_input() {
127        let r = compress("alembic upgrade head", UPGRADE).unwrap();
128        assert!(r.len() < UPGRADE.len());
129    }
130
131    #[test]
132    fn empty_is_ok() {
133        assert_eq!(compress("alembic upgrade head", "").unwrap(), "alembic: ok");
134    }
135
136    #[test]
137    fn surfaces_errors() {
138        let out = "INFO  [alembic.runtime.migration] Context impl PostgresqlImpl.\nFAILED: Can't locate revision identified by 'deadbeef'\n";
139        let r = compress("alembic upgrade head", out).unwrap();
140        assert!(r.contains("FAILED"), "{r}");
141        assert!(r.contains("deadbeef"), "{r}");
142    }
143}
lean_ctx/core/patterns/alembic.rs

lean_ctx/core/patterns/
alembic.rs
