contract.json

JsonParams dataclass

JsonParams(path: PathLike, encoding: str = 'UTF-8')

Parameters for JSON file processing.

Defines parameters for reading JSON files with specified encoding.

Attributes:

Name	Type	Description
path	PathLike	Path to the JSON file.
encoding	str	Character encoding for the JSON file. Defaults to UTF-8.

Json

Json(
    name: str,
    *,
    description: Optional[str] = None,
    reader_params: Optional[TReaderParams] = None,
    **kwargs,
)

Bases: DataStream[dict[str, str], JsonParams]

JSON file data stream provider.

A data stream implementation for reading single JSON objects from files.

Parameters:

Name	Type	Description	Default
DataStream		Base class for data stream providers.	required

Examples:

from contraqctor.contract.json import Json, JsonParams

# Create and load a JSON stream
config_stream = Json(
    "config",
    reader_params=JsonParams(path="config/settings.json")
)
config_stream.load()

# Access the data
config = config_stream.data
api_key = config.get("api_key")

Source code in src/contraqctor/contract/base.py

def __init__(
    self: Self,
    name: str,
    *,
    description: Optional[str] = None,
    reader_params: Optional[_typing.TReaderParams] = None,
    **kwargs,
) -> None:
    if "::" in name:
        raise ValueError("Name cannot contain '::' character.")
    self._name = name

    self._description = description
    self._reader_params = reader_params if reader_params is not None else _typing.UnsetParams
    self._data = _typing.UnsetData
    self._parent: Optional["DataStream"] = None

name property

name: str

Get the name of the data stream.

Returns:

Name	Type	Description
str	str	Name identifier of the data stream.

resolved_name property

resolved_name: str

Get the full hierarchical name of the data stream.

Generates a path-like name showing the stream's position in the hierarchy, using '::' as a separator between parent and child names.

Returns:

Name	Type	Description
str	str	The fully resolved name including all parent names.

description property

description: Optional[str]

Get the description of the data stream.

Returns:

Type	Description
Optional[str]	Optional[str]: Description of the data stream, or None if not provided.

parent property

parent: Optional[DataStream]

Get the parent data stream.

Returns:

Type	Description
Optional[DataStream]	Optional[DataStream]: Parent data stream, or None if this is a root stream.

is_collection property

is_collection: bool

Check if this data stream is a collection of other streams.

Returns:

Name	Type	Description
bool	bool	True if this is a collection stream, False otherwise.

reader_params property

reader_params: TReaderParams

Get the parameters for the data reader.

Returns:

Name	Type	Description
TReaderParams	TReaderParams	Parameters for the data reader.

has_data property

has_data: bool

Check if the data stream has loaded data.

Returns:

Name	Type	Description
bool	bool	True if data has been loaded, False otherwise.

data property

data: TData

Get the loaded data.

Returns:

Name	Type	Description
TData	TData	The loaded data.

Raises:

Type	Description
ValueError	If data has not been loaded yet.

read

read(
    reader_params: Optional[TReaderParams] = None,
) -> TData

Read data using the configured reader.

Parameters:

Name	Type	Description	Default
reader_params	Optional[TReaderParams]	Optional parameters to override the default reader parameters.	None

Returns:

Name	Type	Description
TData	TData	Data read from the source.

Raises:

Type	Description
ValueError	If reader parameters are not set.

Source code in src/contraqctor/contract/base.py

def read(self, reader_params: Optional[_typing.TReaderParams] = None) -> _typing.TData:
    """Read data using the configured reader.

    Args:
        reader_params: Optional parameters to override the default reader parameters.

    Returns:
        TData: Data read from the source.

    Raises:
        ValueError: If reader parameters are not set.
    """
    reader_params = reader_params if reader_params is not None else self._reader_params
    if _typing.is_unset(reader_params):
        raise ValueError("Reader parameters are not set. Cannot read data.")
    return self._reader(reader_params)

bind_reader_params

bind_reader_params(params: TReaderParams) -> Self

Bind reader parameters to the data stream.

Parameters:

Name	Type	Description	Default
params	TReaderParams	Parameters to bind to the data stream's reader.	required

Returns:

Name	Type	Description
Self	Self	The data stream instance for method chaining.

Raises:

Type	Description
ValueError	If reader parameters have already been set.

Source code in src/contraqctor/contract/base.py

def bind_reader_params(self, params: _typing.TReaderParams) -> Self:
    """Bind reader parameters to the data stream.

    Args:
        params: Parameters to bind to the data stream's reader.

    Returns:
        Self: The data stream instance for method chaining.

    Raises:
        ValueError: If reader parameters have already been set.
    """
    if not _typing.is_unset(self._reader_params):
        raise ValueError("Reader parameters are already set. Cannot bind again.")
    self._reader_params = params
    return self

at

at(name: str) -> DataStream

Get a child data stream by name.

Parameters:

Name	Type	Description	Default
name	str	Name of the child data stream to retrieve.	required

Returns:

Name	Type	Description
DataStream	DataStream	The child data stream with the given name.

Raises:

Type	Description
NotImplementedError	If the data stream does not support child access.

Examples:

# Access stream in a collection
collection = data_collection.load()
temp_stream = collection.at("temperature")

# Or using dictionary-style syntax
humidity_stream = collection["humidity"]

Source code in src/contraqctor/contract/base.py

def at(self, name: str) -> "DataStream":
    """Get a child data stream by name.

    Args:
        name: Name of the child data stream to retrieve.

    Returns:
        DataStream: The child data stream with the given name.

    Raises:
        NotImplementedError: If the data stream does not support child access.

    Examples:
        ```python
        # Access stream in a collection
        collection = data_collection.load()
        temp_stream = collection.at("temperature")

        # Or using dictionary-style syntax
        humidity_stream = collection["humidity"]
        ```
    """
    raise NotImplementedError("This method is not implemented for DataStream.")

load

load() -> Self

Load data into the data stream.

Reads data from the source and stores it in the data stream.

Returns:

Name	Type	Description
Self	Self	The data stream instance for method chaining.

Examples:

from contraqctor.contract import csv

# Create and load a CSV stream
params = csv.CsvParams(path="data/measurements.csv")
csv_stream = csv.Csv("measurements", reader_params=params)
csv_stream.load()

# Access the data
df = csv_stream.data
print(f"Loaded {len(df)} rows")

Source code in src/contraqctor/contract/base.py

def load(self) -> Self:
    """Load data into the data stream.

    Reads data from the source and stores it in the data stream.

    Returns:
        Self: The data stream instance for method chaining.

    Examples:
        ```python
        from contraqctor.contract import csv

        # Create and load a CSV stream
        params = csv.CsvParams(path="data/measurements.csv")
        csv_stream = csv.Csv("measurements", reader_params=params)
        csv_stream.load()

        # Access the data
        df = csv_stream.data
        print(f"Loaded {len(df)} rows")
        ```
    """
    self._data = self.read()
    return self

load_all

load_all(
    strict: bool = False,
) -> list[tuple[DataStream, Exception], None, None]

Recursively load this data stream and all child streams.

Performs depth-first traversal to load all streams in the hierarchy.

Parameters:

Name	Type	Description	Default
strict	bool	If True, raises exceptions immediately; otherwise collects and returns them.	False

Returns:

Name	Type	Description
list	list[tuple[DataStream, Exception], None, None]	List of tuples containing streams and exceptions that occurred during loading.

Raises:

Type	Description
Exception	If strict is True and an exception occurs during loading.

Examples:

# Load all streams and handle errors
errors = collection.load_all(strict=False)

if errors:
    for stream, error in errors:
        print(f"Error loading {stream.name}: {error}")

Source code in src/contraqctor/contract/base.py

def load_all(self, strict: bool = False) -> list[tuple["DataStream", Exception], None, None]:
    """Recursively load this data stream and all child streams.

    Performs depth-first traversal to load all streams in the hierarchy.

    Args:
        strict: If True, raises exceptions immediately; otherwise collects and returns them.

    Returns:
        list: List of tuples containing streams and exceptions that occurred during loading.

    Raises:
        Exception: If strict is True and an exception occurs during loading.

    Examples:
        ```python
        # Load all streams and handle errors
        errors = collection.load_all(strict=False)

        if errors:
            for stream, error in errors:
                print(f"Error loading {stream.name}: {error}")
        ```
    """
    self.load()
    exceptions = []
    for stream in self:
        if stream is None:
            continue
        try:
            exceptions += stream.load_all(strict=strict)
        except Exception as e:
            if strict:
                raise e
            exceptions.append((stream, e))
    return exceptions

MultiLineJson

MultiLineJson(
    name: str,
    *,
    description: Optional[str] = None,
    reader_params: Optional[TReaderParams] = None,
    **kwargs,
)

Bases: DataStream[list[dict[str, str]], JsonParams]

Multi-line JSON file data stream provider.

A data stream implementation for reading JSON files where each line contains a separate JSON object.

Parameters:

Name	Type	Description	Default
DataStream		Base class for data stream providers.	required

Examples:

from contraqctor.contract.json import MultiLineJson, JsonParams

# Create and load a multi-line JSON stream
logs_stream = MultiLineJson(
    "server_logs",
    reader_params=JsonParams(path="logs/server_logs.jsonl")
)
logs_stream.load()

# Process log entries
for entry in logs_stream.data:
    if entry.get("level") == "ERROR":
        print(f"Error: {entry.get('message')}")

Source code in src/contraqctor/contract/base.py

def __init__(
    self: Self,
    name: str,
    *,
    description: Optional[str] = None,
    reader_params: Optional[_typing.TReaderParams] = None,
    **kwargs,
) -> None:
    if "::" in name:
        raise ValueError("Name cannot contain '::' character.")
    self._name = name

    self._description = description
    self._reader_params = reader_params if reader_params is not None else _typing.UnsetParams
    self._data = _typing.UnsetData
    self._parent: Optional["DataStream"] = None

name property

name: str

Get the name of the data stream.

Returns:

Name	Type	Description
str	str	Name identifier of the data stream.

resolved_name property

resolved_name: str

Get the full hierarchical name of the data stream.

Generates a path-like name showing the stream's position in the hierarchy, using '::' as a separator between parent and child names.

Returns:

Name	Type	Description
str	str	The fully resolved name including all parent names.

description property

description: Optional[str]

Get the description of the data stream.

Returns:

Type	Description
Optional[str]	Optional[str]: Description of the data stream, or None if not provided.

parent property

parent: Optional[DataStream]

Get the parent data stream.

Returns:

Type	Description
Optional[DataStream]	Optional[DataStream]: Parent data stream, or None if this is a root stream.

is_collection property

is_collection: bool

Check if this data stream is a collection of other streams.

Returns:

Name	Type	Description
bool	bool	True if this is a collection stream, False otherwise.

reader_params property

reader_params: TReaderParams

Get the parameters for the data reader.

Returns:

Name	Type	Description
TReaderParams	TReaderParams	Parameters for the data reader.

has_data property

has_data: bool

Check if the data stream has loaded data.

Returns:

Name	Type	Description
bool	bool	True if data has been loaded, False otherwise.

data property

data: TData

Get the loaded data.

Returns:

Name	Type	Description
TData	TData	The loaded data.

Raises:

Type	Description
ValueError	If data has not been loaded yet.

read

read(
    reader_params: Optional[TReaderParams] = None,
) -> TData

Read data using the configured reader.

Parameters:

Name	Type	Description	Default
reader_params	Optional[TReaderParams]	Optional parameters to override the default reader parameters.	None

Returns:

Name	Type	Description
TData	TData	Data read from the source.

Raises:

Type	Description
ValueError	If reader parameters are not set.

Source code in src/contraqctor/contract/base.py

def read(self, reader_params: Optional[_typing.TReaderParams] = None) -> _typing.TData:
    """Read data using the configured reader.

    Args:
        reader_params: Optional parameters to override the default reader parameters.

    Returns:
        TData: Data read from the source.

    Raises:
        ValueError: If reader parameters are not set.
    """
    reader_params = reader_params if reader_params is not None else self._reader_params
    if _typing.is_unset(reader_params):
        raise ValueError("Reader parameters are not set. Cannot read data.")
    return self._reader(reader_params)

bind_reader_params

bind_reader_params(params: TReaderParams) -> Self

Bind reader parameters to the data stream.

Parameters:

Name	Type	Description	Default
params	TReaderParams	Parameters to bind to the data stream's reader.	required

Returns:

Name	Type	Description
Self	Self	The data stream instance for method chaining.

Raises:

Type	Description
ValueError	If reader parameters have already been set.

Source code in src/contraqctor/contract/base.py

def bind_reader_params(self, params: _typing.TReaderParams) -> Self:
    """Bind reader parameters to the data stream.

    Args:
        params: Parameters to bind to the data stream's reader.

    Returns:
        Self: The data stream instance for method chaining.

    Raises:
        ValueError: If reader parameters have already been set.
    """
    if not _typing.is_unset(self._reader_params):
        raise ValueError("Reader parameters are already set. Cannot bind again.")
    self._reader_params = params
    return self

at

at(name: str) -> DataStream

Get a child data stream by name.

Parameters:

Name	Type	Description	Default
name	str	Name of the child data stream to retrieve.	required

Returns:

Name	Type	Description
DataStream	DataStream	The child data stream with the given name.

Raises:

Type	Description
NotImplementedError	If the data stream does not support child access.

Examples:

# Access stream in a collection
collection = data_collection.load()
temp_stream = collection.at("temperature")

# Or using dictionary-style syntax
humidity_stream = collection["humidity"]

Source code in src/contraqctor/contract/base.py

def at(self, name: str) -> "DataStream":
    """Get a child data stream by name.

    Args:
        name: Name of the child data stream to retrieve.

    Returns:
        DataStream: The child data stream with the given name.

    Raises:
        NotImplementedError: If the data stream does not support child access.

    Examples:
        ```python
        # Access stream in a collection
        collection = data_collection.load()
        temp_stream = collection.at("temperature")

        # Or using dictionary-style syntax
        humidity_stream = collection["humidity"]
        ```
    """
    raise NotImplementedError("This method is not implemented for DataStream.")

load

load() -> Self

Load data into the data stream.

Reads data from the source and stores it in the data stream.

Returns:

Name	Type	Description
Self	Self	The data stream instance for method chaining.

Examples:

from contraqctor.contract import csv

# Create and load a CSV stream
params = csv.CsvParams(path="data/measurements.csv")
csv_stream = csv.Csv("measurements", reader_params=params)
csv_stream.load()

# Access the data
df = csv_stream.data
print(f"Loaded {len(df)} rows")

Source code in src/contraqctor/contract/base.py

def load(self) -> Self:
    """Load data into the data stream.

    Reads data from the source and stores it in the data stream.

    Returns:
        Self: The data stream instance for method chaining.

    Examples:
        ```python
        from contraqctor.contract import csv

        # Create and load a CSV stream
        params = csv.CsvParams(path="data/measurements.csv")
        csv_stream = csv.Csv("measurements", reader_params=params)
        csv_stream.load()

        # Access the data
        df = csv_stream.data
        print(f"Loaded {len(df)} rows")
        ```
    """
    self._data = self.read()
    return self

load_all

load_all(
    strict: bool = False,
) -> list[tuple[DataStream, Exception], None, None]

Recursively load this data stream and all child streams.

Performs depth-first traversal to load all streams in the hierarchy.

Parameters:

Name	Type	Description	Default
strict	bool	If True, raises exceptions immediately; otherwise collects and returns them.	False

Returns:

Name	Type	Description
list	list[tuple[DataStream, Exception], None, None]	List of tuples containing streams and exceptions that occurred during loading.

Raises:

Type	Description
Exception	If strict is True and an exception occurs during loading.

Examples:

# Load all streams and handle errors
errors = collection.load_all(strict=False)

if errors:
    for stream, error in errors:
        print(f"Error loading {stream.name}: {error}")

Source code in src/contraqctor/contract/base.py

def load_all(self, strict: bool = False) -> list[tuple["DataStream", Exception], None, None]:
    """Recursively load this data stream and all child streams.

    Performs depth-first traversal to load all streams in the hierarchy.

    Args:
        strict: If True, raises exceptions immediately; otherwise collects and returns them.

    Returns:
        list: List of tuples containing streams and exceptions that occurred during loading.

    Raises:
        Exception: If strict is True and an exception occurs during loading.

    Examples:
        ```python
        # Load all streams and handle errors
        errors = collection.load_all(strict=False)

        if errors:
            for stream, error in errors:
                print(f"Error loading {stream.name}: {error}")
        ```
    """
    self.load()
    exceptions = []
    for stream in self:
        if stream is None:
            continue
        try:
            exceptions += stream.load_all(strict=strict)
        except Exception as e:
            if strict:
                raise e
            exceptions.append((stream, e))
    return exceptions

PydanticModelParams dataclass

PydanticModelParams(
    path: PathLike,
    model: Type[_TModel],
    encoding: str = "UTF-8",
)

Bases: FilePathBaseParam, Generic[_TModel]

Parameters for Pydantic model-based JSON file processing.

Extends the base file path parameters with Pydantic model specification for parsing JSON into typed objects.

Attributes:

Name	Type	Description
model	Type[_TModel]	Pydantic model class to use for parsing JSON data.
encoding	str	Character encoding for the JSON file. Defaults to UTF-8.

Examples:

from pydantic import BaseModel
from contraqctor.contract.json import PydanticModelParams

class User(BaseModel):
    user_id: str
    name: str
    active: bool = True

params = PydanticModelParams(path="users/profile.json", model=User)

PydanticModel

PydanticModel(
    name: str,
    *,
    description: Optional[str] = None,
    reader_params: Optional[TReaderParams] = None,
    **kwargs,
)

Bases: DataStream[_TModel, PydanticModelParams[_TModel]]

Pydantic model-based JSON data stream provider.

A data stream implementation for reading JSON files as Pydantic model instances.

Parameters:

Name	Type	Description	Default
DataStream		Base class for data stream providers.	required

Examples:

from pydantic import BaseModel
from contraqctor.contract.json import PydanticModel, PydanticModelParams

class ServerConfig(BaseModel):
    host: str
    port: int
    debug: bool = False

params = PydanticModelParams(path="config/server.json", model=ServerConfig)

config_stream = PydanticModel("server_config", reader_params=params).load()
server_config = config_stream.data
print(f"Server: {server_config.host}:{server_config.port}")

Source code in src/contraqctor/contract/base.py

def __init__(
    self: Self,
    name: str,
    *,
    description: Optional[str] = None,
    reader_params: Optional[_typing.TReaderParams] = None,
    **kwargs,
) -> None:
    if "::" in name:
        raise ValueError("Name cannot contain '::' character.")
    self._name = name

    self._description = description
    self._reader_params = reader_params if reader_params is not None else _typing.UnsetParams
    self._data = _typing.UnsetData
    self._parent: Optional["DataStream"] = None

name property

name: str

Get the name of the data stream.

Returns:

Name	Type	Description
str	str	Name identifier of the data stream.

resolved_name property

resolved_name: str

Get the full hierarchical name of the data stream.

Generates a path-like name showing the stream's position in the hierarchy, using '::' as a separator between parent and child names.

Returns:

Name	Type	Description
str	str	The fully resolved name including all parent names.

description property

description: Optional[str]

Get the description of the data stream.

Returns:

Type	Description
Optional[str]	Optional[str]: Description of the data stream, or None if not provided.

parent property

parent: Optional[DataStream]

Get the parent data stream.

Returns:

Type	Description
Optional[DataStream]	Optional[DataStream]: Parent data stream, or None if this is a root stream.

is_collection property

is_collection: bool

Check if this data stream is a collection of other streams.

Returns:

Name	Type	Description
bool	bool	True if this is a collection stream, False otherwise.

reader_params property

reader_params: TReaderParams

Get the parameters for the data reader.

Returns:

Name	Type	Description
TReaderParams	TReaderParams	Parameters for the data reader.

has_data property

has_data: bool

Check if the data stream has loaded data.

Returns:

Name	Type	Description
bool	bool	True if data has been loaded, False otherwise.

data property

data: TData

Get the loaded data.

Returns:

Name	Type	Description
TData	TData	The loaded data.

Raises:

Type	Description
ValueError	If data has not been loaded yet.

read

read(
    reader_params: Optional[TReaderParams] = None,
) -> TData

Read data using the configured reader.

Parameters:

Name	Type	Description	Default
reader_params	Optional[TReaderParams]	Optional parameters to override the default reader parameters.	None

Returns:

Name	Type	Description
TData	TData	Data read from the source.

Raises:

Type	Description
ValueError	If reader parameters are not set.

Source code in src/contraqctor/contract/base.py

def read(self, reader_params: Optional[_typing.TReaderParams] = None) -> _typing.TData:
    """Read data using the configured reader.

    Args:
        reader_params: Optional parameters to override the default reader parameters.

    Returns:
        TData: Data read from the source.

    Raises:
        ValueError: If reader parameters are not set.
    """
    reader_params = reader_params if reader_params is not None else self._reader_params
    if _typing.is_unset(reader_params):
        raise ValueError("Reader parameters are not set. Cannot read data.")
    return self._reader(reader_params)

bind_reader_params

bind_reader_params(params: TReaderParams) -> Self

Bind reader parameters to the data stream.

Parameters:

Name	Type	Description	Default
params	TReaderParams	Parameters to bind to the data stream's reader.	required

Returns:

Name	Type	Description
Self	Self	The data stream instance for method chaining.

Raises:

Type	Description
ValueError	If reader parameters have already been set.

Source code in src/contraqctor/contract/base.py

def bind_reader_params(self, params: _typing.TReaderParams) -> Self:
    """Bind reader parameters to the data stream.

    Args:
        params: Parameters to bind to the data stream's reader.

    Returns:
        Self: The data stream instance for method chaining.

    Raises:
        ValueError: If reader parameters have already been set.
    """
    if not _typing.is_unset(self._reader_params):
        raise ValueError("Reader parameters are already set. Cannot bind again.")
    self._reader_params = params
    return self

at

at(name: str) -> DataStream

Get a child data stream by name.

Parameters:

Name	Type	Description	Default
name	str	Name of the child data stream to retrieve.	required

Returns:

Name	Type	Description
DataStream	DataStream	The child data stream with the given name.

Raises:

Type	Description
NotImplementedError	If the data stream does not support child access.

Examples:

# Access stream in a collection
collection = data_collection.load()
temp_stream = collection.at("temperature")

# Or using dictionary-style syntax
humidity_stream = collection["humidity"]

Source code in src/contraqctor/contract/base.py

def at(self, name: str) -> "DataStream":
    """Get a child data stream by name.

    Args:
        name: Name of the child data stream to retrieve.

    Returns:
        DataStream: The child data stream with the given name.

    Raises:
        NotImplementedError: If the data stream does not support child access.

    Examples:
        ```python
        # Access stream in a collection
        collection = data_collection.load()
        temp_stream = collection.at("temperature")

        # Or using dictionary-style syntax
        humidity_stream = collection["humidity"]
        ```
    """
    raise NotImplementedError("This method is not implemented for DataStream.")

load

load() -> Self

Load data into the data stream.

Reads data from the source and stores it in the data stream.

Returns:

Name	Type	Description
Self	Self	The data stream instance for method chaining.

Examples:

from contraqctor.contract import csv

# Create and load a CSV stream
params = csv.CsvParams(path="data/measurements.csv")
csv_stream = csv.Csv("measurements", reader_params=params)
csv_stream.load()

# Access the data
df = csv_stream.data
print(f"Loaded {len(df)} rows")

Source code in src/contraqctor/contract/base.py

def load(self) -> Self:
    """Load data into the data stream.

    Reads data from the source and stores it in the data stream.

    Returns:
        Self: The data stream instance for method chaining.

    Examples:
        ```python
        from contraqctor.contract import csv

        # Create and load a CSV stream
        params = csv.CsvParams(path="data/measurements.csv")
        csv_stream = csv.Csv("measurements", reader_params=params)
        csv_stream.load()

        # Access the data
        df = csv_stream.data
        print(f"Loaded {len(df)} rows")
        ```
    """
    self._data = self.read()
    return self

load_all

load_all(
    strict: bool = False,
) -> list[tuple[DataStream, Exception], None, None]

Recursively load this data stream and all child streams.

Performs depth-first traversal to load all streams in the hierarchy.

Parameters:

Name	Type	Description	Default
strict	bool	If True, raises exceptions immediately; otherwise collects and returns them.	False

Returns:

Name	Type	Description
list	list[tuple[DataStream, Exception], None, None]	List of tuples containing streams and exceptions that occurred during loading.

Raises:

Type	Description
Exception	If strict is True and an exception occurs during loading.

Examples:

# Load all streams and handle errors
errors = collection.load_all(strict=False)

if errors:
    for stream, error in errors:
        print(f"Error loading {stream.name}: {error}")

Source code in src/contraqctor/contract/base.py

def load_all(self, strict: bool = False) -> list[tuple["DataStream", Exception], None, None]:
    """Recursively load this data stream and all child streams.

    Performs depth-first traversal to load all streams in the hierarchy.

    Args:
        strict: If True, raises exceptions immediately; otherwise collects and returns them.

    Returns:
        list: List of tuples containing streams and exceptions that occurred during loading.

    Raises:
        Exception: If strict is True and an exception occurs during loading.

    Examples:
        ```python
        # Load all streams and handle errors
        errors = collection.load_all(strict=False)

        if errors:
            for stream, error in errors:
                print(f"Error loading {stream.name}: {error}")
        ```
    """
    self.load()
    exceptions = []
    for stream in self:
        if stream is None:
            continue
        try:
            exceptions += stream.load_all(strict=strict)
        except Exception as e:
            if strict:
                raise e
            exceptions.append((stream, e))
    return exceptions

ManyPydanticModelParams dataclass

ManyPydanticModelParams(
    path: PathLike,
    model: Type[_TModel],
    encoding: str = "UTF-8",
    index: Optional[str] = None,
    column_names: Optional[dict[str, str]] = None,
)

Bases: FilePathBaseParam, Generic[_TModel]

Parameters for loading multiple Pydantic models from a file.

Extends the base file path parameters with Pydantic model specification and options for converting to a DataFrame.

Attributes:

Name	Type	Description
model	Type[_TModel]	Pydantic model class to use for parsing JSON data.
encoding	str	Character encoding for the JSON file. Defaults to UTF-8.
index	Optional[str]	Optional column name to set as the DataFrame index.
column_names	Optional[dict[str, str]]	Optional dictionary mapping original column names to new names.

Examples:

Defining parameters to load multiple models:

from pydantic import BaseModel
from contraqctor.contract.json import ManyPydanticModelParams

# Define a Pydantic model for log entries
class LogEntry(BaseModel):
    timestamp: str
    level: str
    message: str

# Create parameters for loading log entries
params = ManyPydanticModelParams(
    path="logs/server_logs.json",
    model=LogEntry,
    index="timestamp",
    column_names={"level": "log_level", "message": "log_message"}
)

ManyPydanticModel

ManyPydanticModel(
    name: str,
    *,
    description: Optional[str] = None,
    reader_params: Optional[TReaderParams] = None,
    **kwargs,
)

Bases: DataStream[DataFrame, ManyPydanticModelParams[_TModel]]

Multi-model JSON data stream provider.

A data stream implementation for reading multiple JSON objects from a file, parsing them as Pydantic models, and returning them as a DataFrame.

Parameters:

Name	Type	Description	Default
DataStream		Base class for data stream providers.	required

Examples:

Loading server logs into a DataFrame:

from contraqctor.contract.json import ManyPydanticModel, ManyPydanticModelParams

# Create and load the data stream
logs_stream = ManyPydanticModel(
    "server_logs_df",
    reader_params=params
)
logs_stream.load()

# Access the logs as a DataFrame
logs_df = logs_stream.data

# Analyze the logs
error_logs = logs_df[logs_df["log_level"] == "ERROR"]

Source code in src/contraqctor/contract/base.py

def __init__(
    self: Self,
    name: str,
    *,
    description: Optional[str] = None,
    reader_params: Optional[_typing.TReaderParams] = None,
    **kwargs,
) -> None:
    if "::" in name:
        raise ValueError("Name cannot contain '::' character.")
    self._name = name

    self._description = description
    self._reader_params = reader_params if reader_params is not None else _typing.UnsetParams
    self._data = _typing.UnsetData
    self._parent: Optional["DataStream"] = None

name property

name: str

Get the name of the data stream.

Returns:

Name	Type	Description
str	str	Name identifier of the data stream.

resolved_name property

resolved_name: str

Get the full hierarchical name of the data stream.

Generates a path-like name showing the stream's position in the hierarchy, using '::' as a separator between parent and child names.

Returns:

Name	Type	Description
str	str	The fully resolved name including all parent names.

description property

description: Optional[str]

Get the description of the data stream.

Returns:

Type	Description
Optional[str]	Optional[str]: Description of the data stream, or None if not provided.

parent property

parent: Optional[DataStream]

Get the parent data stream.

Returns:

Type	Description
Optional[DataStream]	Optional[DataStream]: Parent data stream, or None if this is a root stream.

is_collection property

is_collection: bool

Check if this data stream is a collection of other streams.

Returns:

Name	Type	Description
bool	bool	True if this is a collection stream, False otherwise.

reader_params property

reader_params: TReaderParams

Get the parameters for the data reader.

Returns:

Name	Type	Description
TReaderParams	TReaderParams	Parameters for the data reader.

has_data property

has_data: bool

Check if the data stream has loaded data.

Returns:

Name	Type	Description
bool	bool	True if data has been loaded, False otherwise.

data property

data: TData

Get the loaded data.

Returns:

Name	Type	Description
TData	TData	The loaded data.

Raises:

Type	Description
ValueError	If data has not been loaded yet.

read

read(
    reader_params: Optional[TReaderParams] = None,
) -> TData

Read data using the configured reader.

Parameters:

Name	Type	Description	Default
reader_params	Optional[TReaderParams]	Optional parameters to override the default reader parameters.	None

Returns:

Name	Type	Description
TData	TData	Data read from the source.

Raises:

Type	Description
ValueError	If reader parameters are not set.

Source code in src/contraqctor/contract/base.py

def read(self, reader_params: Optional[_typing.TReaderParams] = None) -> _typing.TData:
    """Read data using the configured reader.

    Args:
        reader_params: Optional parameters to override the default reader parameters.

    Returns:
        TData: Data read from the source.

    Raises:
        ValueError: If reader parameters are not set.
    """
    reader_params = reader_params if reader_params is not None else self._reader_params
    if _typing.is_unset(reader_params):
        raise ValueError("Reader parameters are not set. Cannot read data.")
    return self._reader(reader_params)

bind_reader_params

bind_reader_params(params: TReaderParams) -> Self

Bind reader parameters to the data stream.

Parameters:

Name	Type	Description	Default
params	TReaderParams	Parameters to bind to the data stream's reader.	required

Returns:

Name	Type	Description
Self	Self	The data stream instance for method chaining.

Raises:

Type	Description
ValueError	If reader parameters have already been set.

Source code in src/contraqctor/contract/base.py

def bind_reader_params(self, params: _typing.TReaderParams) -> Self:
    """Bind reader parameters to the data stream.

    Args:
        params: Parameters to bind to the data stream's reader.

    Returns:
        Self: The data stream instance for method chaining.

    Raises:
        ValueError: If reader parameters have already been set.
    """
    if not _typing.is_unset(self._reader_params):
        raise ValueError("Reader parameters are already set. Cannot bind again.")
    self._reader_params = params
    return self

at

at(name: str) -> DataStream

Get a child data stream by name.

Parameters:

Name	Type	Description	Default
name	str	Name of the child data stream to retrieve.	required

Returns:

Name	Type	Description
DataStream	DataStream	The child data stream with the given name.

Raises:

Type	Description
NotImplementedError	If the data stream does not support child access.

Examples:

# Access stream in a collection
collection = data_collection.load()
temp_stream = collection.at("temperature")

# Or using dictionary-style syntax
humidity_stream = collection["humidity"]

Source code in src/contraqctor/contract/base.py

def at(self, name: str) -> "DataStream":
    """Get a child data stream by name.

    Args:
        name: Name of the child data stream to retrieve.

    Returns:
        DataStream: The child data stream with the given name.

    Raises:
        NotImplementedError: If the data stream does not support child access.

    Examples:
        ```python
        # Access stream in a collection
        collection = data_collection.load()
        temp_stream = collection.at("temperature")

        # Or using dictionary-style syntax
        humidity_stream = collection["humidity"]
        ```
    """
    raise NotImplementedError("This method is not implemented for DataStream.")

load

load() -> Self

Load data into the data stream.

Reads data from the source and stores it in the data stream.

Returns:

Name	Type	Description
Self	Self	The data stream instance for method chaining.

Examples:

from contraqctor.contract import csv

# Create and load a CSV stream
params = csv.CsvParams(path="data/measurements.csv")
csv_stream = csv.Csv("measurements", reader_params=params)
csv_stream.load()

# Access the data
df = csv_stream.data
print(f"Loaded {len(df)} rows")

Source code in src/contraqctor/contract/base.py

def load(self) -> Self:
    """Load data into the data stream.

    Reads data from the source and stores it in the data stream.

    Returns:
        Self: The data stream instance for method chaining.

    Examples:
        ```python
        from contraqctor.contract import csv

        # Create and load a CSV stream
        params = csv.CsvParams(path="data/measurements.csv")
        csv_stream = csv.Csv("measurements", reader_params=params)
        csv_stream.load()

        # Access the data
        df = csv_stream.data
        print(f"Loaded {len(df)} rows")
        ```
    """
    self._data = self.read()
    return self

load_all

load_all(
    strict: bool = False,
) -> list[tuple[DataStream, Exception], None, None]

Recursively load this data stream and all child streams.

Performs depth-first traversal to load all streams in the hierarchy.

Parameters:

Name	Type	Description	Default
strict	bool	If True, raises exceptions immediately; otherwise collects and returns them.	False

Returns:

Name	Type	Description
list	list[tuple[DataStream, Exception], None, None]	List of tuples containing streams and exceptions that occurred during loading.

Raises:

Type	Description
Exception	If strict is True and an exception occurs during loading.

Examples:

# Load all streams and handle errors
errors = collection.load_all(strict=False)

if errors:
    for stream, error in errors:
        print(f"Error loading {stream.name}: {error}")

Source code in src/contraqctor/contract/base.py

def load_all(self, strict: bool = False) -> list[tuple["DataStream", Exception], None, None]:
    """Recursively load this data stream and all child streams.

    Performs depth-first traversal to load all streams in the hierarchy.

    Args:
        strict: If True, raises exceptions immediately; otherwise collects and returns them.

    Returns:
        list: List of tuples containing streams and exceptions that occurred during loading.

    Raises:
        Exception: If strict is True and an exception occurs during loading.

    Examples:
        ```python
        # Load all streams and handle errors
        errors = collection.load_all(strict=False)

        if errors:
            for stream, error in errors:
                print(f"Error loading {stream.name}: {error}")
        ```
    """
    self.load()
    exceptions = []
    for stream in self:
        if stream is None:
            continue
        try:
            exceptions += stream.load_all(strict=strict)
        except Exception as e:
            if strict:
                raise e
            exceptions.append((stream, e))
    return exceptions

SoftwareEventsParams dataclass

SoftwareEventsParams(
    path: PathLike,
    model: Type[_TModel],
    encoding: str = "UTF-8",
    index: Optional[str] = None,
    column_names: Optional[dict[str, str]] = None,
)

Bases: ManyPydanticModelParams

Parameters for software events file processing.

A specialized version of ManyPydanticModelParams that defaults to using the SoftwareEvent model from aind_behavior_services.

Attributes:

Name	Type	Description
model	Type[SoftwareEvent]	Set to SoftwareEvent model and not modifiable after initialization.
encoding	str	Character encoding for the JSON file. Defaults to UTF-8.
index	Optional[str]	Optional column name to set as the DataFrame index.
column_names	Optional[dict[str, str]]	Optional dictionary mapping original column names to new names.

Examples:

Defining parameters for loading software events:

from contraqctor.contract.json import SoftwareEventsParams

# Create parameters for software events
params = SoftwareEventsParams(
    path="events/software_events.json",
    index="event_id",
    column_names={"timestamp": "event_time"}
)

SoftwareEvents

SoftwareEvents(
    name: str,
    *,
    description: Optional[str] = None,
    reader_params: Optional[TReaderParams] = None,
    **kwargs,
)

Bases: ManyPydanticModel[SoftwareEvent]

Software events data stream provider.

A specialized data stream for reading software event logs from JSON files using the SoftwareEvent model from aind_behavior_services.

Parameters:

Name	Type	Description	Default
ManyPydanticModel		Base class for multi-model data stream providers.	required

Examples:

Analyzing software events data:

from contraqctor.contract.json import SoftwareEvents, SoftwareEventsParams

# Create parameters for software events
params = SoftwareEventsParams(
    path="events/software_events.json",
    index="event_id"
)

# Create and load the software events stream
events_stream = SoftwareEvents(
    "software_events",
    reader_params=params
)
events_stream.load()

# Access the events data
events_df = events_stream.data

# Perform analysis, e.g., count events by type
event_counts = events_df["event_type"].value_counts()

Source code in src/contraqctor/contract/base.py

def __init__(
    self: Self,
    name: str,
    *,
    description: Optional[str] = None,
    reader_params: Optional[_typing.TReaderParams] = None,
    **kwargs,
) -> None:
    if "::" in name:
        raise ValueError("Name cannot contain '::' character.")
    self._name = name

    self._description = description
    self._reader_params = reader_params if reader_params is not None else _typing.UnsetParams
    self._data = _typing.UnsetData
    self._parent: Optional["DataStream"] = None

name property

name: str

Get the name of the data stream.

Returns:

Name	Type	Description
str	str	Name identifier of the data stream.

resolved_name property

resolved_name: str

Get the full hierarchical name of the data stream.

Generates a path-like name showing the stream's position in the hierarchy, using '::' as a separator between parent and child names.

Returns:

Name	Type	Description
str	str	The fully resolved name including all parent names.

description property

description: Optional[str]

Get the description of the data stream.

Returns:

Type	Description
Optional[str]	Optional[str]: Description of the data stream, or None if not provided.

parent property

parent: Optional[DataStream]

Get the parent data stream.

Returns:

Type	Description
Optional[DataStream]	Optional[DataStream]: Parent data stream, or None if this is a root stream.

is_collection property

is_collection: bool

Check if this data stream is a collection of other streams.

Returns:

Name	Type	Description
bool	bool	True if this is a collection stream, False otherwise.

reader_params property

reader_params: TReaderParams

Get the parameters for the data reader.

Returns:

Name	Type	Description
TReaderParams	TReaderParams	Parameters for the data reader.

has_data property

has_data: bool

Check if the data stream has loaded data.

Returns:

Name	Type	Description
bool	bool	True if data has been loaded, False otherwise.

data property

data: TData

Get the loaded data.

Returns:

Name	Type	Description
TData	TData	The loaded data.

Raises:

Type	Description
ValueError	If data has not been loaded yet.

read

read(
    reader_params: Optional[TReaderParams] = None,
) -> TData

Read data using the configured reader.

Parameters:

Name	Type	Description	Default
reader_params	Optional[TReaderParams]	Optional parameters to override the default reader parameters.	None

Returns:

Name	Type	Description
TData	TData	Data read from the source.

Raises:

Type	Description
ValueError	If reader parameters are not set.

Source code in src/contraqctor/contract/base.py

def read(self, reader_params: Optional[_typing.TReaderParams] = None) -> _typing.TData:
    """Read data using the configured reader.

    Args:
        reader_params: Optional parameters to override the default reader parameters.

    Returns:
        TData: Data read from the source.

    Raises:
        ValueError: If reader parameters are not set.
    """
    reader_params = reader_params if reader_params is not None else self._reader_params
    if _typing.is_unset(reader_params):
        raise ValueError("Reader parameters are not set. Cannot read data.")
    return self._reader(reader_params)

bind_reader_params

bind_reader_params(params: TReaderParams) -> Self

Bind reader parameters to the data stream.

Parameters:

Name	Type	Description	Default
params	TReaderParams	Parameters to bind to the data stream's reader.	required

Returns:

Name	Type	Description
Self	Self	The data stream instance for method chaining.

Raises:

Type	Description
ValueError	If reader parameters have already been set.

Source code in src/contraqctor/contract/base.py

def bind_reader_params(self, params: _typing.TReaderParams) -> Self:
    """Bind reader parameters to the data stream.

    Args:
        params: Parameters to bind to the data stream's reader.

    Returns:
        Self: The data stream instance for method chaining.

    Raises:
        ValueError: If reader parameters have already been set.
    """
    if not _typing.is_unset(self._reader_params):
        raise ValueError("Reader parameters are already set. Cannot bind again.")
    self._reader_params = params
    return self

at

at(name: str) -> DataStream

Get a child data stream by name.

Parameters:

Name	Type	Description	Default
name	str	Name of the child data stream to retrieve.	required

Returns:

Name	Type	Description
DataStream	DataStream	The child data stream with the given name.

Raises:

Type	Description
NotImplementedError	If the data stream does not support child access.

Examples:

# Access stream in a collection
collection = data_collection.load()
temp_stream = collection.at("temperature")

# Or using dictionary-style syntax
humidity_stream = collection["humidity"]

Source code in src/contraqctor/contract/base.py

def at(self, name: str) -> "DataStream":
    """Get a child data stream by name.

    Args:
        name: Name of the child data stream to retrieve.

    Returns:
        DataStream: The child data stream with the given name.

    Raises:
        NotImplementedError: If the data stream does not support child access.

    Examples:
        ```python
        # Access stream in a collection
        collection = data_collection.load()
        temp_stream = collection.at("temperature")

        # Or using dictionary-style syntax
        humidity_stream = collection["humidity"]
        ```
    """
    raise NotImplementedError("This method is not implemented for DataStream.")

load

load() -> Self

Load data into the data stream.

Reads data from the source and stores it in the data stream.

Returns:

Name	Type	Description
Self	Self	The data stream instance for method chaining.

Examples:

from contraqctor.contract import csv

# Create and load a CSV stream
params = csv.CsvParams(path="data/measurements.csv")
csv_stream = csv.Csv("measurements", reader_params=params)
csv_stream.load()

# Access the data
df = csv_stream.data
print(f"Loaded {len(df)} rows")

Source code in src/contraqctor/contract/base.py

def load(self) -> Self:
    """Load data into the data stream.

    Reads data from the source and stores it in the data stream.

    Returns:
        Self: The data stream instance for method chaining.

    Examples:
        ```python
        from contraqctor.contract import csv

        # Create and load a CSV stream
        params = csv.CsvParams(path="data/measurements.csv")
        csv_stream = csv.Csv("measurements", reader_params=params)
        csv_stream.load()

        # Access the data
        df = csv_stream.data
        print(f"Loaded {len(df)} rows")
        ```
    """
    self._data = self.read()
    return self

load_all

load_all(
    strict: bool = False,
) -> list[tuple[DataStream, Exception], None, None]

Recursively load this data stream and all child streams.

Performs depth-first traversal to load all streams in the hierarchy.

Parameters:

Name	Type	Description	Default
strict	bool	If True, raises exceptions immediately; otherwise collects and returns them.	False

Returns:

Name	Type	Description
list	list[tuple[DataStream, Exception], None, None]	List of tuples containing streams and exceptions that occurred during loading.

Raises:

Type	Description
Exception	If strict is True and an exception occurs during loading.

Examples:

# Load all streams and handle errors
errors = collection.load_all(strict=False)

if errors:
    for stream, error in errors:
        print(f"Error loading {stream.name}: {error}")

Source code in src/contraqctor/contract/base.py

def load_all(self, strict: bool = False) -> list[tuple["DataStream", Exception], None, None]:
    """Recursively load this data stream and all child streams.

    Performs depth-first traversal to load all streams in the hierarchy.

    Args:
        strict: If True, raises exceptions immediately; otherwise collects and returns them.

    Returns:
        list: List of tuples containing streams and exceptions that occurred during loading.

    Raises:
        Exception: If strict is True and an exception occurs during loading.

    Examples:
        ```python
        # Load all streams and handle errors
        errors = collection.load_all(strict=False)

        if errors:
            for stream, error in errors:
                print(f"Error loading {stream.name}: {error}")
        ```
    """
    self.load()
    exceptions = []
    for stream in self:
        if stream is None:
            continue
        try:
            exceptions += stream.load_all(strict=strict)
        except Exception as e:
            if strict:
                raise e
            exceptions.append((stream, e))
    return exceptions