Preserve carriage returns in quoted CSV/TSV fields (#895)

sqlite_utils/cli.py

@@ -1226,11 +1226,14 @@ def _insert_docs(docs, tracker=None):
     # The --sniff option needs us to buffer the file to peek ahead
     sniff_buffer = None
     decoded_buffer = None
+    newline = "" if (csv or tsv) else None
     if sniff:
         sniff_buffer = io.BufferedReader(file, buffer_size=4096)
-        decoded_buffer = io.TextIOWrapper(sniff_buffer, encoding=encoding)
+        decoded_buffer = io.TextIOWrapper(
+            sniff_buffer, encoding=encoding, newline=newline
+        )
     else:
-        decoded_buffer = io.TextIOWrapper(file, encoding=encoding)
+        decoded_buffer = io.TextIOWrapper(file, encoding=encoding, newline=newline)
 
     tracker = None
     with file_progress(decoded_buffer, silent=silent) as decoded:

sqlite_utils/utils.py

@@ -346,7 +346,7 @@ class Format(enum.Enum):
         return (json.loads(line) for line in fp if line.strip()), Format.NL
     elif format == Format.CSV:
         use_encoding: str = encoding or "utf-8-sig"
-        decoded_fp = io.TextIOWrapper(fp, encoding=use_encoding)
+        decoded_fp = io.TextIOWrapper(fp, encoding=use_encoding, newline="")
         if dialect is not None:
             reader = csv.DictReader(decoded_fp, dialect=dialect)
         else:

tests/test_cli_insert.py

@@ -235,6 +235,21 @@ def test_insert_csv_tsv(content, options, db_path, tmpdir):
     assert [{"foo": "1", "bar": "2", "baz": "cat,dog"}] == list(db.table("data").rows)
 
 
+@pytest.mark.parametrize("fmt_opt", (["--csv"], ["--tsv"]))
+def test_insert_csv_tsv_preserves_carriage_returns_in_quoted_fields(fmt_opt, db_path):
+    delimiter = "\t" if "--tsv" in fmt_opt else ","
+    content = f'id{delimiter}v\r\n1{delimiter}"line1\r\nline2"\r\n'
+    result = CliRunner().invoke(
+        cli.cli,
+        ["insert", db_path, "data", "-"] + fmt_opt + ["--no-detect-types"],
+        catch_exceptions=False,
+        input=content.encode("utf-8"),
+    )
+    assert result.exit_code == 0
+    db = Database(db_path)
+    assert list(db.table("data").rows) == [{"id": "1", "v": "line1\r\nline2"}]
+
+
 @pytest.mark.parametrize("empty_null", (True, False))
 def test_insert_csv_empty_null(db_path, empty_null):
     options = ["--csv", "--no-detect-types"]

tests/test_rows_from_file.py

@@ -131,3 +131,31 @@ def test_detect_format_keeps_streaming_reader_open(
     assert not buffered_readers[0].closed
     assert len(list(rows)) == 2000
     assert buffered_readers[0].closed
+
+
+@pytest.mark.parametrize(
+    "file_bytes,expected_format,expected_rows",
+    [
+        (
+            b'id,v\r\n1,"line1\r\nline2"\r\n',
+            Format.CSV,
+            [{"id": "1", "v": "line1\r\nline2"}],
+        ),
+        (
+            b'id\tv\r\n1\t"tab1\r\ntab2"\r\n',
+            Format.TSV,
+            [{"id": "1", "v": "tab1\r\ntab2"}],
+        ),
+        (
+            b'id,v\r\n1,"isolated\rreturn"\r\n',
+            Format.CSV,
+            [{"id": "1", "v": "isolated\rreturn"}],
+        ),
+    ],
+)
+def test_rows_from_file_preserves_carriage_returns_in_quoted_fields(
+    file_bytes, expected_format, expected_rows
+):
+    rows, detected_format = rows_from_file(BytesIO(file_bytes), format=expected_format)
+    assert detected_format == expected_format
+    assert list(rows) == expected_rows