(
cls,
table: Table,
*,
format: str = "json",
delimiter: str = ",",
key: ColumnReference | None = None,
value: ColumnReference | None = None,
headers: Iterable[ColumnReference] | None = None,
topic_name: ColumnReference | None = None,
schema_registry_settings: SchemaRegistrySettings | None = None,
subject: str | None = None,
allowed_key_types: tuple[dt.DType, ...] | None = (dt.BYTES, dt.STR, dt.ANY),
allowed_value_types: tuple[dt.DType, ...] | None = (dt.BYTES, dt.STR, dt.ANY),
)
| 387 | |
| 388 | @classmethod |
| 389 | def construct( |
| 390 | cls, |
| 391 | table: Table, |
| 392 | *, |
| 393 | format: str = "json", |
| 394 | delimiter: str = ",", |
| 395 | key: ColumnReference | None = None, |
| 396 | value: ColumnReference | None = None, |
| 397 | headers: Iterable[ColumnReference] | None = None, |
| 398 | topic_name: ColumnReference | None = None, |
| 399 | schema_registry_settings: SchemaRegistrySettings | None = None, |
| 400 | subject: str | None = None, |
| 401 | allowed_key_types: tuple[dt.DType, ...] | None = (dt.BYTES, dt.STR, dt.ANY), |
| 402 | allowed_value_types: tuple[dt.DType, ...] | None = (dt.BYTES, dt.STR, dt.ANY), |
| 403 | ) -> MessageQueueOutputFormat: |
| 404 | if delimiter != "," and format != "dsv": |
| 405 | raise ValueError( |
| 406 | f"'delimiter' is only meaningful for the 'dsv' format, but " |
| 407 | f"{format!r} was specified. Drop the 'delimiter' argument " |
| 408 | f"or use format='dsv'." |
| 409 | ) |
| 410 | if subject is not None and schema_registry_settings is None: |
| 411 | raise ValueError( |
| 412 | "'subject' was provided without 'schema_registry_settings'. " |
| 413 | "The 'subject' parameter only has an effect when a schema " |
| 414 | "registry is configured; either pass 'schema_registry_settings' " |
| 415 | "or remove 'subject'." |
| 416 | ) |
| 417 | if schema_registry_settings is not None and subject is None: |
| 418 | raise ValueError( |
| 419 | "'schema_registry_settings' was provided without 'subject'. " |
| 420 | "When a schema registry is configured, 'subject' must also be " |
| 421 | "set so the formatter knows which subject to encode under." |
| 422 | ) |
| 423 | if subject is not None and not subject: |
| 424 | raise ValueError( |
| 425 | "'subject' must be a non-empty string; got an empty string. " |
| 426 | "Schema Registry subjects identify a named schema version, " |
| 427 | "and an empty subject is never a valid registry entry." |
| 428 | ) |
| 429 | if schema_registry_settings is not None and format != "json": |
| 430 | raise ValueError( |
| 431 | f"'schema_registry_settings' is only meaningful for the 'json' " |
| 432 | f"format, but {format!r} was specified. The Confluent Schema " |
| 433 | "Registry currently encodes JSON payloads only; remove " |
| 434 | "'schema_registry_settings' or use format='json'." |
| 435 | ) |
| 436 | |
| 437 | key_field_index = None |
| 438 | header_fields: dict[str, int] = {} |
| 439 | extracted_field_indices: dict[str, int] = {} |
| 440 | columns_to_extract: list[ColumnReference] = [] |
| 441 | |
| 442 | if topic_name is not None: |
| 443 | topic_name_index = cls.add_column_reference_to_extract( |
| 444 | topic_name, columns_to_extract, extracted_field_indices |
| 445 | ) |
| 446 | if topic_name._column.dtype not in (dt.STR, dt.ANY): |
no test coverage detected