def list_tables()

in pyiceberg/catalog/dynamodb.py [0:0]

• 30 lines of code
• 5 McCabe index (conditional complexity)

    def list_tables(self, namespace: Union[str, Identifier]) -> List[Identifier]:
        """List Iceberg tables under the given namespace in the catalog.

        Args:
            namespace (str | Identifier): Namespace identifier to search.

        Returns:
            List[Identifier]: list of table identifiers.
        """
        database_name = self.identifier_to_database(namespace, NoSuchNamespaceError)

        paginator = self.dynamodb.get_paginator("query")

        try:
            page_iterator = paginator.paginate(
                TableName=self.dynamodb_table_name,
                IndexName=DYNAMODB_NAMESPACE_GSI,
                KeyConditionExpression=f"{DYNAMODB_COL_NAMESPACE} = :namespace ",
                ExpressionAttributeValues={
                    ":namespace": {
                        "S": database_name,
                    }
                },
            )
        except (
            self.dynamodb.exceptions.ProvisionedThroughputExceededException,
            self.dynamodb.exceptions.RequestLimitExceeded,
            self.dynamodb.exceptions.InternalServerError,
            self.dynamodb.exceptions.ResourceNotFoundException,
        ) as e:
            raise GenericDynamoDbError(e.message) from e

        table_identifiers = []
        for page in page_iterator:
            for item in page["Items"]:
                _dict = _convert_dynamo_item_to_regular_dict(item)
                identifier_col = _dict[DYNAMODB_COL_IDENTIFIER]
                if identifier_col == DYNAMODB_NAMESPACE:
                    continue

                table_identifiers.append(self.identifier_to_tuple(identifier_col))

        return table_identifiers