cdm_data_loaders.core.transforms ¶

def resolve_schema_map(
    default_table: str,
    *,
    transform: RecordTransformFn | None = None,
    unreconciled_output_schemas: ArrowSchemaRegistry | None = None,
    default_pyarrow_schema: pa.Schema | None = None,
) -> ArrowSchemaRegistry | None:
    """Resolve the active table-to-PyArrow-schema registry for a reader.

    :param default_table: default target table name for the pipeline/entity
    :type default_table: str
    :param transform: optional ``RecordTransformFn`` configured for ``default_table``
    :type transform: RecordTransformFn | None
    :param unreconciled_output_schemas: optional ``ArrowSchemaRegistry``
    :type unreconciled_output_schemas: ArrowSchemaRegistry | None
    :param default_pyarrow_schema: default PyArrow schema when no override is configured
    :type default_pyarrow_schema: pa.Schema | None
    :return: ``ArrowSchemaRegistry`` mapping destination table names to ``pa.Schema`` objects, or ``None``
    """
    if unreconciled_output_schemas is not None:
        if transform is None and default_table not in unreconciled_output_schemas.registry:
            err_msg = f"unreconciled_output_schemas for {default_table!r} must include {default_table!r} when transform is None"
            raise ValueError(err_msg)
        return unreconciled_output_schemas
    if default_pyarrow_schema is not None and transform is None:
        return ArrowSchemaRegistry(
            registry={
                default_table: default_pyarrow_schema,
            }
        )
    return None

def validate_routed_output(raw_routed: object) -> dict[str, list[dict[str, Any]]]:
    """Validate the return value of ``transform_fn`` and strip empty table lists.

    :param raw_routed: raw output from ``transform_fn`` for a single record
    :type  raw_routed: object
    :raises ValueError: if ``raw_routed`` is not a mapping, any table name is empty,
        or rows for any table are not a sequence of non-empty dicts
    :return: validated dictionary mapping destination table names to non-empty lists of row dictionaries
    :rtype : dict[str, list[dict[str, Any]]]
    """
    routed: dict[str, list[dict[str, Any]]] = {}
    errors: list[str] = []
    if not isinstance(raw_routed, Mapping):
        errors.append("transform_fn must return a mapping of table names to sequences of row dicts")
    else:
        for table_name, rows in raw_routed.items():
            if not isinstance(table_name, str) or not table_name.strip():
                errors.append("transform_fn output keys must be non-empty table names")

            if not isinstance(rows, Sequence) or isinstance(rows, (str, bytes, bytearray, Mapping)):
                errors.append(f"transform_fn rows for table {table_name!r} must be a sequence of dicts")
                continue

            row_list = list(rows)
            if not all(isinstance(row, dict) for row in row_list) or not all(row for row in row_list):
                errors.append(f"transform_fn rows for table {table_name!r} must be non-empty dicts")
                continue

            if not errors and row_list:
                routed[table_name.strip()] = row_list

    if errors:
        raise ValueError("\n".join(errors))
    return routed

def route_and_coerce_record(
    record: dict[str, Any],
    default_table: str,
    *,
    transform: RecordTransformFn | None = None,
    schema_map: ArrowSchemaRegistry | None = None,
) -> dict[str, list[dict[str, Any]]]:
    """Route a validated record through ``transform`` and optionally coerce rows for PyArrow.

    :param record: parsed and validated input record dictionary
    :type record: dict[str, Any]
    :param default_table: destination table name when ``transform`` is ``None``
    :type default_table: str
    :param transform: optional ``RecordTransformFn`` for ``default_table``
    :type transform: RecordTransformFn | None
    :param schema_map: active destination ``ArrowSchemaRegistry``, if any
    :type schema_map: ArrowSchemaRegistry | None
    :return: dictionary mapping destination table names to non-empty lists of row dictionaries
    """
    routed = validate_routed_output(transform(record)) if transform is not None else {default_table: [record]}

    if not routed:
        return {}

    if schema_map is not None:
        return {
            out_table: (
                [coerce_record_for_pyarrow(row, schema_map.registry[out_table]) for row in rows]
                if out_table in schema_map.registry
                else rows
            )
            for out_table, rows in routed.items()
        }

    return routed

def validate_table_transforms(
    settings: Any,  # noqa: ANN401
    valid_tables: Iterable[str],
) -> None:
    """Ensure every key in ``settings.table_transforms`` matches a configured input table."""
    table_transforms: TransformRegistry | None = getattr(settings, "table_transforms", None)
    if table_transforms is None:
        return
    validate_registry_table_names(table_transforms.registry, valid_tables, "transforms_module")

def uses_pyarrow_format(settings: Any) -> bool:  # noqa: ANN401
    """Return whether ``settings.loader_file_format`` is a PyArrow-backed loader format."""
    return getattr(settings, "loader_file_format", LoaderFileFormatEnum.PARQUET) in PYARROW_FILE_FORMATS

def needs_default_pyarrow_schema(
    table_name: str,
    *,
    use_pyarrow: bool = True,
    table_transforms: TransformRegistry | None = None,
    output_schemas: ArrowSchemaRegistry | None = None,
) -> bool:
    """Return whether ``table_name`` needs its PyArrow schema generated from the input schema.

    :param table_name: target table name
    :type table_name: str
    :param use_pyarrow: whether PyArrow output is enabled
    :type use_pyarrow: bool
    :param table_transforms: optional configured table transforms registry
    :type table_transforms: TransformRegistry | None
    :param output_schemas: optional user-supplied output schemas registry
    :type output_schemas: ArrowSchemaRegistry | None
    :return: ``True`` if a default PyArrow schema should be generated for ``table_name``
    :rtype: bool
    """
    if not use_pyarrow:
        return False
    if table_transforms is not None and table_name in table_transforms.registry:
        return False
    return not (output_schemas is not None and table_name in output_schemas.registry)