Skip to content

cdm_data_loaders.converters.cli.iceberg_base

Shared CLI settings and catalog traversal for Iceberg catalog schema dumpers.

IcebergCatalogDumpSettings

Bases: LoggerSettings

Shared settings for dumping table schemas from an Iceberg data catalog.

Source code in src/cdm_data_loaders/converters/cli/iceberg_base.py 
26
27
28
29
30
31
32
33
class IcebergCatalogDumpSettings(LoggerSettings):
    """Shared settings for dumping table schemas from an Iceberg data catalog."""

    model_config: SettingsConfigDict = default_settings_with_shortcuts(cli_shortcuts=CATALOG_CLI_SHORTCUTS)

    catalog: Annotated[NonEmptyStr, Field(description="Name of the catalog to retrieve schemas from")]
    output_dir: LocalOutputDir
    group_by_namespace: bool = Field(default=False, description="Export one schema per namespace")

TableConverter

Bases: Protocol

Protocol for converting an Iceberg table into a schema or shared namespace document.

Source code in src/cdm_data_loaders/converters/cli/iceberg_base.py 
36
37
38
39
40
41
42
43
44
45
46
47
class TableConverter[SchemaT](Protocol):
    """Protocol for converting an Iceberg table into a schema or shared namespace document."""

    def __call__(
        self,
        table: Table,
        identifier: tuple[str, ...],
        *,
        schema: SchemaT | None = None,
    ) -> SchemaT:
        """Convert ``table`` to ``SchemaT``, optionally accumulating into ``schema``."""
        ...

__call__ 

__call__(
    table: Table,
    identifier: tuple[str, ...],
    *,
    schema: SchemaT | None = None
) -> SchemaT

Convert table to SchemaT, optionally accumulating into schema.

Source code in src/cdm_data_loaders/converters/cli/iceberg_base.py 
39
40
41
42
43
44
45
46
47
def __call__(
    self,
    table: Table,
    identifier: tuple[str, ...],
    *,
    schema: SchemaT | None = None,
) -> SchemaT:
    """Convert ``table`` to ``SchemaT``, optionally accumulating into ``schema``."""
    ...

CatalogDumpDriver dataclass

Format-specific callbacks for dumping Iceberg catalog schemas.

Source code in src/cdm_data_loaders/converters/cli/iceberg_base.py 
50
51
52
53
54
55
56
57
58
@dataclass(frozen=True, slots=True)
class CatalogDumpDriver[SchemaT]:
    """Format-specific callbacks for dumping Iceberg catalog schemas."""

    file_suffix: str
    make_namespace_schema: Callable[[str], SchemaT]
    convert_table: TableConverter[SchemaT]
    has_grouped_entries: Callable[[SchemaT], bool]
    write_schema: Callable[[SchemaT, Path, str], None]

write_schema_text 

write_schema_text(
    text: str, out_file: Path, log: Logger
) -> None

Write serialized schema text to disk, warning before replacing an existing file.

Parameters:

Name Type Description Default
text str

Serialized schema text to write.

 required
out_file Path

Destination file path.

 required
log Logger

Logger instance.

 required
Source code in src/cdm_data_loaders/converters/cli/iceberg_base.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def write_schema_text(text: str, out_file: Path, log: Logger) -> None:
    """Write serialized schema text to disk, warning before replacing an existing file.

    :param text: Serialized schema text to write.
    :type text: str
    :param out_file: Destination file path.
    :type out_file: Path
    :param log: Logger instance.
    :type log: Logger
    """
    if out_file.exists():
        log.warning("Overwriting existing schema file %s", out_file)
    out_file.write_text(text, encoding="utf-8")
    log.info("Wrote schema to %s", out_file)

dump_catalog 

dump_catalog(
    settings: IcebergCatalogDumpSettings,
    driver: CatalogDumpDriver[SchemaT],
    log: Logger,
) -> None

Iterate through an Iceberg catalog and write converted table or namespace schemas.

Parameters:

Name Type Description Default
settings IcebergCatalogDumpSettings

Catalog dump settings.

 required
driver CatalogDumpDriver[SchemaT]

Format-specific schema conversion and serialization driver.

 required
log Logger

Logger instance.

 required
Source code in src/cdm_data_loaders/converters/cli/iceberg_base.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
def dump_catalog[SchemaT](
    settings: IcebergCatalogDumpSettings,
    driver: CatalogDumpDriver[SchemaT],
    log: Logger,
) -> None:
    """Iterate through an Iceberg catalog and write converted table or namespace schemas.

    :param settings: Catalog dump settings.
    :type settings: IcebergCatalogDumpSettings
    :param driver: Format-specific schema conversion and serialization driver.
    :type driver: CatalogDumpDriver[SchemaT]
    :param log: Logger instance.
    :type log: Logger
    """
    generated_at = datetime.now(UTC).isoformat()
    failures = 0

    with load_catalog(settings.catalog) as catalog:
        namespaces = catalog.list_namespaces()
        log.info("Dumping schemas for %d namespace(s) from catalog %r", len(namespaces), settings.catalog)
        for namespace in namespaces:
            namespace_name = ".".join(namespace)
            namespace_schema: SchemaT | None = (
                driver.make_namespace_schema(namespace_name) if settings.group_by_namespace else None
            )
            for identifier in sorted(catalog.list_tables(namespace)):
                try:
                    table = catalog.load_table(identifier)
                    schema_doc = driver.convert_table(table, identifier, schema=namespace_schema)
                except Exception:
                    failures += 1
                    log.exception("Failed to convert table %s; skipping", identifier)
                    continue

                if namespace_schema is None:
                    driver.write_schema(
                        schema_doc,
                        settings.output_dir / (".".join(identifier) + driver.file_suffix),
                        generated_at,
                    )

            if namespace_schema is not None and driver.has_grouped_entries(namespace_schema):
                driver.write_schema(
                    namespace_schema,
                    settings.output_dir / (namespace_name + driver.file_suffix),
                    generated_at,
                )

    if failures:
        log.warning("Completed with %d table(s) skipped due to conversion failures", failures)