mirror of
https://github.com/simonw/sqlite-utils.git
synced 2026-11-04 17:28:34 +01:00
Merge bb93d57055 into 9c66bafa52
This commit is contained in:
commit
f0488d019c
3 changed files with 45 additions and 1 deletions
|
|
@ -285,6 +285,20 @@ def _extra_key_strategy(
|
||||||
yield row_out
|
yield row_out
|
||||||
|
|
||||||
|
|
||||||
|
# Delimiters rows_from_file() looks for when it detects the format. Without a list
|
||||||
|
# csv.Sniffer takes any character that repeats on the sampled lines, so a single-column
|
||||||
|
# file was split on a letter, or raised "Could not determine delimiter".
|
||||||
|
_SNIFF_DELIMITERS = ",;\t|"
|
||||||
|
|
||||||
|
|
||||||
|
def _sniff_csv_dialect(sample: str) -> type[csv.Dialect]:
|
||||||
|
try:
|
||||||
|
return csv.Sniffer().sniff(sample, delimiters=_SNIFF_DELIMITERS)
|
||||||
|
except csv.Error:
|
||||||
|
# None of them: a single-column file, read as plain CSV
|
||||||
|
return csv.excel
|
||||||
|
|
||||||
|
|
||||||
def rows_from_file(
|
def rows_from_file(
|
||||||
fp: BinaryIO,
|
fp: BinaryIO,
|
||||||
format: Format | None = None,
|
format: Format | None = None,
|
||||||
|
|
@ -389,7 +403,7 @@ def rows_from_file(
|
||||||
with buffered:
|
with buffered:
|
||||||
return rows_from_file(buffered, format=Format.JSON)
|
return rows_from_file(buffered, format=Format.JSON)
|
||||||
else:
|
else:
|
||||||
dialect = csv.Sniffer().sniff(
|
dialect = _sniff_csv_dialect(
|
||||||
first_bytes.decode(encoding or "utf-8-sig", "ignore")
|
first_bytes.decode(encoding or "utf-8-sig", "ignore")
|
||||||
)
|
)
|
||||||
rows, _ = rows_from_file(
|
rows, _ = rows_from_file(
|
||||||
|
|
|
||||||
|
|
@ -157,6 +157,17 @@ def test_memory_csv_encoding(tmpdir, use_stdin):
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def test_memory_csv_single_column(tmpdir):
|
||||||
|
csv_path = str(tmpdir / "names.csv")
|
||||||
|
with open(csv_path, "w") as fp:
|
||||||
|
fp.write("name\nAnn\nBob\n")
|
||||||
|
result = CliRunner().invoke(
|
||||||
|
cli.cli, ["memory", csv_path, "select * from names", "--nl"]
|
||||||
|
)
|
||||||
|
assert result.exit_code == 0, result.output
|
||||||
|
assert result.output.strip() == '{"name": "Ann"}\n{"name": "Bob"}'
|
||||||
|
|
||||||
|
|
||||||
def test_memory_csv_headers_only(tmpdir):
|
def test_memory_csv_headers_only(tmpdir):
|
||||||
csv_path = str(tmpdir / "headers_only.csv")
|
csv_path = str(tmpdir / "headers_only.csv")
|
||||||
with open(csv_path, "w") as fp:
|
with open(csv_path, "w") as fp:
|
||||||
|
|
|
||||||
|
|
@ -11,6 +11,7 @@ from sqlite_utils.utils import Format, RowError, rows_from_file
|
||||||
"input,expected_format",
|
"input,expected_format",
|
||||||
(
|
(
|
||||||
(b"id,name\n1,Cleo", Format.CSV),
|
(b"id,name\n1,Cleo", Format.CSV),
|
||||||
|
(b"id;name\n1;Cleo", Format.CSV),
|
||||||
(b"id\tname\n1\tCleo", Format.TSV),
|
(b"id\tname\n1\tCleo", Format.TSV),
|
||||||
(b'[{"id": "1", "name": "Cleo"}]', Format.JSON),
|
(b'[{"id": "1", "name": "Cleo"}]', Format.JSON),
|
||||||
),
|
),
|
||||||
|
|
@ -22,6 +23,24 @@ def test_rows_from_file_detect_format(input, expected_format):
|
||||||
assert rows_list == [{"id": "1", "name": "Cleo"}]
|
assert rows_list == [{"id": "1", "name": "Cleo"}]
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize(
|
||||||
|
"input,expected",
|
||||||
|
(
|
||||||
|
# csv.Sniffer could not determine a delimiter for this one
|
||||||
|
(b"name\nAnn\nBob\n", [{"name": "Ann"}, {"name": "Bob"}]),
|
||||||
|
# and took "l", which every line holds twice, for this one
|
||||||
|
(
|
||||||
|
b"url\nhttps://example.com/a\nhttps://example.org/b\n",
|
||||||
|
[{"url": "https://example.com/a"}, {"url": "https://example.org/b"}],
|
||||||
|
),
|
||||||
|
),
|
||||||
|
)
|
||||||
|
def test_rows_from_file_detect_format_single_column(input, expected):
|
||||||
|
rows, format = rows_from_file(BytesIO(input))
|
||||||
|
assert format == Format.CSV
|
||||||
|
assert list(rows) == expected
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.parametrize("input", (b"", b" \n\t"))
|
@pytest.mark.parametrize("input", (b"", b" \n\t"))
|
||||||
def test_rows_from_file_empty_input(input):
|
def test_rows_from_file_empty_input(input):
|
||||||
rows, format = rows_from_file(BytesIO(input))
|
rows, format = rows_from_file(BytesIO(input))
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue