Skip to content

API reference

Package

csvdir

Read every CSV in a directory as one iterator.

CsvChunksDir

Iterate CSV rows in fixed-size list[dict[str, str]] chunks.

chunksize must be a positive integer.

Header validation matches {class}~csvdir.dir_reader.CsvDir: use strict_headers, expected_headers, and on_mismatch ('error' or 'skip').

with_paths 

with_paths()

Return an iterator yielding (path, chunk) pairs.

enumerate 

enumerate()

Return an iterator yielding (name, chunk) pairs, where name is the stem from the path.

CsvDirFile dataclass

File-like object that concatenates CSV files in a directory into one logical CSV stream. Emit one header line, then bodies in sorted file path order (same discovery order as CsvDir).

Stitching compares headers in sequence — the emitted header fixes column order. CsvDir / read_dir treat header names as sets, so columns may be permuted across files.

Canonical header sequence:

  • If expected_headers is set, that list is canonical.
  • Else if strict_headers is true, use the first discovered file after sorting (same rule as CsvDir).
  • Else choose the lexicographically smallest delimiter.join(headers) among scanned files.

Matching files contribute their body lines after the sole header emit; mismatches skip or raise according to on_mismatch.

pandas compatibility:

  • read(size=-1), readline(), readlines(); size uses Python string/code-unit counts.
  • seek(0) to restart (other seeks raise).
  • tell(), readable(), seekable(), close(), context manager.

Data is lazy; directory contents are not fully buffered in memory.

CsvDir dataclass

Iterate rows from all CSV files in a directory as dict[str, str].

Header checks use column names as sets (order-independent); see CsvDirFile for order-sensitive stitched streams.

With strict_headers, the pinned schema follows the first sorted file inside that iteration only (expected_headers is not mutated).

with_names 

with_names()

Alias for enumerate(); tests expect names without extension here.

read_dir_chunks 

read_dir_chunks(path, *, extension='csv', delimiter=',', chunksize=1000, encoding='utf-8', newline='', quotechar='"', escapechar=None, strict_headers=False, expected_headers=None, on_mismatch='error', recurse=False, case_insensitive=True, include_hidden=False)

Return a {class}~csvdir.chunks_dir.CsvChunksDir for path with fixed chunksize.

Raises:

Type Description
ValueError

if chunksize is less than 1.

read_dir 

read_dir(path: str | None = None, *, extension: str = 'csv', delimiter: str = ',', chunksize: None = None, encoding: str = 'utf-8', newline: str = '', quotechar: str = '"', escapechar: str | None = None, strict_headers: bool = False, expected_headers: list[str] | None = None, on_mismatch: Literal['error', 'skip'] = 'error', recurse: bool = False, case_insensitive: bool = True, include_hidden: bool = False) -> CsvDir

read_dir(path: str | None = None, *, extension: str = 'csv', delimiter: str = ',', chunksize: int, encoding: str = 'utf-8', newline: str = '', quotechar: str = '"', escapechar: str | None = None, strict_headers: bool = False, expected_headers: list[str] | None = None, on_mismatch: Literal['error', 'skip'] = 'error', recurse: bool = False, case_insensitive: bool = True, include_hidden: bool = False) -> CsvChunksDir

read_dir(path=None, *, extension='csv', delimiter=',', chunksize=None, encoding='utf-8', newline='', quotechar='"', escapechar=None, strict_headers=False, expected_headers=None, on_mismatch='error', recurse=False, case_insensitive=True, include_hidden=False)

Discover and iterate CSV files under path.

Returns {class}~csvdir.chunks_dir.CsvChunksDir when chunksize is set, otherwise {class}~csvdir.dir_reader.CsvDir.

Raises:

Type Description
ValueError

if chunksize is not None and is less than 1.

Factories

csvdir.factory.read_dir 

read_dir(path: str | None = None, *, extension: str = 'csv', delimiter: str = ',', chunksize: None = None, encoding: str = 'utf-8', newline: str = '', quotechar: str = '"', escapechar: str | None = None, strict_headers: bool = False, expected_headers: list[str] | None = None, on_mismatch: Literal['error', 'skip'] = 'error', recurse: bool = False, case_insensitive: bool = True, include_hidden: bool = False) -> CsvDir

read_dir(path: str | None = None, *, extension: str = 'csv', delimiter: str = ',', chunksize: int, encoding: str = 'utf-8', newline: str = '', quotechar: str = '"', escapechar: str | None = None, strict_headers: bool = False, expected_headers: list[str] | None = None, on_mismatch: Literal['error', 'skip'] = 'error', recurse: bool = False, case_insensitive: bool = True, include_hidden: bool = False) -> CsvChunksDir

read_dir(path=None, *, extension='csv', delimiter=',', chunksize=None, encoding='utf-8', newline='', quotechar='"', escapechar=None, strict_headers=False, expected_headers=None, on_mismatch='error', recurse=False, case_insensitive=True, include_hidden=False)

Discover and iterate CSV files under path.

Returns {class}~csvdir.chunks_dir.CsvChunksDir when chunksize is set, otherwise {class}~csvdir.dir_reader.CsvDir.

Raises:

Type Description
ValueError

if chunksize is not None and is less than 1.

csvdir.chunks_factory.read_dir_chunks 

read_dir_chunks(path, *, extension='csv', delimiter=',', chunksize=1000, encoding='utf-8', newline='', quotechar='"', escapechar=None, strict_headers=False, expected_headers=None, on_mismatch='error', recurse=False, case_insensitive=True, include_hidden=False)

Return a {class}~csvdir.chunks_dir.CsvChunksDir for path with fixed chunksize.

Raises:

Type Description
ValueError

if chunksize is less than 1.

Readers

csvdir.dir_reader.CsvDir dataclass

Iterate rows from all CSV files in a directory as dict[str, str].

Header checks use column names as sets (order-independent); see CsvDirFile for order-sensitive stitched streams.

With strict_headers, the pinned schema follows the first sorted file inside that iteration only (expected_headers is not mutated).

with_names 

with_names()

Alias for enumerate(); tests expect names without extension here.

csvdir.chunks_dir.CsvChunksDir

Iterate CSV rows in fixed-size list[dict[str, str]] chunks.

chunksize must be a positive integer.

Header validation matches {class}~csvdir.dir_reader.CsvDir: use strict_headers, expected_headers, and on_mismatch ('error' or 'skip').

with_paths 

with_paths()

Return an iterator yielding (path, chunk) pairs.

enumerate 

enumerate()

Return an iterator yielding (name, chunk) pairs, where name is the stem from the path.

csvdir.concat_file.CsvDirFile dataclass

File-like object that concatenates CSV files in a directory into one logical CSV stream. Emit one header line, then bodies in sorted file path order (same discovery order as CsvDir).

Stitching compares headers in sequence — the emitted header fixes column order. CsvDir / read_dir treat header names as sets, so columns may be permuted across files.

Canonical header sequence:

  • If expected_headers is set, that list is canonical.
  • Else if strict_headers is true, use the first discovered file after sorting (same rule as CsvDir).
  • Else choose the lexicographically smallest delimiter.join(headers) among scanned files.

Matching files contribute their body lines after the sole header emit; mismatches skip or raise according to on_mismatch.

pandas compatibility:

  • read(size=-1), readline(), readlines(); size uses Python string/code-unit counts.
  • seek(0) to restart (other seeks raise).
  • tell(), readable(), seekable(), close(), context manager.

Data is lazy; directory contents are not fully buffered in memory.

Tagged iterators

csvdir.iter_enum.IterEnumCsvDir dataclass

Non-chunked enumerating iterator over a directory of CSVs.

Behavior expected by tests
  • enumerate() yields (stem_without_ext, row_dict)
  • iter_column()/select_columns() yield (filename_with_ext, value_or_dict)

iter_column 

iter_column(column_name)

Yield (filename_with_ext, value) for each row of each file for the given column. Header checking follows the same rules as enumerate().

select_columns 

select_columns(columns)

Yield (filename_with_ext, {col:value,...}) for each row of each file for the given columns. Header checking follows the same rules as enumerate().

csvdir.iter_path.IterPathCsvDir dataclass

iter_column 

iter_column(column_name)

Yield (path, value) for a single column across files.

If the requested column is missing in a file
  • on_mismatch == "skip": skip that file
  • on_mismatch == "error": raise ValueError

select_columns 

select_columns(columns)

Yield (path, {subset}) per row for the requested columns.

If any requested column is missing in a file
  • on_mismatch == "skip": skip that file
  • on_mismatch == "error": raise ValueError

csvdir.chunks_iter_enum.IterEnumCsvChunksDir dataclass

Chunked enumerating iterator over a directory of CSVs.

Test expectations
  • enumerate() yields (stem_without_ext, chunk_of_row_dicts)
  • iter_column_chunks()/select_columns_chunks() yield (filename_with_ext, chunk)

csvdir.chunks_iter_path.IterPathCsvChunksDir dataclass

Path utilities

csvdir.pathing

get_name 

get_name(path)

Return the filename without its extension.

get_csv_paths 

get_csv_paths(path, extension, *, recurse=False, case_insensitive=True, include_hidden=False)

Return sorted file paths under path with the given extension.

Parameters:

Name Type Description Default
path str

Directory to search.

 required
extension str

File extension (without the dot).

 required
recurse bool

If True, searches subdirectories recursively.

 False
case_insensitive bool

If True, extension matching ignores case.

 True
include_hidden bool

If True, includes files and directories starting with '.'.

 False

Utilities

csvdir.utils

strip_bom_from_headers 

strip_bom_from_headers(cols)

Remove a UTF-8 BOM marker from the start of header names.

pick_encoding 

pick_encoding(path, preferred, newline)

Choose a working text encoding for this file. Try the preferred encoding first, then safe fallbacks. Reject decodes that contain NULs (common when mis-decoding UTF-16 as UTF-8).

sniff_quotechar 

sniff_quotechar(path, *, delimiter, encoding, newline, fallback)

Detect quotechar from a sample; prefer a char actually wrapping fields containing the delimiter, then try csv.Sniffer, then smart fallbacks.

read_header 

read_header(path, *, encoding='utf-8', newline='', delimiter=',', quotechar='"', escapechar=None)

Return the header row for a CSV file, with BOM stripped and quoting auto-detected.

check_headers 

check_headers(file_headers, expected)

Compare two header lists; return (match, missing_in_file, extra_in_file).