fsspeckit.common API Reference

common

Cross-cutting utilities for fsspeckit.

This package contains dependency-free utilities shared across all components: - Datetime parsing and manipulation utilities - Logging configuration and helpers - General purpose utility functions (parallelism, filesystem sync) - Partition column helpers - Path and security validation

Schema, polars, and type-conversion utilities live in fsspeckit.datasets because they require optional dependencies (pyarrow, numpy, polars).

Attributes

fsspeckit.common.VALID_COMPRESSION_CODECS module-attribute

VALID_COMPRESSION_CODECS = frozenset(
    {
        "snappy",
        "gzip",
        "lz4",
        "zstd",
        "brotli",
        "uncompressed",
        "none",
    }
)

Functions

fsspeckit.common.get_timestamp_column

get_timestamp_column(df: Any) -> str | list[str]

Get timestamp column names from a DataFrame or PyArrow Table.

Automatically detects and normalizes different DataFrame types to work with pandas DataFrames, Polars DataFrames/LazyFrames, and PyArrow Tables.

Parameters:

Name	Type	Description	Default
df	Any	A Polars DataFrame/LazyFrame, PyArrow Table, or pandas DataFrame. The function automatically converts pandas DataFrames and PyArrow Tables to Polars LazyFrames for consistent timestamp detection.	required

Returns:

Type	Description
str | list[str]	List of strings containing timestamp column names. Returns an empty list
str | list[str]	if no timestamp columns are found.

Examples:

>>> import pandas as pd
>>> import polars as pl
>>> import pyarrow as pa
>>>
>>> # Works with pandas DataFrames
>>> df_pd = pd.DataFrame({"ts": pd.date_range("2023-01-01", periods=3)})
>>> get_timestamp_column(df_pd)
['ts']
>>>
>>> # Works with Polars DataFrames
>>> df_pl = pl.DataFrame({"ts": [datetime(2023, 1, 1)]})
>>> get_timestamp_column(df_pl)
['ts']
>>>
>>> # Works with PyArrow Tables
>>> table = pa.table({"ts": pa.array([datetime(2023, 1, 1)])})
>>> get_timestamp_column(table)
['ts']

Source code in src/fsspeckit/common/datetime.py

def get_timestamp_column(df: Any) -> str | list[str]:
    """Get timestamp column names from a DataFrame or PyArrow Table.

    Automatically detects and normalizes different DataFrame types to work with
    pandas DataFrames, Polars DataFrames/LazyFrames, and PyArrow Tables.

    Args:
        df: A Polars DataFrame/LazyFrame, PyArrow Table, or pandas DataFrame.
            The function automatically converts pandas DataFrames and PyArrow Tables
            to Polars LazyFrames for consistent timestamp detection.

    Returns:
        List of strings containing timestamp column names. Returns an empty list
        if no timestamp columns are found.

    Examples:
        >>> import pandas as pd
        >>> import polars as pl
        >>> import pyarrow as pa
        >>>
        >>> # Works with pandas DataFrames
        >>> df_pd = pd.DataFrame({"ts": pd.date_range("2023-01-01", periods=3)})
        >>> get_timestamp_column(df_pd)
        ['ts']
        >>>
        >>> # Works with Polars DataFrames
        >>> df_pl = pl.DataFrame({"ts": [datetime(2023, 1, 1)]})
        >>> get_timestamp_column(df_pl)
        ['ts']
        >>>
        >>> # Works with PyArrow Tables
        >>> table = pa.table({"ts": pa.array([datetime(2023, 1, 1)])})
        >>> get_timestamp_column(table)
        ['ts']
    """
    from fsspeckit.common.optional import (
        _import_pandas,
        _import_polars,
        _import_pyarrow,
    )

    pl = _import_polars()
    pa = _import_pyarrow()
    pd = _import_pandas()

    # Import polars.selectors at runtime
    import polars.selectors as cs

    # Normalise supported input types to a Polars LazyFrame
    if isinstance(df, pa.Table):
        df = pl.from_arrow(df).lazy()
    elif isinstance(df, pd.DataFrame):
        df = pl.from_pandas(df).lazy()

    return df.select(cs.datetime() | cs.date()).collect_schema().names()

fsspeckit.common.get_timedelta_str

get_timedelta_str(
    timedelta_string: str, to: str = "polars"
) -> str

Convert timedelta strings between different formats.

Converts timedelta strings between Polars and DuckDB formats, with graceful fallback for unknown units. Never raises errors for unknown units - instead returns a reasonable string representation.

Parameters:

Name	Type	Description	Default
timedelta_string	str	Input timedelta string (e.g., "1h", "2d", "5invalid").	required
to	str	Target format - "polars" or "duckdb". Defaults to "polars".	'polars'

Returns:

Type	Description
str	String in the target format. For unknown units, returns "value unit"
str	format without raising errors.

Examples:

>>> # Valid Polars units
>>> get_timedelta_str("1h", to="polars")
'1h'
>>> get_timedelta_str("1d", to="polars")
'1d'
>>>
>>> # Convert to DuckDB format
>>> get_timedelta_str("1h", to="duckdb")
'1 hour'
>>> get_timedelta_str("1s", to="duckdb")
'1 second'
>>>
>>> # Unknown units - graceful fallback
>>> get_timedelta_str("1invalid", to="polars")
'1 invalid'
>>> get_timedelta_str("5unknown", to="duckdb")
'5 unknown'

Source code in src/fsspeckit/common/datetime.py

def get_timedelta_str(timedelta_string: str, to: str = "polars") -> str:
    """Convert timedelta strings between different formats.

    Converts timedelta strings between Polars and DuckDB formats, with graceful
    fallback for unknown units. Never raises errors for unknown units - instead
    returns a reasonable string representation.

    Args:
        timedelta_string: Input timedelta string (e.g., "1h", "2d", "5invalid").
        to: Target format - "polars" or "duckdb". Defaults to "polars".

    Returns:
        String in the target format. For unknown units, returns "value unit"
        format without raising errors.

    Examples:
        >>> # Valid Polars units
        >>> get_timedelta_str("1h", to="polars")
        '1h'
        >>> get_timedelta_str("1d", to="polars")
        '1d'
        >>>
        >>> # Convert to DuckDB format
        >>> get_timedelta_str("1h", to="duckdb")
        '1 hour'
        >>> get_timedelta_str("1s", to="duckdb")
        '1 second'
        >>>
        >>> # Unknown units - graceful fallback
        >>> get_timedelta_str("1invalid", to="polars")
        '1 invalid'
        >>> get_timedelta_str("5unknown", to="duckdb")
        '5 unknown'
    """
    polars_timedelta_units = [
        "ns",
        "us",
        "ms",
        "s",
        "m",
        "h",
        "d",
        "w",
        "mo",
        "y",
    ]
    duckdb_timedelta_units = [
        "nanosecond",
        "microsecond",
        "millisecond",
        "second",
        "minute",
        "hour",
        "day",
        "week",
        "month",
        "year",
    ]

    unit = re.sub("[0-9]", "", timedelta_string).strip()
    val = timedelta_string.replace(unit, "").strip()
    base_unit = re.sub("s$", "", unit)

    if to == "polars":
        # Already a valid Polars unit
        if unit in polars_timedelta_units:
            return timedelta_string
        # Try to map known DuckDB units to Polars units
        mapping = dict(zip(duckdb_timedelta_units, polars_timedelta_units))
        target = mapping.get(base_unit)
        if target is not None:
            return f"{val}{target}"
        # Fallback for unknown units: preserve value and unit as-is
        return f"{val} {unit}".strip()

    # DuckDB branch
    if unit in polars_timedelta_units:
        mapping = dict(zip(polars_timedelta_units, duckdb_timedelta_units))
        return f"{val} {mapping[unit]}"

    # Unknown Polars-style unit when targeting DuckDB: keep unit without trailing "s"
    return f"{val} {base_unit}".strip()

fsspeckit.common.timestamp_from_string cached

timestamp_from_string(
    timestamp_str: str,
    tz: str | None = None,
    naive: bool = False,
) -> datetime | date | time

Converts a timestamp string (ISO 8601 format) into a datetime, date, or time object using only standard Python libraries.

Handles strings with or without timezone information (e.g., '2023-01-01T10:00:00+02:00', '2023-01-01', '10:00:00'). Supports timezone offsets like '+HH:MM' or '+HHMM'. For named timezones (e.g., 'Europe/Paris'), requires Python 3.9+ and the 'tzdata' package to be installed.

Parameters:

Name	Type	Description	Default
timestamp_str	str	The string representation of the timestamp (ISO 8601 format).	required
tz	str	Target timezone identifier (e.g., 'UTC', '+02:00', 'Europe/Paris'). If provided, the output datetime/time will be localized or converted to this timezone. Defaults to None.	None
naive	bool	If True, return a naive datetime/time (no timezone info), even if the input string or tz parameter specifies one. Defaults to False.	False

Returns:

Type	Description
datetime | date | time	Union[dt.datetime, dt.date, dt.time]: The parsed datetime, date, or time object.

Raises:

Type	Description
ValueError	If the timestamp string format is invalid or the timezone is invalid/unsupported.

Source code in src/fsspeckit/common/datetime.py

@lru_cache(maxsize=128)
def timestamp_from_string(
    timestamp_str: str,
    tz: str | None = None,
    naive: bool = False,
) -> dt.datetime | dt.date | dt.time:
    """
    Converts a timestamp string (ISO 8601 format) into a datetime, date, or time object
    using only standard Python libraries.

    Handles strings with or without timezone information (e.g., '2023-01-01T10:00:00+02:00',
    '2023-01-01', '10:00:00'). Supports timezone offsets like '+HH:MM' or '+HHMM'.
    For named timezones (e.g., 'Europe/Paris'), requires Python 3.9+ and the 'tzdata'
    package to be installed.

    Args:
        timestamp_str (str): The string representation of the timestamp (ISO 8601 format).
        tz (str, optional): Target timezone identifier (e.g., 'UTC', '+02:00', 'Europe/Paris').
            If provided, the output datetime/time will be localized or converted to this timezone.
            Defaults to None.
        naive (bool, optional): If True, return a naive datetime/time (no timezone info),
            even if the input string or `tz` parameter specifies one. Defaults to False.

    Returns:
        Union[dt.datetime, dt.date, dt.time]: The parsed datetime, date, or time object.

    Raises:
        ValueError: If the timestamp string format is invalid or the timezone is
                    invalid/unsupported.
    """

    # Regex to parse timezone offsets like +HH:MM or +HHMM
    _TZ_OFFSET_REGEX = re.compile(r"([+-])(\d{2}):?(\d{2})")

    def _parse_tz_offset(tz_str: str) -> dt.tzinfo | None:
        """Parses a timezone offset string into a timezone object."""
        match = _TZ_OFFSET_REGEX.fullmatch(tz_str)
        if match:
            sign, hours, minutes = match.groups()
            offset_seconds = (int(hours) * 3600 + int(minutes) * 60) * (
                -1 if sign == "-" else 1
            )
            if abs(offset_seconds) >= 24 * 3600:
                raise ValueError(f"Invalid timezone offset: {tz_str}")
            return dt.timezone(dt.timedelta(seconds=offset_seconds), name=tz_str)
        return None

    def _get_tzinfo(tz_identifier: str | None) -> dt.tzinfo | None:
        """Gets a tzinfo object from a string (offset or IANA name)."""
        if tz_identifier is None:
            return None
        if tz_identifier.upper() == "UTC":
            return dt.timezone.utc

        # Try parsing as offset first
        offset_tz = _parse_tz_offset(tz_identifier)
        if offset_tz:
            return offset_tz

        # Try parsing as IANA name using zoneinfo (if available)
        if ZoneInfo:
            try:
                return ZoneInfo(tz_identifier)
            except ZoneInfoNotFoundError:
                raise ValueError(
                    f"Timezone '{tz_identifier}' not found. Install 'tzdata' or use offset format."
                )
            except Exception as e:  # Catch other potential zoneinfo errors
                raise ValueError(f"Error loading timezone '{tz_identifier}': {e}")
        else:
            # zoneinfo not available
            raise ValueError(
                f"Invalid timezone: '{tz_identifier}'. Use offset format (e.g., '+02:00') "
                "or run Python 3.9+ with 'tzdata' installed for named timezones."
            )

    target_tz: dt.tzinfo | None = _get_tzinfo(tz)
    parsed_obj: dt.datetime | dt.date | dt.time | None = None

    # Preprocess: Replace space separator, strip whitespace
    processed_str = timestamp_str.strip().replace(" ", "T")

    # Attempt parsing (datetime, date, time) using fromisoformat
    try:
        # Python < 3.11 fromisoformat has limitations (e.g., no Z, no +HHMM offset)
        # This implementation assumes Python 3.11+ for full ISO 8601 support via fromisoformat
        # or that input strings use formats compatible with older versions (e.g., +HH:MM)
        parsed_obj = dt.datetime.fromisoformat(processed_str)
    except ValueError:
        try:
            parsed_obj = dt.date.fromisoformat(processed_str)
        except ValueError:
            try:
                # Time parsing needs care, especially with offsets in older Python
                parsed_obj = dt.time.fromisoformat(processed_str)
            except ValueError:
                # Add fallback for simple HH:MM:SS if needed, though less robust
                # try:
                #     parsed_obj = dt.datetime.strptime(processed_str, "%H:%M:%S").time()
                # except ValueError:
                raise ValueError(f"Invalid timestamp format: '{timestamp_str}'")

    # Apply timezone logic if we have a datetime or time object
    if isinstance(parsed_obj, (dt.datetime, dt.time)):
        is_aware = (
            parsed_obj.tzinfo is not None
            and parsed_obj.tzinfo.utcoffset(
                parsed_obj if isinstance(parsed_obj, dt.datetime) else None
            )
            is not None
        )

        if target_tz:
            if is_aware:
                # Convert existing aware object to target timezone (only for datetime)
                if isinstance(parsed_obj, dt.datetime):
                    parsed_obj = parsed_obj.astimezone(target_tz)
                # else: dt.time cannot be converted without a date context. Keep original tz.
            else:
                # Localize naive object to target timezone
                parsed_obj = parsed_obj.replace(tzinfo=target_tz)
            is_aware = True  # Object is now considered aware

        # Handle naive flag: remove tzinfo if requested
        if naive and is_aware:
            parsed_obj = parsed_obj.replace(tzinfo=None)

    # If it's a date object, tz/naive flags are ignored
    elif isinstance(parsed_obj, dt.date):
        pass

    return parsed_obj

fsspeckit.common.get_logger

get_logger(name: str = 'fsspeckit') -> logger

Get a logger instance for the given name.

Parameters:

Name	Type	Description	Default
name	str	Logger name, typically the module name.	'fsspeckit'

Returns:

Type	Description
logger	Configured logger instance.

Example

logger = get_logger(__name__)
logger.info("This is a log message")

Source code in src/fsspeckit/common/logging/config.py

def get_logger(name: str = "fsspeckit") -> "logger":
    """Get a logger instance for the given name.

    Args:
        name: Logger name, typically the module name.

    Returns:
        Configured logger instance.

    Example:
        ```python
        logger = get_logger(__name__)
        logger.info("This is a log message")
        ```
    """
    return logger.bind(name=name)

fsspeckit.common.setup_logging

setup_logging(
    level: str | None = None,
    disable: bool = False,
    format_string: str | None = None,
) -> None

Configure the Loguru logger for fsspeckit.

Removes the default handler and adds a new one targeting stderr with customizable level and format.

Parameters:

Name	Type	Description	Default
level	str | None	Log level (DEBUG, INFO, WARNING, ERROR, CRITICAL). If None, uses fsspeckit_LOG_LEVEL environment variable or defaults to "INFO".	None
disable	bool	Whether to disable logging for fsspeckit package.	False
format_string	str | None	Custom format string for log messages. If None, uses a default comprehensive format.	None

Example

# Basic setup
setup_logging()

# Custom level and format
setup_logging(level="DEBUG", format_string="{time} | {level} | {message}")

# Disable logging
setup_logging(disable=True)

Source code in src/fsspeckit/common/logging/config.py

def setup_logging(
    level: str | None = None,
    disable: bool = False,
    format_string: str | None = None,
) -> None:
    """Configure the Loguru logger for fsspeckit.

    Removes the default handler and adds a new one targeting stderr
    with customizable level and format.

    Args:
        level: Log level (DEBUG, INFO, WARNING, ERROR, CRITICAL).
               If None, uses fsspeckit_LOG_LEVEL environment variable
               or defaults to "INFO".
        disable: Whether to disable logging for fsspeckit package.
        format_string: Custom format string for log messages.
                      If None, uses a default comprehensive format.

    Example:
        ```python
        # Basic setup
        setup_logging()

        # Custom level and format
        setup_logging(level="DEBUG", format_string="{time} | {level} | {message}")

        # Disable logging
        setup_logging(disable=True)
        ```
    """
    # Determine log level
    if level is None:
        level = os.getenv("fsspeckit_LOG_LEVEL", "INFO")

    # Default format if none provided
    if format_string is None:
        format_string = (
            "<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | "
            "<level>{level: <8}</level> | "
            "<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - "
            "<level>{message}</level>"
        )

    # Remove the default handler added by Loguru
    logger.remove()

    # Add new handler with custom configuration
    logger.add(
        sys.stderr,
        level=level.upper(),
        format=format_string,
    )

    # Optionally disable logging for this package
    if disable:
        logger.disable("fsspeckit")

fsspeckit.common.get_partitions_from_path

get_partitions_from_path(
    path: str, partitioning: str | list[str] | None = None
) -> list[tuple]

Extract dataset partitions from a file path.

Parses file paths to extract partition information based on different partitioning schemes. This is the canonical implementation used across all fsspeckit backends.

Parameters:

Name	Type	Description	Default
path	str	File path potentially containing partition information.	required
partitioning	str | list[str] | None	Partitioning scheme: - "hive": Hive-style partitioning (key=value) - str: Single partition column name - list[str]: Multiple partition column names - None: Return empty list	None

Returns:

Type	Description
list[tuple]	List of tuples containing (column, value) pairs.

Examples:

>>> # Hive-style partitioning
>>> get_partitions_from_path("data/year=2023/month=01/file.parquet", "hive")
[('year', '2023'), ('month', '01')]

>>> # Single partition column
>>> get_partitions_from_path("data/2023/01/file.parquet", "year")
[('year', '2023')]

>>> # Multiple partition columns
>>> get_partitions_from_path("data/2023/01/file.parquet", ["year", "month"])
[('year', '2023'), ('month', '01')]

>>> # No partitioning
>>> get_partitions_from_path("data/file.parquet", None)
[]

Source code in src/fsspeckit/common/partitions.py

def get_partitions_from_path(
    path: str, partitioning: str | list[str] | None = None
) -> list[tuple]:
    """
    Extract dataset partitions from a file path.

    Parses file paths to extract partition information based on
    different partitioning schemes. This is the canonical implementation
    used across all fsspeckit backends.

    Args:
        path: File path potentially containing partition information.
        partitioning: Partitioning scheme:
            - "hive": Hive-style partitioning (key=value)
            - str: Single partition column name
            - list[str]: Multiple partition column names
            - None: Return empty list

    Returns:
        List of tuples containing (column, value) pairs.

    Examples:
        >>> # Hive-style partitioning
        >>> get_partitions_from_path("data/year=2023/month=01/file.parquet", "hive")
        [('year', '2023'), ('month', '01')]

        >>> # Single partition column
        >>> get_partitions_from_path("data/2023/01/file.parquet", "year")
        [('year', '2023')]

        >>> # Multiple partition columns
        >>> get_partitions_from_path("data/2023/01/file.parquet", ["year", "month"])
        [('year', '2023'), ('month', '01')]

        >>> # No partitioning
        >>> get_partitions_from_path("data/file.parquet", None)
        []
    """
    if "." in path:
        path = os.path.dirname(path)

    parts = path.split("/")

    if isinstance(partitioning, str):
        if partitioning == "hive":
            return [tuple(p.split("=")) for p in parts if "=" in p]
        else:
            # Single partition column - take the last directory that looks like a value
            # This is a simple heuristic for cases like data/2023/file.parquet
            if parts and parts[-1]:
                return [(partitioning, parts[-1])]
            return []
    elif isinstance(partitioning, list):
        # Multiple partition columns - map column names to path parts from right to left
        if not parts:
            return []

        # Take the last N parts where N is the number of partition columns
        partition_parts = (
            parts[-len(partitioning) :] if len(parts) >= len(partitioning) else parts
        )
        return list(zip(partitioning, partition_parts))
    else:
        return []

fsspeckit.common.normalize_partition_value

normalize_partition_value(value: str) -> str

Normalize a partition value for consistent comparison.

Parameters:

Name	Type	Description	Default
value	str	Raw partition value from path.	required

Returns:

Type	Description
str	Normalized partition value.

Source code in src/fsspeckit/common/partitions.py

def normalize_partition_value(value: str) -> str:
    """
    Normalize a partition value for consistent comparison.

    Args:
        value: Raw partition value from path.

    Returns:
        Normalized partition value.
    """
    if value is None:
        return None
    return value.strip().strip("\"'").replace("\\", "")

fsspeckit.common.validate_partition_columns

validate_partition_columns(
    partitions: list[tuple[str, str]],
    expected_columns: list[str] | None = None,
) -> bool

Validate partition columns against expected schema.

Parameters:

Name	Type	Description	Default
partitions	list[tuple[str, str]]	List of (column, value) tuples.	required
expected_columns	list[str] | None	Optional list of expected column names.	None

Returns:

Type	Description
bool	True if partitions are valid, False otherwise.

Source code in src/fsspeckit/common/partitions.py

def validate_partition_columns(
    partitions: list[tuple[str, str]], expected_columns: list[str] | None = None
) -> bool:
    """
    Validate partition columns against expected schema.

    Args:
        partitions: List of (column, value) tuples.
        expected_columns: Optional list of expected column names.

    Returns:
        True if partitions are valid, False otherwise.
    """
    if not partitions:
        return True

    if expected_columns is not None:
        partition_columns = {col for col, _ in partitions}
        expected_set = set(expected_columns)

        # Check if all partition columns are expected
        if not partition_columns.issubset(expected_set):
            return False

        # Check if all expected columns are present (strict validation)
        if not expected_set.issubset(partition_columns):
            return False

    # Validate that no column names are empty
    for col, val in partitions:
        if not col or not col.strip():
            return False

    return True

fsspeckit.common.build_partition_path

build_partition_path(
    base_path: str,
    partitions: list[tuple[str, str]],
    partitioning: str = "hive",
) -> str

Build a file path with partition directories.

Parameters:

Name	Type	Description	Default
base_path	str	Base directory path.	required
partitions	list[tuple[str, str]]	List of (column, value) tuples.	required
partitioning	str	Partitioning scheme ("hive" or "directory").	'hive'

Returns:

Type	Description
str	Path string with partition directories.

Source code in src/fsspeckit/common/partitions.py

def build_partition_path(
    base_path: str, partitions: list[tuple[str, str]], partitioning: str = "hive"
) -> str:
    """
    Build a file path with partition directories.

    Args:
        base_path: Base directory path.
        partitions: List of (column, value) tuples.
        partitioning: Partitioning scheme ("hive" or "directory").

    Returns:
        Path string with partition directories.
    """
    if not partitions:
        return base_path

    if partitioning == "hive":
        # Hive-style: column=value/column=value
        partition_dirs = [f"{col}={val}" for col, val in partitions]
    else:
        # Directory-style: value/value (order matters)
        partition_dirs = [val for _, val in partitions]

    return "/".join([base_path.rstrip("/")] + partition_dirs)

fsspeckit.common.extract_partition_filters

extract_partition_filters(
    paths: list[str],
    partitioning: str | list[str] | None = None,
) -> dict[str, set[str]]

Extract unique partition values from a list of paths.

Parameters:

Name	Type	Description	Default
paths	list[str]	List of file paths.	required
partitioning	str | list[str] | None	Partitioning scheme.	None

Returns:

Type	Description
dict[str, set[str]]	Dictionary mapping column names to sets of unique values.

Source code in src/fsspeckit/common/partitions.py

def extract_partition_filters(
    paths: list[str], partitioning: str | list[str] | None = None
) -> dict[str, set[str]]:
    """
    Extract unique partition values from a list of paths.

    Args:
        paths: List of file paths.
        partitioning: Partitioning scheme.

    Returns:
        Dictionary mapping column names to sets of unique values.
    """
    partition_values = {}

    for path in paths:
        partitions = get_partitions_from_path(path, partitioning)
        for col, val in partitions:
            if col not in partition_values:
                partition_values[col] = set()
            partition_values[col].add(val)

    return partition_values

fsspeckit.common.filter_paths_by_partitions

filter_paths_by_partitions(
    paths: list[str],
    partition_filters: dict[str, str | list[str]],
    partitioning: str | list[str] | None = None,
) -> list[str]

Filter paths based on partition values.

Parameters:

Name	Type	Description	Default
paths	list[str]	List of file paths to filter.	required
partition_filters	dict[str, str | list[str]]	Dictionary mapping column names to filter values.	required
partitioning	str | list[str] | None	Partitioning scheme.	None

Returns:

Type	Description
list[str]	Filtered list of paths.

Source code in src/fsspeckit/common/partitions.py

def filter_paths_by_partitions(
    paths: list[str],
    partition_filters: dict[str, str | list[str]],
    partitioning: str | list[str] | None = None,
) -> list[str]:
    """
    Filter paths based on partition values.

    Args:
        paths: List of file paths to filter.
        partition_filters: Dictionary mapping column names to filter values.
        partitioning: Partitioning scheme.

    Returns:
        Filtered list of paths.
    """
    filtered_paths = []

    for path in paths:
        partitions = dict(get_partitions_from_path(path, partitioning))

        # Check if path matches all filters
        matches = True
        for col, filter_val in partition_filters.items():
            if col not in partitions:
                matches = False
                break

            path_val = partitions[col]

            # Handle list of allowed values
            if isinstance(filter_val, list):
                if path_val not in filter_val:
                    matches = False
                    break
            else:
                # Single value comparison
                if path_val != filter_val:
                    matches = False
                    break

        if matches:
            filtered_paths.append(path)

    return filtered_paths

fsspeckit.common.infer_partitioning_scheme

infer_partitioning_scheme(
    paths: list[str], max_samples: int = 100
) -> dict[str, Any]

Infer the partitioning scheme from a sample of paths.

Parameters:

Name	Type	Description	Default
paths	list[str]	List of file paths to analyze.	required
max_samples	int	Maximum number of paths to sample.	100

Returns:

Type	Description
dict[str, Any]	Dictionary with inferred scheme information.

Source code in src/fsspeckit/common/partitions.py

def infer_partitioning_scheme(
    paths: list[str], max_samples: int = 100
) -> dict[str, Any]:
    """
    Infer the partitioning scheme from a sample of paths.

    Args:
        paths: List of file paths to analyze.
        max_samples: Maximum number of paths to sample.

    Returns:
        Dictionary with inferred scheme information.
    """
    if not paths:
        return {"scheme": None, "confidence": 0.0}

    # Sample paths for analysis
    sample_paths = paths[:max_samples] if len(paths) > max_samples else paths

    # Check for Hive-style partitioning
    hive_partitions = []
    directory_partitions = []

    for path in sample_paths:
        # Remove filename and get directory parts
        dir_path = os.path.dirname(path)
        parts = dir_path.split("/")

        # Look for key=value patterns
        hive_parts = [p for p in parts if "=" in p and p.split("=")[0].strip()]
        if hive_parts:
            hive_partitions.append(len(hive_parts))

        # Look for directory-style partitions (numeric dates, etc.)
        dir_parts = [
            p
            for p in parts
            if p.replace("/", "").replace("-", "").replace("_", "").isdigit()
        ]
        if dir_parts:
            directory_partitions.append(len(dir_parts))

    # Determine most likely scheme
    result = {"scheme": None, "confidence": 0.0}

    if hive_partitions:
        avg_hive_parts = sum(hive_partitions) / len(hive_partitions)
        if avg_hive_parts >= 1:  # At least 1 partition level on average
            result["scheme"] = "hive"
            result["confidence"] = min(
                1.0, avg_hive_parts / 3.0
            )  # Normalize by expected max
            result["avg_partitions"] = avg_hive_parts
            return result

    if directory_partitions:
        avg_dir_parts = sum(directory_partitions) / len(directory_partitions)
        if avg_dir_parts >= 1:  # At least 1 partition level on average
            result["scheme"] = "directory"
            result["confidence"] = min(
                1.0, avg_dir_parts / 3.0
            )  # Normalize by expected max
            result["avg_partitions"] = avg_dir_parts
            return result

    # No clear partitioning detected
    return result

fsspeckit.common.get_partition_columns_from_paths

get_partition_columns_from_paths(
    paths: list[str],
    partitioning: str | list[str] | None = None,
) -> list[str]

Get all unique partition column names from a list of paths.

Parameters:

Name	Type	Description	Default
paths	list[str]	List of file paths.	required
partitioning	str | list[str] | None	Partitioning scheme.	None

Returns:

Type	Description
list[str]	List of unique partition column names.

Source code in src/fsspeckit/common/partitions.py

def get_partition_columns_from_paths(
    paths: list[str], partitioning: str | list[str] | None = None
) -> list[str]:
    """
    Get all unique partition column names from a list of paths.

    Args:
        paths: List of file paths.
        partitioning: Partitioning scheme.

    Returns:
        List of unique partition column names.
    """
    columns = set()

    for path in paths:
        partitions = get_partitions_from_path(path, partitioning)
        for col, _ in partitions:
            columns.add(col)

    return sorted(list(columns))

fsspeckit.common.create_partition_expression

create_partition_expression(
    partitions: list[tuple[str, str]],
    backend: str = "pyarrow",
) -> Any

Create a partition filter expression for different backends.

Parameters:

Name	Type	Description	Default
partitions	list[tuple[str, str]]	List of (column, value) tuples.	required
backend	str	Target backend ("pyarrow", "duckdb").	'pyarrow'

Returns:

Type	Description
Any	Backend-specific filter expression.

Source code in src/fsspeckit/common/partitions.py

def create_partition_expression(
    partitions: list[tuple[str, str]], backend: str = "pyarrow"
) -> Any:
    """
    Create a partition filter expression for different backends.

    Args:
        partitions: List of (column, value) tuples.
        backend: Target backend ("pyarrow", "duckdb").

    Returns:
        Backend-specific filter expression.
    """
    if not partitions:
        return None

    if backend == "pyarrow":
        import pyarrow.dataset as ds

        # Build PyArrow dataset filter expression
        expressions = []
        for col, val in partitions:
            expressions.append(ds.field(col) == val)

        # Combine with AND logic
        result = expressions[0]
        for expr in expressions[1:]:
            result = result & expr
        return result

    elif backend == "duckdb":
        # Build DuckDB WHERE clause
        conditions = []
        for col, val in partitions:
            if isinstance(val, str):
                conditions.append(f"\"{col}\" = '{val}'")
            else:
                conditions.append(f'"{col}" = {val}')

        return " AND ".join(conditions)

    else:
        raise ValueError(f"Unsupported backend: {backend}")

fsspeckit.common.apply_partition_pruning

apply_partition_pruning(
    paths: list[str],
    partition_filters: dict[str, Any],
    partitioning: str | list[str] | None = None,
) -> list[str]

Apply partition pruning to reduce the set of files to scan.

This is an optimization that eliminates files based on partition values before any I/O operations.

Parameters:

Name	Type	Description	Default
paths	list[str]	List of all file paths.	required
partition_filters	dict[str, Any]	Dictionary of partition filters to apply.	required
partitioning	str | list[str] | None	Partitioning scheme.	None

Returns:

Type	Description
list[str]	Pruned list of paths.

Source code in src/fsspeckit/common/partitions.py

def apply_partition_pruning(
    paths: list[str],
    partition_filters: dict[str, Any],
    partitioning: str | list[str] | None = None,
) -> list[str]:
    """
    Apply partition pruning to reduce the set of files to scan.

    This is an optimization that eliminates files based on partition
    values before any I/O operations.

    Args:
        paths: List of all file paths.
        partition_filters: Dictionary of partition filters to apply.
        partitioning: Partitioning scheme.

    Returns:
        Pruned list of paths.
    """
    if not partition_filters:
        return paths

    return filter_paths_by_partitions(paths, partition_filters, partitioning)

fsspeckit.common.run_parallel

run_parallel(
    func: Callable,
    *args: Any,
    n_jobs: int = -1,
    backend: str = "threading",
    verbose: bool = True,
    **kwargs: Any,
) -> list[Any]

Runs a function for a list of parameters in parallel.

Requires: fsspeckit[datasets] extra for joblib dependency.

Parameters:

Name	Type	Description	Default
func	Callable	function to run in parallel	required
*args	Any	Positional arguments. Can be single values or any non-string iterables (including generators)	()
n_jobs	int	Number of joblib workers. Defaults to -1	-1
backend	str	joblib backend. Valid options are loky,threading,multiprocessing or sequential. Defaults to "threading"	'threading'
verbose	bool	Show progress bar. Defaults to True	True
**kwargs	Any	Keyword arguments. Can be single values or any non-string iterables (including generators)	{}

Returns:

Type	Description
list[Any]	list[any]: Function output

Raises:

Type	Description
ImportError	If joblib is not available. Install with: pip install fsspeckit[datasets]
ValueError	If no iterable arguments are provided or iterables have different lengths

Examples:

>>> # Single iterable argument
>>> run_parallel(func, [1,2,3], fixed_arg=42)

>>> # Multiple iterables in args and kwargs
>>> run_parallel(func, [1,2,3], val=[7,8,9], fixed=42)

>>> # Only kwargs iterables
>>> run_parallel(func, x=[1,2,3], y=[4,5,6], fixed=42)

>>> # Generator support
>>> def gen():
...     yield from [1, 2, 3]
>>> run_parallel(str, gen())  # Returns ['1', '2', '3']

Source code in src/fsspeckit/common/parallel.py

def run_parallel(
    func: Callable,
    *args: Any,
    n_jobs: int = -1,
    backend: str = "threading",
    verbose: bool = True,
    **kwargs: Any,
) -> list[Any]:
    """Runs a function for a list of parameters in parallel.

    Requires: fsspeckit[datasets] extra for joblib dependency.

    Args:
        func (Callable): function to run in parallel
        *args: Positional arguments. Can be single values or any non-string iterables (including generators)
        n_jobs (int, optional): Number of joblib workers. Defaults to -1
        backend (str, optional): joblib backend. Valid options are
            `loky`,`threading`,`multiprocessing` or `sequential`. Defaults to "threading"
        verbose (bool, optional): Show progress bar. Defaults to True
        **kwargs: Keyword arguments. Can be single values or any non-string iterables (including generators)

    Returns:
        list[any]: Function output

    Raises:
        ImportError: If joblib is not available. Install with: pip install fsspeckit[datasets]
        ValueError: If no iterable arguments are provided or iterables have different lengths

    Examples:
        >>> # Single iterable argument
        >>> run_parallel(func, [1,2,3], fixed_arg=42)

        >>> # Multiple iterables in args and kwargs
        >>> run_parallel(func, [1,2,3], val=[7,8,9], fixed=42)

        >>> # Only kwargs iterables
        >>> run_parallel(func, x=[1,2,3], y=[4,5,6], fixed=42)

        >>> # Generator support
        >>> def gen():
        ...     yield from [1, 2, 3]
        >>> run_parallel(str, gen())  # Returns ['1', '2', '3']
    """
    if backend == "threading" and n_jobs == -1:
        n_jobs = min(256, (os.cpu_count() or 1) + 4)

    parallel_kwargs = {"n_jobs": n_jobs, "backend": backend, "verbose": 0}

    # Prepare and validate arguments
    iterables, fixed_args, iterable_kwargs, fixed_kwargs, first_iterable_len = (
        _prepare_parallel_args(args, kwargs)
    )

    # Create parameter combinations
    all_iterables = iterables + list(iterable_kwargs.values())

    # Handle empty iterables case
    if first_iterable_len == 0:
        return []

    param_combinations = list(zip(*all_iterables))

    # Execute with or without progress tracking
    if not verbose:
        return _execute_parallel_without_progress(
            func,
            iterables,
            fixed_args,
            iterable_kwargs,
            fixed_kwargs,
            param_combinations,
            parallel_kwargs,
        )
    else:
        return _execute_parallel_with_progress(
            func,
            iterables,
            fixed_args,
            iterable_kwargs,
            fixed_kwargs,
            param_combinations,
            parallel_kwargs,
        )

fsspeckit.common.sync_dir

sync_dir(
    src_fs: AbstractFileSystem,
    dst_fs: AbstractFileSystem,
    src_path: str = "",
    dst_path: str = "",
    server_side: bool = True,
    chunk_size: int = 8 * 1024 * 1024,
    parallel: bool = False,
    n_jobs: int = -1,
    verbose: bool = True,
) -> dict[str, list[str]]

Sync two directories between different filesystems.

Compares files in the source and destination directories, copies new or updated files from source to destination, and deletes stale files from destination.

Parameters:

Name	Type	Description	Default
src_fs	AbstractFileSystem	Source filesystem (fsspec AbstractFileSystem)	required
dst_fs	AbstractFileSystem	Destination filesystem (fsspec AbstractFileSystem)	required
src_path	str	Path in source filesystem to sync. Default is root ('').	''
dst_path	str	Path in destination filesystem to sync. Default is root ('').	''
server_side	bool	Whether to use server-side copy if supported. Default is True.	True
chunk_size	int	Size of chunks to read/write files (in bytes). Default is 8MB.	8 * 1024 * 1024
parallel	bool	Whether to perform copy/delete operations in parallel. Default is False.	False
n_jobs	int	Number of parallel jobs if parallel=True. Default is -1 (all cores).	-1
verbose	bool	Whether to show progress bars. Default is True.	True

Returns:

Name	Type	Description
dict	dict[str, list[str]]	Summary of added and deleted files

Source code in src/fsspeckit/common/sync.py

def sync_dir(
    src_fs: AbstractFileSystem,
    dst_fs: AbstractFileSystem,
    src_path: str = "",
    dst_path: str = "",
    server_side: bool = True,
    chunk_size: int = 8 * 1024 * 1024,
    parallel: bool = False,
    n_jobs: int = -1,
    verbose: bool = True,
) -> dict[str, list[str]]:
    """Sync two directories between different filesystems.

    Compares files in the source and destination directories, copies new or updated files from source to destination,
    and deletes stale files from destination.

    Args:
        src_fs: Source filesystem (fsspec AbstractFileSystem)
        dst_fs: Destination filesystem (fsspec AbstractFileSystem)
        src_path: Path in source filesystem to sync. Default is root ('').
        dst_path: Path in destination filesystem to sync. Default is root ('').
        server_side: Whether to use server-side copy if supported. Default is True.
        chunk_size: Size of chunks to read/write files (in bytes). Default is 8MB.
        parallel: Whether to perform copy/delete operations in parallel. Default is False.
        n_jobs: Number of parallel jobs if parallel=True. Default is -1 (all cores).
        verbose: Whether to show progress bars. Default is True.

    Returns:
        dict: Summary of added and deleted files
    """

    src_mapper = src_fs.get_mapper(src_path)
    dst_mapper = dst_fs.get_mapper(dst_path)

    add_files = sorted(src_mapper.keys() - dst_mapper.keys())
    delete_files = sorted(dst_mapper.keys() - src_mapper.keys())

    return sync_files(
        add_files=add_files,
        delete_files=delete_files,
        src_fs=src_fs,
        dst_fs=dst_fs,
        src_path=src_path,
        dst_path=dst_path,
        chunk_size=chunk_size,
        server_side=server_side,
        parallel=parallel,
        n_jobs=n_jobs,
        verbose=verbose,
    )

fsspeckit.common.sync_files

sync_files(
    add_files: list[str],
    delete_files: list[str],
    src_fs: AbstractFileSystem,
    dst_fs: AbstractFileSystem,
    src_path: str = "",
    dst_path: str = "",
    server_side: bool = False,
    chunk_size: int = 8 * 1024 * 1024,
    parallel: bool = False,
    n_jobs: int = -1,
    verbose: bool = True,
) -> dict[str, list[str]]

Sync files between two filesystems by copying new files and deleting old ones.

Parameters:

Name	Type	Description	Default
add_files	list[str]	List of file paths to add (copy from source to destination)	required
delete_files	list[str]	List of file paths to delete from destination	required
src_fs	AbstractFileSystem	Source filesystem (fsspec AbstractFileSystem)	required
dst_fs	AbstractFileSystem	Destination filesystem (fsspec AbstractFileSystem)	required
src_path	str	Base path in source filesystem. Default is root ('').	''
dst_path	str	Base path in destination filesystem. Default is root ('').	''
server_side	bool	Whether to use server-side copy if supported. Default is False.	False
chunk_size	int	Size of chunks to read/write files (in bytes). Default is 8MB.	8 * 1024 * 1024
parallel	bool	Whether to perform copy/delete operations in parallel. Default is False.	False
n_jobs	int	Number of parallel jobs if parallel=True. Default is -1 (all cores).	-1
verbose	bool	Whether to show progress bars. Default is True.	True

Returns:

Name	Type	Description
dict	dict[str, list[str]]	Summary of added and deleted files

Source code in src/fsspeckit/common/sync.py

def sync_files(
    add_files: list[str],
    delete_files: list[str],
    src_fs: AbstractFileSystem,
    dst_fs: AbstractFileSystem,
    src_path: str = "",
    dst_path: str = "",
    server_side: bool = False,
    chunk_size: int = 8 * 1024 * 1024,
    parallel: bool = False,
    n_jobs: int = -1,
    verbose: bool = True,
) -> dict[str, list[str]]:
    """Sync files between two filesystems by copying new files and deleting old ones.

    Args:
        add_files: List of file paths to add (copy from source to destination)
        delete_files: List of file paths to delete from destination
        src_fs: Source filesystem (fsspec AbstractFileSystem)
        dst_fs: Destination filesystem (fsspec AbstractFileSystem)
        src_path: Base path in source filesystem. Default is root ('').
        dst_path: Base path in destination filesystem. Default is root ('').
        server_side: Whether to use server-side copy if supported. Default is False.
        chunk_size: Size of chunks to read/write files (in bytes). Default is 8MB.
        parallel: Whether to perform copy/delete operations in parallel. Default is False.
        n_jobs: Number of parallel jobs if parallel=True. Default is -1 (all cores).
        verbose: Whether to show progress bars. Default is True.

    Returns:
        dict: Summary of added and deleted files
    """
    CHUNK = chunk_size
    RETRIES = 3

    server_side = check_fs_identical(src_fs, dst_fs) and server_side

    src_mapper = src_fs.get_mapper(src_path)
    dst_mapper = dst_fs.get_mapper(dst_path)

    if len(add_files):
        # Copy new files
        if parallel:
            if server_side:
                try:
                    run_parallel(
                        server_side_copy_file,
                        add_files,
                        src_mapper=src_mapper,
                        dst_mapper=dst_mapper,
                        RETRIES=RETRIES,
                        n_jobs=n_jobs,
                        verbose=verbose,
                    )
                except (RuntimeError, OSError) as e:
                    logger.warning(
                        "Server-side copy failed for some files, falling back to client-side: %s",
                        str(e),
                    )
                    # Fallback to client-side copy if server-side fails
                    run_parallel(
                        copy_file,
                        add_files,
                        src_fs=src_fs,
                        dst_fs=dst_fs,
                        src_path=src_path,
                        dst_path=dst_path,
                        CHUNK=CHUNK,
                        RETRIES=RETRIES,
                        n_jobs=n_jobs,
                        verbose=verbose,
                    )

            else:
                run_parallel(
                    copy_file,
                    add_files,
                    src_fs=src_fs,
                    dst_fs=dst_fs,
                    src_path=src_path,
                    dst_path=dst_path,
                    CHUNK=CHUNK,
                    RETRIES=RETRIES,
                    n_jobs=n_jobs,
                    verbose=verbose,
                )
        else:
            if verbose:
                from rich.progress import track

                for key in track(
                    add_files,
                    description="Copying new files...",
                    total=len(add_files),
                ):
                    if server_side:
                        try:
                            server_side_copy_file(
                                key, src_mapper, dst_mapper, RETRIES
                            )
                        except (RuntimeError, OSError):
                            copy_file(
                                key, src_fs, dst_fs, src_path, dst_path, CHUNK, RETRIES
                            )
                    else:
                        copy_file(
                            key, src_fs, dst_fs, src_path, dst_path, CHUNK, RETRIES
                        )
            else:
                for key in add_files:
                    if server_side:
                        try:
                            server_side_copy_file(
                                key, src_mapper, dst_mapper, RETRIES
                            )
                        except (RuntimeError, OSError):
                            copy_file(
                                key, src_fs, dst_fs, src_path, dst_path, CHUNK, RETRIES
                            )
                    else:
                        copy_file(
                            key, src_fs, dst_fs, src_path, dst_path, CHUNK, RETRIES
                        )

    if len(delete_files):
        # Delete old files from destination
        if parallel:
            run_parallel(
                delete_file,
                delete_files,
                dst_fs=dst_fs,
                dst_path=dst_path,
                RETRIES=RETRIES,
                n_jobs=n_jobs,
                verbose=verbose,
            )
        else:
            if verbose:
                from rich.progress import track

                for key in track(
                    delete_files,
                    description="Deleting stale files...",
                    total=len(delete_files),
                ):
                    delete_file(key, dst_fs, dst_path, RETRIES)
            else:
                for key in delete_files:
                    delete_file(key, dst_fs, dst_path, RETRIES)

    return {"added_files": add_files, "deleted_files": delete_files}

fsspeckit.common.validate_path

validate_path(
    path: str, base_dir: str | None = None
) -> str

Validate a filesystem path for security issues.

Checks for: - Embedded null bytes and control characters - Path traversal attempts (../ sequences escaping base_dir) - Empty or whitespace-only paths

Parameters:

Name	Type	Description	Default
path	str	The path to validate.	required
base_dir	str | None	Optional base directory. If provided, the path must resolve to a location within this directory (prevents path traversal).	None

Returns:

Type	Description
str	The validated path (unchanged if valid).

Raises:

Type	Description
ValueError	If the path contains forbidden characters, is empty, or escapes the base directory.

Examples:

>>> validate_path("/data/file.parquet")
'/data/file.parquet'

>>> validate_path("../../../etc/passwd", base_dir="/data")
ValueError: Path escapes base directory

>>> validate_path("file\x00.parquet")
ValueError: Path contains forbidden characters

Source code in src/fsspeckit/common/security.py

def validate_path(path: str, base_dir: str | None = None) -> str:
    """Validate a filesystem path for security issues.

    Checks for:
    - Embedded null bytes and control characters
    - Path traversal attempts (../ sequences escaping base_dir)
    - Empty or whitespace-only paths

    Args:
        path: The path to validate.
        base_dir: Optional base directory. If provided, the path must resolve
            to a location within this directory (prevents path traversal).

    Returns:
        The validated path (unchanged if valid).

    Raises:
        ValueError: If the path contains forbidden characters, is empty,
            or escapes the base directory.

    Examples:
        >>> validate_path("/data/file.parquet")
        '/data/file.parquet'

        >>> validate_path("../../../etc/passwd", base_dir="/data")
        ValueError: Path escapes base directory

        >>> validate_path("file\\x00.parquet")
        ValueError: Path contains forbidden characters
    """
    if not path or not path.strip():
        raise ValueError("Path cannot be empty or whitespace-only")

    # Check for forbidden control characters
    for char in path:
        if char in _FORBIDDEN_PATH_CHARS:
            raise ValueError(f"Path contains forbidden control character: {repr(char)}")

    # Check for path traversal when base_dir is specified
    if base_dir is not None:
        import os

        # Normalize both paths for comparison
        base_resolved = os.path.normpath(os.path.abspath(base_dir))

        # Handle relative paths by joining with base
        if not os.path.isabs(path):
            full_path = os.path.join(base_dir, path)
        else:
            full_path = path

        path_resolved = os.path.normpath(os.path.abspath(full_path))

        # Check if resolved path starts with base directory
        if (
            not path_resolved.startswith(base_resolved + os.sep)
            and path_resolved != base_resolved
        ):
            raise ValueError(f"Path '{path}' escapes base directory '{base_dir}'")

    return path

fsspeckit.common.validate_compression_codec

validate_compression_codec(codec: str) -> str

Validate that a compression codec is in the allowed set.

This prevents injection of arbitrary values into SQL queries or filesystem operations that accept codec parameters.

Parameters:

Name	Type	Description	Default
codec	str	The compression codec name to validate.	required

Returns:

Type	Description
str	The validated codec name (lowercased).

Raises:

Type	Description
ValueError	If the codec is not in the allowed set.

Examples:

>>> validate_compression_codec("snappy")
'snappy'

>>> validate_compression_codec("GZIP")
'gzip'

>>> validate_compression_codec("malicious; DROP TABLE")
ValueError: Invalid compression codec

Source code in src/fsspeckit/common/security.py

def validate_compression_codec(codec: str) -> str:
    """Validate that a compression codec is in the allowed set.

    This prevents injection of arbitrary values into SQL queries or
    filesystem operations that accept codec parameters.

    Args:
        codec: The compression codec name to validate.

    Returns:
        The validated codec name (lowercased).

    Raises:
        ValueError: If the codec is not in the allowed set.

    Examples:
        >>> validate_compression_codec("snappy")
        'snappy'

        >>> validate_compression_codec("GZIP")
        'gzip'

        >>> validate_compression_codec("malicious; DROP TABLE")
        ValueError: Invalid compression codec
    """
    if not codec or not isinstance(codec, str):
        raise ValueError("Compression codec must be a non-empty string")

    normalized = codec.lower().strip()

    if normalized not in VALID_COMPRESSION_CODECS:
        valid_list = ", ".join(sorted(VALID_COMPRESSION_CODECS - {"none"}))
        raise ValueError(
            f"Invalid compression codec: '{codec}'. Must be one of: {valid_list}"
        )

    return normalized

fsspeckit.common.scrub_credentials

scrub_credentials(message: str) -> str

Remove or mask credential-like values from a string.

This is intended for use before logging error messages that might contain sensitive information like access keys or tokens.

Parameters:

Name	Type	Description	Default
message	str	The string to scrub.	required

Returns:

Type	Description
str	The string with credential-like values replaced with [REDACTED].

Examples:

>>> scrub_credentials("Error: access_key_id=AKIAIOSFODNN7EXAMPLE")
'Error: access_key_id=[REDACTED]'

>>> scrub_credentials("Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...")
'[REDACTED]'

Source code in src/fsspeckit/common/security.py

def scrub_credentials(message: str) -> str:
    """Remove or mask credential-like values from a string.

    This is intended for use before logging error messages that might
    contain sensitive information like access keys or tokens.

    Args:
        message: The string to scrub.

    Returns:
        The string with credential-like values replaced with [REDACTED].

    Examples:
        >>> scrub_credentials("Error: access_key_id=AKIAIOSFODNN7EXAMPLE")
        'Error: access_key_id=[REDACTED]'

        >>> scrub_credentials("Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...")
        '[REDACTED]'
    """
    if not message:
        return message

    result = message

    for pattern in _CREDENTIAL_PATTERNS:
        # Replace matched groups with [REDACTED]
        def redact_match(match: re.Match) -> str:
            groups = match.groups()
            if len(groups) >= 2:
                # Pattern with key=value format - keep the key, redact the value
                return match.group(0).replace(groups[-1], "[REDACTED]")
            else:
                # Single match - redact entire thing
                return "[REDACTED]"

        result = pattern.sub(redact_match, result)

    return result

fsspeckit.common.scrub_exception

scrub_exception(exc: BaseException) -> str

Scrub credentials from an exception's string representation.

Parameters:

Name	Type	Description	Default
exc	BaseException	The exception to scrub.	required

Returns:

Type	Description
str	A scrubbed string representation of the exception.

Source code in src/fsspeckit/common/security.py

def scrub_exception(exc: BaseException) -> str:
    """Scrub credentials from an exception's string representation.

    Args:
        exc: The exception to scrub.

    Returns:
        A scrubbed string representation of the exception.
    """
    return scrub_credentials(str(exc))

fsspeckit.common.safe_format_error

safe_format_error(
    operation: str,
    path: str | None = None,
    error: BaseException | None = None,
    **context: Any,
) -> str

Format an error message with credentials scrubbed.

Parameters:

Name	Type	Description	Default
operation	str	Description of the operation that failed.	required
path	str | None	Optional path involved in the operation.	None
error	BaseException | None	Optional exception that occurred.	None
**context	Any	Additional context key-value pairs.	{}

Returns:

Type	Description
str	A formatted, credential-scrubbed error message.

Source code in src/fsspeckit/common/security.py

def safe_format_error(
    operation: str,
    path: str | None = None,
    error: BaseException | None = None,
    **context: Any,
) -> str:
    """Format an error message with credentials scrubbed.

    Args:
        operation: Description of the operation that failed.
        path: Optional path involved in the operation.
        error: Optional exception that occurred.
        **context: Additional context key-value pairs.

    Returns:
        A formatted, credential-scrubbed error message.
    """
    parts = [f"Failed to {operation}"]

    if path:
        parts.append(f"at '{path}'")

    if error:
        parts.append(f": {scrub_exception(error)}")

    if context:
        context_str = ", ".join(
            f"{k}={scrub_credentials(str(v))}" for k, v in context.items()
        )
        parts.append(f" ({context_str})")

    return " ".join(parts)

fsspeckit.common.validate_columns

validate_columns(
    columns: list[str] | None, valid_columns: list[str]
) -> list[str] | None

Validate that requested columns exist in the schema.

This is a helper to prevent column injection in SQL-like operations.

Parameters:

Name	Type	Description	Default
columns	list[str] | None	List of column names to validate, or None.	required
valid_columns	list[str]	List of valid column names from the schema.	required

Returns:

Type	Description
list[str] | None	The validated columns list, or None if columns was None.

Raises:

Type	Description
ValueError	If any column is not in the valid set.

Source code in src/fsspeckit/common/security.py

def validate_columns(
    columns: list[str] | None, valid_columns: list[str]
) -> list[str] | None:
    """Validate that requested columns exist in the schema.

    This is a helper to prevent column injection in SQL-like operations.

    Args:
        columns: List of column names to validate, or None.
        valid_columns: List of valid column names from the schema.

    Returns:
        The validated columns list, or None if columns was None.

    Raises:
        ValueError: If any column is not in the valid set.
    """
    if columns is None:
        return None

    valid_set = set(valid_columns)
    invalid = [col for col in columns if col not in valid_set]

    if invalid:
        raise ValueError(
            f"Invalid column(s): {', '.join(invalid)}. "
            f"Valid columns are: {', '.join(sorted(valid_set))}"
        )

    return columns