Function construct_schema_and_data_format

python/pathway/io/_utils.py:237–359

(
    format: str,
    *,
    schema: type[Schema] | None = None,
    with_metadata: bool = False,
    autogenerate_key: bool = False,
    csv_settings: CsvParserSettings | None = None,
    json_field_paths: dict[str, str] | None = None,
    schema_registry_settings: SchemaRegistrySettings | None = None,
    with_native_record_key: bool = False,
    _stacklevel: int = 1,
)

def construct_schema_and_data_format(
    format: str,
    *,
    schema: type[Schema] | None = None,
    with_metadata: bool = False,
    autogenerate_key: bool = False,
    csv_settings: CsvParserSettings | None = None,
    json_field_paths: dict[str, str] | None = None,
    schema_registry_settings: SchemaRegistrySettings | None = None,
    with_native_record_key: bool = False,
    _stacklevel: int = 1,
) -> tuple[type[Schema], api.DataFormat]:
    data_format_type = get_data_format_type(format, SUPPORTED_INPUT_FORMATS)

    if data_format_type == "identity":
        kwargs = locals()
        unexpected_params = [
            "schema",
            "csv_settings",
            "json_field_paths",
            "schema_registry_settings",
        ]
        for param in unexpected_params:
            if param in kwargs and kwargs[param] is not None:
                raise ValueError(f"Unexpected argument for {format!r} format: {param}")

        parse_utf8 = format not in ("binary", "only_metadata", "raw")
        schema = construct_raw_data_schema_by_flags(
            with_native_record_key=with_native_record_key,
            parse_utf8=parse_utf8,
            with_metadata=with_metadata,
        )
        schema, api_schema = read_schema(schema)

        return schema, api.DataFormat(
            format_type=data_format_type,
            **api_schema,
            parse_utf8=parse_utf8,
            key_generation_policy=(
                api.KeyGenerationPolicy.ALWAYS_AUTOGENERATE
                if autogenerate_key
                else api.KeyGenerationPolicy.PREFER_MESSAGE_KEY
            ),
            schema_registry_settings=maybe_schema_registry_settings(
                schema_registry_settings
            ),
            message_queue_key_field=(
                MESSAGE_QUEUE_KEY_COLUMN_NAME if with_native_record_key else None
            ),
        )

    schema = assert_schema_not_none(schema, data_format_type)
    if METADATA_COLUMN_NAME in schema.column_names():
        if with_metadata:
            raise ValueError(
                f"The schema already declares a {METADATA_COLUMN_NAME!r} column, "
                f"which conflicts with 'with_metadata=True'. Either remove "
                f"{METADATA_COLUMN_NAME!r} from the schema or set "