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 `nyctea.Config`.	`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.

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 `nyctea.Config`.
        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.
    """
    use_lazy = schema.resolved_lazy if lazy is None else lazy

    if typed:
        dtype_overrides = {}
        for canonical_name, col_schema in schema.columns.items():
            target_dtype = _to_dtype(col_schema.dtype)
            dtype_overrides[canonical_name] = target_dtype
            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)
    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 `nyctea.Config`.	`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 `nyctea.Config`.

    Returns:
        pl.LazyFrame or pl.DataFrame depending on lazy flag.
    """
    use_lazy = schema.resolved_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 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 nyctea.Config.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 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)