console

create()

Operation to create a namespace.

Source code in pyiceberg/cli/console.py

@run.group()
def create() -> None:
    """Operation to create a namespace."""

describe(ctx, entity, identifier)

Describe a namespace or a table.

Source code in pyiceberg/cli/console.py

@run.command()
@click.option("--entity", type=click.Choice(["any", "namespace", "table"]), default="any")
@click.argument("identifier")
@click.pass_context
@catch_exception()
def describe(ctx: Context, entity: Literal["name", "namespace", "table"], identifier: str) -> None:
    """Describe a namespace or a table."""
    catalog, output = _catalog_and_output(ctx)
    identifier_tuple = Catalog.identifier_to_tuple(identifier)

    is_namespace = False
    if entity in {"namespace", "any"} and len(identifier_tuple) > 0:
        try:
            namespace_properties = catalog.load_namespace_properties(identifier_tuple)
            output.describe_properties(namespace_properties)
            is_namespace = True
        except NoSuchNamespaceError as exc:
            if entity != "any" or len(identifier_tuple) == 1:  # type: ignore
                raise exc

    is_table = False
    if entity in {"table", "any"} and len(identifier_tuple) > 1:
        try:
            catalog_table = catalog.load_table(identifier)
            output.describe_table(catalog_table)
            is_table = True
        except NoSuchTableError as exc:
            if entity != "any":
                raise exc

    if is_namespace is False and is_table is False:
        raise NoSuchTableError(f"Table or namespace does not exist: {identifier}")

drop()

Operations to drop a namespace or table.

Source code in pyiceberg/cli/console.py

@run.group()
def drop() -> None:
    """Operations to drop a namespace or table."""

files(ctx, identifier, history)

List all the files of the table.

Source code in pyiceberg/cli/console.py

@run.command()
@click.argument("identifier")
@click.option("--history", is_flag=True)
@click.pass_context
@catch_exception()
def files(ctx: Context, identifier: str, history: bool) -> None:
    """List all the files of the table."""
    catalog, output = _catalog_and_output(ctx)

    catalog_table = catalog.load_table(identifier)
    output.files(catalog_table, history)

get()

Fetch properties on tables/namespaces.

Source code in pyiceberg/cli/console.py

@properties.group()
def get() -> None:
    """Fetch properties on tables/namespaces."""

get_namespace(ctx, identifier, property_name)

Fetch properties on a namespace.

Source code in pyiceberg/cli/console.py

@get.command("namespace")
@click.argument("identifier")
@click.argument("property_name", required=False)
@click.pass_context
@catch_exception()
def get_namespace(ctx: Context, identifier: str, property_name: str) -> None:
    """Fetch properties on a namespace."""
    catalog, output = _catalog_and_output(ctx)
    identifier_tuple = Catalog.identifier_to_tuple(identifier)

    namespace_properties = catalog.load_namespace_properties(identifier_tuple)
    assert namespace_properties

    if property_name:
        if property_value := namespace_properties.get(property_name):
            output.text(property_value)
        else:
            raise NoSuchPropertyException(f"Could not find property {property_name} on namespace {identifier}")
    else:
        output.describe_properties(namespace_properties)

get_table(ctx, identifier, property_name)

Fetch properties on a table.

Source code in pyiceberg/cli/console.py

@get.command("table")
@click.argument("identifier")
@click.argument("property_name", required=False)
@click.pass_context
@catch_exception()
def get_table(ctx: Context, identifier: str, property_name: str) -> None:
    """Fetch properties on a table."""
    catalog, output = _catalog_and_output(ctx)
    identifier_tuple = Catalog.identifier_to_tuple(identifier)

    metadata = catalog.load_table(identifier_tuple).metadata
    assert metadata

    if property_name:
        if property_value := metadata.properties.get(property_name):
            output.text(property_value)
        else:
            raise NoSuchPropertyException(f"Could not find property {property_name} on table {identifier}")
    else:
        output.describe_properties(metadata.properties)

list(ctx, parent)

List tables or namespaces.

Source code in pyiceberg/cli/console.py

@run.command()
@click.pass_context
@click.argument("parent", required=False)
@catch_exception()
def list(ctx: Context, parent: Optional[str]) -> None:  # pylint: disable=redefined-builtin
    """List tables or namespaces."""
    catalog, output = _catalog_and_output(ctx)

    identifiers = catalog.list_namespaces(parent or ())
    if not identifiers and parent:
        identifiers = catalog.list_tables(parent)
    output.identifiers(identifiers)

list_refs(ctx, identifier, type, verbose)

List all the refs in the provided table.

Source code in pyiceberg/cli/console.py

@run.command()
@click.argument("identifier")
@click.option("--type", required=False)
@click.option("--verbose", type=click.BOOL)
@click.pass_context
@catch_exception()
def list_refs(ctx: Context, identifier: str, type: str, verbose: bool) -> None:
    """List all the refs in the provided table."""
    catalog, output = _catalog_and_output(ctx)
    table = catalog.load_table(identifier)
    refs = table.refs()
    if type:
        type = type.lower()
        if type not in {"branch", "tag"}:
            raise ValueError(f"Type must be either branch or tag, got: {type}")

    relevant_refs = [
        (ref_name, ref.snapshot_ref_type, _retention_properties(ref, table.properties))
        for (ref_name, ref) in refs.items()
        if not type or ref.snapshot_ref_type == type
    ]

    output.describe_refs(relevant_refs)

location(ctx, identifier)

Return the location of the table.

Source code in pyiceberg/cli/console.py

@run.command()
@click.argument("identifier")
@click.pass_context
@catch_exception()
def location(ctx: Context, identifier: str) -> None:
    """Return the location of the table."""
    catalog, output = _catalog_and_output(ctx)
    table = catalog.load_table(identifier)
    output.text(table.location())

namespace(ctx, identifier, property_name)

Remove a property from a namespace.

Source code in pyiceberg/cli/console.py

@remove.command()  # type: ignore
@click.argument("identifier")
@click.argument("property_name")
@click.pass_context
@catch_exception()
def namespace(ctx: Context, identifier: str, property_name: str) -> None:  # noqa: F811
    """Remove a property from a namespace."""
    catalog, output = _catalog_and_output(ctx)

    result = catalog.update_namespace_properties(identifier, removals={property_name})

    if result.removed == [property_name]:
        output.text(f"Property {property_name} removed from {identifier}")
    else:
        raise NoSuchPropertyException(f"Property {property_name} does not exist on {identifier}")

properties()

Properties on tables/namespaces.

Source code in pyiceberg/cli/console.py

@run.group()
def properties() -> None:
    """Properties on tables/namespaces."""

remove()

Remove a property from tables/namespaces.

Source code in pyiceberg/cli/console.py

@properties.group()
def remove() -> None:
    """Remove a property from tables/namespaces."""

rename(ctx, from_identifier, to_identifier)

Rename a table.

Source code in pyiceberg/cli/console.py

@run.command()
@click.argument("from_identifier")
@click.argument("to_identifier")
@click.pass_context
@catch_exception()
def rename(ctx: Context, from_identifier: str, to_identifier: str) -> None:
    """Rename a table."""
    catalog, output = _catalog_and_output(ctx)

    catalog.rename_table(from_identifier, to_identifier)
    output.text(f"Renamed table from {from_identifier} to {to_identifier}")

schema(ctx, identifier)

Get the schema of the table.

Source code in pyiceberg/cli/console.py

@run.command()
@click.argument("identifier")
@click.pass_context
@catch_exception()
def schema(ctx: Context, identifier: str) -> None:
    """Get the schema of the table."""
    catalog, output = _catalog_and_output(ctx)
    table = catalog.load_table(identifier)
    output.schema(table.schema())

set()

Set a property on tables/namespaces.

Source code in pyiceberg/cli/console.py

@properties.group()
def set() -> None:
    """Set a property on tables/namespaces."""

spec(ctx, identifier)

Return the partition spec of the table.

Source code in pyiceberg/cli/console.py

@run.command()
@click.argument("identifier")
@click.pass_context
@catch_exception()
def spec(ctx: Context, identifier: str) -> None:
    """Return the partition spec of the table."""
    catalog, output = _catalog_and_output(ctx)
    table = catalog.load_table(identifier)
    output.spec(table.spec())

table(ctx, identifier, property_name)

Remove a property from a table.

Source code in pyiceberg/cli/console.py

@remove.command()  # type: ignore
@click.argument("identifier")
@click.argument("property_name")
@click.pass_context
@catch_exception()
def table(ctx: Context, identifier: str, property_name: str) -> None:  # noqa: F811
    """Remove a property from a table."""
    catalog, output = _catalog_and_output(ctx)
    table = catalog.load_table(identifier)
    if property_name in table.metadata.properties:
        output.exception(NotImplementedError("Writing is WIP"))
        ctx.exit(1)
    else:
        raise NoSuchPropertyException(f"Property {property_name} does not exist on {identifier}")

uuid(ctx, identifier)

Return the UUID of the table.

Source code in pyiceberg/cli/console.py

@run.command()
@click.argument("identifier")
@click.pass_context
@catch_exception()
def uuid(ctx: Context, identifier: str) -> None:
    """Return the UUID of the table."""
    catalog, output = _catalog_and_output(ctx)
    metadata = catalog.load_table(identifier).metadata
    output.uuid(metadata.table_uuid)

version(ctx)

Print pyiceberg version.

Source code in pyiceberg/cli/console.py

@run.command()
@click.pass_context
@catch_exception()
def version(ctx: Context) -> None:
    """Print pyiceberg version."""
    ctx.obj["output"].version(__version__)