renox_core/
import.rs

1//! Imports: the rows of a CSV file, each checked with a form's rules, then
2//! written in one transaction (Filament's import action).
3//!
4//! A row is read like a form: its columns, named by the file's first line,
5//! fill a struct that derives `Deserialize` and `Validate`, so the same
6//! rules (and messages, in the user's language) check a row as check the
7//! form that adds one record. Rows that fail are reported with their row
8//! number; the others are written by your closure, each in a savepoint of
9//! one transaction, so a row the database refuses undoes only its own
10//! writes.
11//!
12//! ```
13//! # use renox::prelude::*;
14//! use renox::import::{Import, ImportReport};
15//! # #[derive(Model, serde::Serialize, Default)]
16//! # #[model(table = "products")]
17//! # struct Product { id: i64, sku: String, name: String, price: i64 }
18//!
19//! /// One line of `sku,name,price`.
20//! #[derive(serde::Deserialize, Validate)]
21//! struct ProductRow {
22//!     #[validate(required, max = 30, alpha_dash)]
23//!     sku: String,
24//!     #[validate(required, max = 100)]
25//!     name: String,
26//!     #[validate(required, min = 0)]
27//!     price: i64,
28//! }
29//!
30//! #[derive(serde::Deserialize, Validate)]
31//! struct ImportForm {
32//!     #[validate(required, mimes(&["csv", "txt"]))]
33//!     file: Option<Upload>,
34//! }
35//!
36//! async fn import(
37//!     State(state): State<AppState>,
38//!     lang: Lang,
39//!     Valid(form): Valid<ImportForm>,
40//! ) -> Result<ImportReport> {
41//!     let file = form.file.ok_or(Error::NotFound)?;
42//!     Import::csv(file.bytes())
43//!         .lang(&lang)
44//!         .run(&state, |tx, row: ProductRow| {
45//!             Box::pin(async move {
46//!                 let product = Product { sku: row.sku, name: row.name, price: row.price, ..Default::default() };
47//!                 Product::create(tx, product).await?;
48//!                 Ok(())
49//!             })
50//!         })
51//!         .await
52//! }
53//! ```
54//!
55//! The handler answers with the [`ImportReport`]: a toast when every row
56//! was imported, else a table of the rows that weren't, which the kit's
57//! `import_action` sheet shows. A file that can't be read at all (not
58//! UTF-8, no rows, too many rows) is a validation error on the form's
59//! `file` field.
60
61use std::collections::HashMap;
62use std::fmt;
63
64use axum::response::{IntoResponse, Response};
65use futures_util::future::BoxFuture;
66use serde::de::DeserializeOwned;
67use serde_json::json;
68
69use crate::auth::User;
70use crate::db::Transaction;
71use crate::validation::extract::{Messages, Parsed, parse_pairs};
72use crate::validation::{Errors, FormContext, Validate, ValidationError, Validator};
73use crate::{AppState, Error, Lang, Result};
74
75/// The most rows one import reads, unless [`Import::max_rows`] says otherwise.
76pub const MAX_IMPORT_ROWS: usize = 10_000;
77
78/// What writes one row (`run`'s closure), boxed.
79type WriteRow<'a, T> =
80    Box<dyn for<'t> FnMut(&'t mut Transaction, T) -> BoxFuture<'t, Result> + Send + 'a>;
81
82/// A data row: its row number and its (column, value) pairs.
83type Record = (usize, Vec<(String, String)>);
84
85/// The most failed rows the report's table lists; the rest are counted.
86const LISTED_FAILURES: usize = 100;
87
88/// A CSV file to import; see the [module docs](self).
89#[must_use = "an import does nothing until it is `run`"]
90pub struct Import {
91    data: Vec<u8>,
92    field: String,
93    delimiter: char,
94    headers: Option<Vec<String>>,
95    renames: Vec<(String, String)>,
96    max_rows: usize,
97    all_or_nothing: bool,
98    lang: Option<Lang>,
99    user: Option<User>,
100}
101
102impl Import {
103    /// The bytes of a CSV file (an upload's, a file's): UTF-8, with or
104    /// without a byte order mark, its first line naming the columns.
105    pub fn csv(data: impl AsRef<[u8]>) -> Self {
106        Self {
107            data: data.as_ref().to_vec(),
108            field: "file".into(),
109            delimiter: ',',
110            headers: None,
111            renames: Vec::new(),
112            max_rows: MAX_IMPORT_ROWS,
113            all_or_nothing: false,
114            lang: None,
115            user: None,
116        }
117    }
118
119    /// Cells are separated by `delimiter` (`;` in files from a spreadsheet
120    /// set to a European language) instead of `,`.
121    pub fn delimiter(mut self, delimiter: char) -> Self {
122        self.delimiter = delimiter;
123        self
124    }
125
126    /// The file has no line of column names: these are its columns, in order.
127    pub fn headers(mut self, names: &[&str]) -> Self {
128        self.headers = Some(names.iter().map(|n| (*n).to_owned()).collect());
129        self
130    }
131
132    /// Reads the column headed `heading` as the field `field`, e.g.
133    /// `rename("Description", "name")`. Without one, a heading names its
134    /// field in lowercase with `_` for spaces and dashes: "Unit price" is
135    /// `unit_price`.
136    pub fn rename(mut self, heading: &str, field: &str) -> Self {
137        self.renames.push((field_name(heading), field.to_owned()));
138        self
139    }
140
141    /// At most `rows` rows (default [`MAX_IMPORT_ROWS`]); a longer file is
142    /// refused before anything is written.
143    pub fn max_rows(mut self, rows: usize) -> Self {
144        self.max_rows = rows;
145        self
146    }
147
148    /// Writes nothing unless every row is valid and saved: one bad row and
149    /// the whole file is refused (the report lists why). By default the
150    /// good rows are written and the bad ones reported.
151    pub fn all_or_nothing(mut self) -> Self {
152        self.all_or_nothing = true;
153        self
154    }
155
156    /// Messages (and the report) in this language; the app's
157    /// `APP_LOCALE` otherwise.
158    pub fn lang(mut self, lang: &Lang) -> Self {
159        self.lang = Some(lang.clone());
160        self
161    }
162
163    /// The user importing, whom the rows' `after` hooks see as
164    /// `form.user` (and the `current_password` rule checks).
165    pub fn user(mut self, user: &User) -> Self {
166        self.user = Some(user.clone());
167        self
168    }
169
170    /// The form field the file came in (`file` by default): a file that
171    /// can't be read is reported as that field's error.
172    pub fn field(mut self, name: &str) -> Self {
173        self.field = name.to_owned();
174        self
175    }
176
177    /// Reads every row into a `T` and checks it with `T`'s rules (its
178    /// `prepare` and `after` hooks too), then calls `write` with each valid
179    /// row inside one transaction, a savepoint per row. An error from
180    /// `write` is that row's failure: a [`ValidationError`] shows its
181    /// messages, `Error::BadRequest` its text, a unique-constraint
182    /// violation says the row repeats a unique value.
183    ///
184    /// `T`'s `authorize` hook isn't asked: authorize the import's route.
185    pub fn run<'a, T, F>(
186        self,
187        state: &'a AppState,
188        write: F,
189    ) -> impl Future<Output = Result<ImportReport>> + Send + 'a
190    where
191        T: DeserializeOwned + Validate + Send + Sync + 'a,
192        F: for<'t> FnMut(&'t mut Transaction, T) -> BoxFuture<'t, Result> + Send + 'a,
193    {
194        let lang = self
195            .lang
196            .clone()
197            .unwrap_or_else(|| state.lang(&state.config.locale));
198        // Boxed: a generic closure held across `.await`s keeps handler
199        // futures from being `Send` (rustc #100013).
200        let mut write: WriteRow<'a, T> = Box::new(write);
201        async move {
202            let records = self.records(&lang)?;
203            let mut report = ImportReport {
204                imported: 0,
205                failed: Vec::new(),
206                all_or_nothing: self.all_or_nothing,
207                lang: lang.clone(),
208            };
209            let messages = Messages {
210                texts: lang.texts(),
211            };
212            let mut valid = Vec::new();
213            for (row, pairs) in records {
214                match check::<T>(state, self.user.as_ref(), &messages, pairs).await? {
215                    Ok(data) => valid.push((row, data)),
216                    Err(errors) => report.failed.push(FailedRow { row, errors }),
217                }
218            }
219            if self.all_or_nothing && !report.failed.is_empty() {
220                return Ok(report);
221            }
222            let mut tx = state.db.begin().await?;
223            let mut imported = 0;
224            for (row, data) in valid {
225                let outcome = tx.savepoint(|tx| write(tx, data)).await;
226                match outcome {
227                    Ok(()) => imported += 1,
228                    Err(err) => {
229                        report.failed.push(FailedRow {
230                            row,
231                            errors: row_errors(err, &lang)?,
232                        });
233                        if self.all_or_nothing {
234                            // Dropping the transaction rolls it back.
235                            return Ok(report);
236                        }
237                    }
238                }
239            }
240            tx.commit().await?;
241            report.imported = imported;
242            report.failed.sort_by_key(|f| f.row);
243            Ok(report)
244        }
245    }
246
247    /// The data rows as (row number, column/value pairs), or the file's
248    /// error on the form's field.
249    fn records(&self, lang: &Lang) -> Result<Vec<Record>> {
250        let refuse = |key: &str, count: usize| -> Error {
251            let mut errors = Errors::new();
252            errors.add(
253                self.field.clone(),
254                lang.t(key, &[("count", &count as &dyn fmt::Display)]),
255            );
256            ValidationError::new(errors).into()
257        };
258        let Ok(text) = std::str::from_utf8(&self.data) else {
259            return Err(refuse("ui.import.not_utf8", 0));
260        };
261        let mut rows = parse_csv(text.trim_start_matches('\u{feff}'), self.delimiter);
262        // A record's index from 0 is its spreadsheet row number minus one.
263        let headers = match &self.headers {
264            Some(headers) => headers.clone(),
265            None if rows.is_empty() => return Err(refuse("ui.import.empty", 0)),
266            None => rows.remove(0).1,
267        };
268        let headers: Vec<String> = headers
269            .iter()
270            .map(|h| {
271                let key = field_name(h);
272                self.renames
273                    .iter()
274                    .find(|(from, _)| *from == key)
275                    .map_or(key, |(_, to)| to.clone())
276            })
277            .collect();
278        let records: Vec<Record> = rows
279            .into_iter()
280            .filter(|(_, cells)| cells.iter().any(|c| !c.trim().is_empty()))
281            .map(|(index, cells)| {
282                let pairs = headers
283                    .iter()
284                    .zip(cells)
285                    .filter(|(h, _)| !h.is_empty())
286                    .map(|(h, c)| (h.clone(), c))
287                    .collect();
288                (index + 1, pairs)
289            })
290            .collect();
291        if records.is_empty() {
292            return Err(refuse("ui.import.empty", 0));
293        }
294        if records.len() > self.max_rows {
295            return Err(refuse("ui.import.too_many", self.max_rows));
296        }
297        Ok(records)
298    }
299}
300
301/// A heading as a field's name: "Unit price " → `unit_price`.
302fn field_name(heading: &str) -> String {
303    heading
304        .trim()
305        .to_lowercase()
306        .split([' ', '-'])
307        .filter(|part| !part.is_empty())
308        .collect::<Vec<_>>()
309        .join("_")
310}
311
312/// A row read and checked as a form is: `Ok` with the data, or its errors.
313async fn check<T: DeserializeOwned + Validate + Send + Sync>(
314    state: &AppState,
315    user: Option<&User>,
316    messages: &Messages,
317    pairs: Vec<(String, String)>,
318) -> Result<std::result::Result<T, Errors>> {
319    let (parsed, _) = parse_pairs::<T>(pairs, &HashMap::new(), messages);
320    let (mut data, mut errors) = match parsed {
321        Parsed::Ok(data, errors) => (data, errors),
322        Parsed::Invalid(errors) => return Ok(Err(errors)),
323    };
324    data.prepare();
325    let rule_errors = Validator::rules_with_texts(&data, messages.texts.clone())
326        .finish_for(state, user)
327        .await?;
328    for (field, list) in rule_errors.iter() {
329        if !errors.has(field) {
330            for message in list {
331                errors.add(field, message.clone());
332            }
333        }
334    }
335    if errors.is_empty() {
336        let form = FormContext {
337            state,
338            user,
339            method: &axum::http::Method::POST,
340            path: "",
341        };
342        data.after(&form, &mut errors).await?;
343    }
344    Ok(if errors.is_empty() {
345        Ok(data)
346    } else {
347        Err(errors)
348    })
349}
350
351/// Why `write` refused a row, as messages for the report.
352fn row_errors(err: Error, lang: &Lang) -> Result<Errors> {
353    let mut errors = Errors::new();
354    match err {
355        Error::Validation(invalid) => return Ok(invalid.errors),
356        Error::BadRequest(message) | Error::Status(_, message) => errors.add("row", message),
357        err if err.is_unique_violation() => errors.add("row", lang.t("ui.import.duplicate", &[])),
358        Error::Internal(err) => {
359            tracing::warn!(?err, "an imported row could not be saved");
360            errors.add("row", lang.t("ui.import.not_saved", &[]));
361        }
362        // Not a row's fault (403, 404…): the import stops.
363        other => return Err(other),
364    }
365    Ok(errors)
366}
367
368/// What an import did: the rows written and the rows that weren't, with
369/// why. As a handler's answer it is a success toast (and `HX-Refresh`)
370/// when every row was imported, else the `renox/import_report.html`
371/// fragment listing the failed rows, which keeps the kit's import sheet
372/// open.
373#[non_exhaustive]
374pub struct ImportReport {
375    /// Rows written.
376    pub imported: usize,
377    /// Rows left out, by row number.
378    pub failed: Vec<FailedRow>,
379    all_or_nothing: bool,
380    lang: Lang,
381}
382
383impl fmt::Debug for ImportReport {
384    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
385        f.debug_struct("ImportReport")
386            .field("imported", &self.imported)
387            .field("failed", &self.failed)
388            .field("all_or_nothing", &self.all_or_nothing)
389            .finish()
390    }
391}
392
393impl ImportReport {
394    /// Every row was imported.
395    pub fn is_clean(&self) -> bool {
396        self.failed.is_empty()
397    }
398
399    /// A one-line summary: "3 rows imported.", "2 rows imported, 1 left
400    /// out.", or "Nothing was imported: 1 row has errors.".
401    pub fn summary(&self) -> String {
402        let failed = self.failed.len() as i64;
403        if self.failed.is_empty() {
404            self.lang
405                .choice("ui.import.done", self.imported as i64, &[])
406        } else if self.imported == 0 {
407            self.lang.choice("ui.import.nothing", failed, &[])
408        } else {
409            self.lang.choice(
410                "ui.import.partial",
411                failed,
412                &[("imported", &self.imported as &dyn fmt::Display)],
413            )
414        }
415    }
416}
417
418impl IntoResponse for ImportReport {
419    fn into_response(self) -> Response {
420        let summary = self.summary();
421        if self.failed.is_empty() {
422            return (
423                crate::Toast::success(summary),
424                crate::HxRefresh,
425                String::new(),
426            )
427                .into_response();
428        }
429        let failed: Vec<_> = self
430            .failed
431            .iter()
432            .take(LISTED_FAILURES)
433            .map(|f| json!({ "row": f.row, "messages": f.messages() }))
434            .collect();
435        crate::view(
436            "renox/import_report.html",
437            minijinja::context! {
438                summary,
439                imported => self.imported,
440                failed,
441                more => self.failed.len().saturating_sub(LISTED_FAILURES),
442            },
443        )
444        .into_response()
445    }
446}
447
448/// A row that wasn't imported.
449#[derive(Debug, Clone)]
450#[non_exhaustive]
451pub struct FailedRow {
452    /// Its number as a spreadsheet shows it: the line of column names is
453    /// row 1, the first data row is row 2.
454    pub row: usize,
455    /// Its errors, by field.
456    pub errors: Errors,
457}
458
459impl FailedRow {
460    /// Every message, in field order.
461    pub fn messages(&self) -> Vec<String> {
462        self.errors
463            .iter()
464            .flat_map(|(_, list)| list.iter().cloned())
465            .collect()
466    }
467}
468
469/// A CSV file with only its line of column names, for people to fill in:
470/// the import sheet's "Download a template" link.
471///
472/// ```
473/// # use renox::prelude::*;
474/// async fn template() -> renox::Download {
475///     renox::import::template("products.csv", &["sku", "name", "price", "stock"])
476/// }
477/// ```
478pub fn template(filename: &str, columns: &[&str]) -> crate::Download {
479    let line = columns
480        .iter()
481        .map(|c| quote_cell(c))
482        .collect::<Vec<_>>()
483        .join(",");
484    crate::Download::bytes(
485        filename.to_owned(),
486        "text/csv; charset=utf-8",
487        format!("\u{feff}{line}\r\n"),
488    )
489}
490
491fn quote_cell(cell: &str) -> String {
492    if cell.contains([',', '"', '\n', '\r', ';']) {
493        format!("\"{}\"", cell.replace('"', "\"\""))
494    } else {
495        cell.to_owned()
496    }
497}
498
499/// RFC 4180 records with their index (from 0): `"…"` around cells holding
500/// the delimiter, quotes (doubled) or line breaks; `\r\n` or `\n` ends a
501/// record.
502fn parse_csv(text: &str, delimiter: char) -> Vec<(usize, Vec<String>)> {
503    let mut records = Vec::new();
504    let mut record: Vec<String> = Vec::new();
505    let mut cell = String::new();
506    let mut quoted = false;
507    let mut chars = text.chars().peekable();
508    let mut started = false;
509    while let Some(c) = chars.next() {
510        started = true;
511        if quoted {
512            match c {
513                '"' if chars.peek() == Some(&'"') => {
514                    chars.next();
515                    cell.push('"');
516                }
517                '"' => quoted = false,
518                c => cell.push(c),
519            }
520            continue;
521        }
522        match c {
523            '"' if cell.is_empty() => quoted = true,
524            c if c == delimiter => record.push(std::mem::take(&mut cell)),
525            '\r' if chars.peek() == Some(&'\n') => {}
526            '\n' | '\r' => {
527                record.push(std::mem::take(&mut cell));
528                records.push((records.len(), std::mem::take(&mut record)));
529                started = false;
530            }
531            c => cell.push(c),
532        }
533    }
534    if started {
535        record.push(cell);
536        records.push((records.len(), record));
537    }
538    records
539}
540
541#[cfg(test)]
542mod tests {
543    use super::*;
544
545    fn cells(text: &str) -> Vec<Vec<String>> {
546        parse_csv(text, ',').into_iter().map(|(_, r)| r).collect()
547    }
548
549    #[test]
550    fn reads_rfc_4180() {
551        assert_eq!(
552            cells("a,b\r\n\"x, y\",\"say \"\"hi\"\"\"\n\"two\nlines\",\n"),
553            vec![
554                vec!["a".to_owned(), "b".into()],
555                vec!["x, y".into(), "say \"hi\"".into()],
556                vec!["two\nlines".into(), String::new()],
557            ]
558        );
559        assert_eq!(cells("a,b"), vec![vec!["a".to_owned(), "b".into()]]);
560        assert_eq!(parse_csv("a;b\n", ';')[0].1, vec!["a", "b"]);
561        assert!(cells("").is_empty());
562    }
563
564    #[test]
565    fn headings_name_fields() {
566        assert_eq!(field_name(" Unit price "), "unit_price");
567        assert_eq!(field_name("e-mail"), "e_mail");
568        assert_eq!(field_name("SKU"), "sku");
569    }
570
571    #[test]
572    fn templates_quote_their_cells() {
573        assert_eq!(quote_cell("name"), "name");
574        assert_eq!(quote_cell("a,b"), "\"a,b\"");
575    }
576
577    #[derive(serde::Deserialize)]
578    struct Row {
579        name: String,
580        qty: i64,
581        #[allow(dead_code)]
582        due: Option<chrono::NaiveDate>,
583    }
584
585    impl Validate for Row {
586        fn rules(&self, v: &mut Validator) {
587            v.field("name", &self.name).required();
588            v.field("qty", &self.qty).min(1);
589        }
590    }
591
592    async fn app() -> crate::testing::TestApp {
593        crate::testing::TestApp::new(crate::App::new()).await
594    }
595
596    /// Rows that don't read: a date that no placeholder makes readable
597    /// fails the row; a number that doesn't read is reported once, not
598    /// again by its `min` rule.
599    #[tokio::test]
600    async fn rows_that_dont_read_are_reported_once() {
601        let app = app().await;
602        let csv = "name,qty,due\nTea,2,someday\nCoffee,lots,2026-01-01\nCake,3,\n";
603        let user = User::default();
604        let report = Import::csv(csv)
605            .user(&user)
606            .run(app.state(), |_tx, _row: Row| Box::pin(async { Ok(()) }))
607            .await
608            .unwrap();
609        assert_eq!(report.imported, 1);
610        let rows: Vec<_> = report.failed.iter().map(|f| f.row).collect();
611        assert_eq!(rows, [2, 3]);
612        assert!(report.failed[0].errors.has("due"));
613        let qty: Vec<_> = report.failed[1]
614            .errors
615            .iter()
616            .filter(|(field, _)| *field == "qty")
617            .flat_map(|(_, messages)| messages.iter())
618            .collect();
619        assert_eq!(qty, ["The qty must be a number."]);
620        let debug = format!("{report:?}");
621        assert!(
622            debug.starts_with("ImportReport { imported: 1, failed: [")
623                && debug.contains("all_or_nothing: false"),
624            "{debug}"
625        );
626    }
627
628    /// Why `write` refused a row: its validation messages, its text, a
629    /// repeat, "not saved" for an internal error; any other error (a 403)
630    /// stops the import.
631    #[tokio::test]
632    async fn writes_that_fail_are_reported_by_kind() {
633        let app = app().await;
634        let csv = "name,qty\ninvalid,1\nbad,1\nboom,1\nfine,1\n";
635        let report = Import::csv(csv)
636            .run(app.state(), |_tx, row: Row| {
637                Box::pin(async move {
638                    match row.name.as_str() {
639                        "invalid" => {
640                            let mut errors = Errors::new();
641                            errors.add("name", "The name is taken.");
642                            Err(ValidationError::new(errors).into())
643                        }
644                        "bad" => Err(Error::BadRequest("Unknown supplier.".into())),
645                        "boom" => Err(Error::Internal(anyhow::anyhow!("disk full"))),
646                        _ => Ok(()),
647                    }
648                })
649            })
650            .await
651            .unwrap();
652        assert_eq!(report.imported, 1);
653        let first = |i: usize, field: &str| report.failed[i].errors.first(field).map(str::to_owned);
654        assert_eq!(first(0, "name").as_deref(), Some("The name is taken."));
655        assert_eq!(first(1, "row").as_deref(), Some("Unknown supplier."));
656        assert!(first(2, "row").is_some_and(|m| !m.contains("disk full")));
657
658        let stopped = Import::csv(csv)
659            .run(app.state(), |_tx, _row: Row| {
660                Box::pin(async { Err(Error::Forbidden) })
661            })
662            .await;
663        assert!(matches!(stopped, Err(Error::Forbidden)));
664    }
665
666    /// A file that can't be read is an error on the form's field, named
667    /// with `field`.
668    #[tokio::test]
669    async fn an_unreadable_file_is_an_error_on_its_field() {
670        let app = app().await;
671        let err = Import::csv([0xFF, 0xFE, 0x00])
672            .field("upload")
673            .run(app.state(), |_tx, _row: Row| Box::pin(async { Ok(()) }))
674            .await
675            .unwrap_err();
676        let Error::Validation(invalid) = err else {
677            panic!("a validation error");
678        };
679        assert!(invalid.errors.has("upload"));
680    }
681}
renox_core/import.rs

renox_core/
import.rs
