Prevent csv QUOTE_NONE writer panics without quotechar

doma17 · doma17 · commit f29b7a31f48d · 2026-06-30T20:24:02.000+09:00
Handle csv.writer rows with QUOTE_NONE through a RustPython-owned unquoted writer path so quotechar=None no longer reaches the unfinished csv_core configuration branch. Escape delimiter, newline, quotechar, and escapechar bytes according to the active dialect, and preserve CPython's single-empty-field error behavior.

Constraint: Match CPython csv.writer behavior without changing Lib/ copied stdlib files.
Rejected: Relying on csv_core QuoteStyle::Never | it does not model CPython QUOTE_NONE escaping with quotechar=None.
Confidence: high
Scope-risk: narrow
Directive: Keep QUOTE_NONE writer behavior separate unless csv_core gains matching CPython semantics.
Tested: prek run --all-files; cargo test --workspace --exclude rustpython_wasm --exclude rustpython-venvlauncher; cargo build --release --features sqlite; pytest -v in extra_tests
Assisted-by: Codex:gpt-5.5
diff --git a/crates/stdlib/src/csv.rs b/crates/stdlib/src/csv.rs
@@ -910,12 +910,8 @@ mod _csv {
                 writer = writer.delimiter(t);
             }
 
-            if let Some(t) = self.quotechar {
-                if let Some(u) = t {
-                    writer = writer.quote(u);
-                } else {
-                    todo!()
-                }
+            if let Some(Some(t)) = self.quotechar {
+                writer = writer.quote(t);
             }
 
             if let Some(t) = self.doublequote {
@@ -1148,6 +1144,24 @@ mod _csv {
         Ok(())
     }
 
+    fn write_unquoted_field(
+        output: &mut Vec<u8>,
+        data: &[u8],
+        dialect: PyDialect,
+        vm: &VirtualMachine,
+    ) -> PyResult<()> {
+        for &byte in data {
+            if field_needs_escape(byte, dialect) {
+                let escapechar = dialect
+                    .escapechar
+                    .ok_or_else(|| new_csv_error(vm, "need to escape, but no escapechar set"))?;
+                output.push(escapechar);
+            }
+            output.push(byte);
+        }
+        Ok(())
+    }
+
     fn field_needs_quotes(data: &[u8], dialect: PyDialect) -> bool {
         data.iter().any(|&byte| {
             byte == dialect.delimiter
@@ -1157,6 +1171,14 @@ mod _csv {
         })
     }
 
+    fn field_needs_escape(byte: u8, dialect: PyDialect) -> bool {
+        byte == dialect.delimiter
+            || dialect.quotechar == Some(byte)
+            || dialect.escapechar == Some(byte)
+            || matches!(byte, b'\r' | b'\n')
+            || matches!(dialect.lineterminator, Terminator::Any(t) if byte == t)
+    }
+
     fn write_lineterminator(output: &mut Vec<u8>, terminator: Terminator) {
         match terminator {
             Terminator::CRLF => output.extend_from_slice(b"\r\n"),
@@ -1222,8 +1244,59 @@ mod _csv {
             self.write.call((s,), vm)
         }
 
+        fn writerow_quote_none(&self, row: PyObjectRef, vm: &VirtualMachine) -> PyResult {
+            let _state = self.state.lock();
+
+            let row: ArgIterable = ArgIterable::try_from_object(vm, row.clone()).map_err(|_e| {
+                new_csv_error(
+                    vm,
+                    format!("'{}' object is not iterable", row.class().name()),
+                )
+            })?;
+
+            let fields = row.iter(vm)?.collect::<PyResult<Vec<_>>>()?;
+            let single_field = fields.len() == 1;
+            let mut output = Vec::new();
+
+            for (index, field) in fields.into_iter().enumerate() {
+                if index > 0 {
+                    output.push(self.dialect.delimiter);
+                }
+
+                let stringified;
+                let data: &[u8] = match_class!(match field {
+                    ref s @ PyStr => s.as_bytes(),
+                    crate::builtins::PyNone => b"",
+                    ref obj => {
+                        stringified = obj.str(vm)?;
+                        stringified.as_bytes()
+                    }
+                });
+
+                if single_field && data.is_empty() {
+                    return Err(new_csv_error(
+                        vm,
+                        "single empty field record must be quoted",
+                    ));
+                }
+
+                write_unquoted_field(&mut output, data, self.dialect, vm)?;
+            }
+
+            write_lineterminator(&mut output, self.dialect.lineterminator);
+
+            let s = core::str::from_utf8(&output)
+                .map_err(|_| vm.new_unicode_decode_error("csv not utf8"))?;
+
+            self.write.call((s,), vm)
+        }
+
         #[pymethod]
         fn writerow(&self, row: PyObjectRef, vm: &VirtualMachine) -> PyResult {
+            if matches!(self.dialect.quoting, QuoteStyle::None) {
+                return self.writerow_quote_none(row, vm);
+            }
+
             if matches!(
                 self.dialect.quoting,
                 QuoteStyle::Strings | QuoteStyle::Notnull
diff --git a/extra_tests/snippets/stdlib_csv.py b/extra_tests/snippets/stdlib_csv.py
@@ -72,3 +72,56 @@ def test_quote_strings_and_notnull_writer():
 
 
 test_quote_strings_and_notnull_writer()
+
+
+def test_quote_none_writer_without_quotechar():
+    no_quotechar_buf = io.StringIO()
+    csv.writer(
+        no_quotechar_buf,
+        quoting=csv.QUOTE_NONE,
+        quotechar=None,
+        escapechar="\\",
+    ).writerow(["a,b", 'x"y'])
+    assert no_quotechar_buf.getvalue() == 'a\\,b,x"y\r\n'
+
+    default_quotechar_buf = io.StringIO()
+    csv.writer(
+        default_quotechar_buf,
+        quoting=csv.QUOTE_NONE,
+        escapechar="\\",
+    ).writerow(["a,b", 'x"y'])
+    assert default_quotechar_buf.getvalue() == 'a\\,b,x\\"y\r\n'
+
+    escapechar_buf = io.StringIO()
+    csv.writer(
+        escapechar_buf,
+        quoting=csv.QUOTE_NONE,
+        quotechar=None,
+        escapechar="\\",
+    ).writerow(["a\\b"])
+    assert escapechar_buf.getvalue() == "a\\\\b\r\n"
+
+    with assert_raises(csv.Error):
+        csv.writer(io.StringIO(), quoting=csv.QUOTE_NONE, quotechar=None).writerow(
+            ["a,b"]
+        )
+
+    with assert_raises(csv.Error):
+        csv.writer(
+            io.StringIO(),
+            quoting=csv.QUOTE_NONE,
+            quotechar=None,
+            escapechar="\\",
+        ).writerow([None])
+
+    two_empty_buf = io.StringIO()
+    csv.writer(
+        two_empty_buf,
+        quoting=csv.QUOTE_NONE,
+        quotechar=None,
+        escapechar="\\",
+    ).writerow([None, ""])
+    assert two_empty_buf.getvalue() == ",\r\n"
+
+
+test_quote_none_writer_without_quotechar()
