Data Ingestion¶

Schema-aware data reading functions.

CSV Reading¶

read_csv¶

`nyctea.ingest.readers.read_csv(path, schema, lazy=None, *, typed=False)` ¶

Read a CSV with all columns as Utf8 without dtype inference.

Parameters:

Name	Type	Description	Default
`path`	`str \| Path`	Path to CSV file.	required
`schema`	`SchemaModel`	SchemaModel describing the expected columns.	required
`lazy`	`bool \| None`	Optional override. Defaults to schema.lazy.	`None`
`typed`	`bool`	When True, read with schema-declared dtypes (like Pandera/Patito). When False (default), read all columns as Utf8 and rely on parsing/coercion.	`False`

Returns:

Type	Description
`DataFrame \| LazyFrame`	pl.LazyFrame or pl.DataFrame depending on lazy flag.

Note

When typed=False, uses infer_schema=False to read all columns as strings, which is more efficient than building a full schema_overrides dict. When typed=True, uses schema_overrides with synonym support.

Source code in src/nyctea/ingest/readers.py

def read_csv(
    path: str | Path,
    schema: SchemaModel,
    lazy: bool | None = None,
    *,
    typed: bool = False,
) -> pl.DataFrame | pl.LazyFrame:
    """Read a CSV with all columns as Utf8 without dtype inference.

    Args:
        path: Path to CSV file.
        schema: SchemaModel describing the expected columns.
        lazy: Optional override. Defaults to schema.lazy.
        typed: When True, read with schema-declared dtypes (like Pandera/Patito).
            When False (default), read all columns as Utf8 and rely on parsing/coercion.

    Returns:
        pl.LazyFrame or pl.DataFrame depending on lazy flag.

    Note:
        When typed=False, uses infer_schema=False to read all columns as strings,
        which is more efficient than building a full schema_overrides dict.
        When typed=True, uses schema_overrides with synonym support.
    """
    use_lazy = schema.lazy if lazy is None else lazy

    if typed:
        # Build dtype dict using all possible column names (canonical + synonyms)
        # This ensures we match the actual CSV column names
        dtype_overrides = {}
        for canonical_name, col_schema in schema.columns.items():
            target_dtype = _to_dtype(col_schema.dtype)
            # Add canonical name
            dtype_overrides[canonical_name] = target_dtype
            # Add all synonyms
            for synonym in col_schema.synonyms:
                dtype_overrides[synonym] = target_dtype

        if use_lazy:
            return pl.scan_csv(path, schema_overrides=dtype_overrides)
        return pl.read_csv(path, schema_overrides=dtype_overrides)
    # Simpler approach: disable schema inference to read everything as strings
    if use_lazy:
        return pl.scan_csv(path, infer_schema=False)
    return pl.read_csv(path, infer_schema=False)

Parquet Reading¶

read_parquet¶

`nyctea.ingest.readers.read_parquet(path, schema, lazy=None)` ¶

Read Parquet using native types from the file.

Parameters:

Name	Type	Description	Default
`path`	`str \| Path \| list[str] \| list[Path]`	Path or paths to Parquet files.	required
`schema`	`SchemaModel`	SchemaModel describing the expected columns.	required
`lazy`	`bool \| None`	Optional override. Defaults to schema.lazy.	`None`

Returns:

Type	Description
`DataFrame \| LazyFrame`	pl.LazyFrame or pl.DataFrame depending on lazy flag.

Source code in src/nyctea/ingest/readers.py

def read_parquet(
    path: str | Path | list[str] | list[Path],
    schema: SchemaModel,
    lazy: bool | None = None,
) -> pl.DataFrame | pl.LazyFrame:
    """Read Parquet using native types from the file.

    Args:
        path: Path or paths to Parquet files.
        schema: SchemaModel describing the expected columns.
        lazy: Optional override. Defaults to schema.lazy.

    Returns:
        pl.LazyFrame or pl.DataFrame depending on lazy flag.
    """
    use_lazy = schema.lazy if lazy is None else lazy
    if use_lazy:
        return pl.scan_parquet(path)
    return pl.read_parquet(path)

Usage Examples¶

Reading CSV with String Types (Recommended)¶

This is the recommended approach for validation workflows:

from nyctea.ingest import read_csv
from nyctea.schema.model import SchemaModel

schema = SchemaModel.from_yaml_file("schema.yaml")

# All columns read as strings
lf = read_csv("data.csv", schema, lazy=True)

This prevents Polars from inferring types, giving Nyctea full control over type coercion and error handling.

Reading CSV with Declared Types¶

Use this when you want Polars to directly cast to the schema dtypes (like Pandera/Patito):

# Polars will cast columns during read
lf = read_csv("data.csv", schema, lazy=True, typed=True)

Handling Synonyms¶

The readers automatically match physical column names using canonical names and synonyms:

# Schema defines:
# canonical: "passenger_id"
# synonym: "PassengerId"

# CSV has column "PassengerId"
lf = read_csv("titanic.csv", schema, lazy=True)
# Column is automatically matched!

Lazy vs Eager¶

Control whether to return a LazyFrame or DataFrame:

# LazyFrame (recommended for large data)
lf = read_csv("data.csv", schema, lazy=True)

# DataFrame (eager evaluation)
df = read_csv("data.csv", schema, lazy=False)

# Use schema default
lf_or_df = read_csv("data.csv", schema)  # Uses schema.lazy

Advanced Usage¶

Multiple Files¶

from nyctea.ingest import read_parquet

# Read multiple Parquet files
lf = read_parquet(["part1.parquet", "part2.parquet"], schema, lazy=True)

Custom Schema Overrides¶

For advanced use cases, you can access the schema override dict:

import polars as pl
from nyctea.schema.model import SchemaModel

schema = SchemaModel.from_yaml_file("schema.yaml")

# Build custom overrides
schema_overrides = {}
for name, col_schema in schema.columns.items():
    schema_overrides[name] = pl.Utf8
    for synonym in col_schema.synonyms:
        schema_overrides[synonym] = pl.Utf8

# Use directly with Polars
lf = pl.scan_csv("data.csv", schema_overrides=schema_overrides)