def list_namespaces()

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

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

    def list_namespaces(self, namespace: Union[str, Identifier] = ()) -> List[Identifier]:
        """List top-level namespaces from the catalog.

        We do not support hierarchical namespace.

        Returns:
            List[Identifier]: a List of namespace identifiers.
        """
        # Hierarchical namespace is not supported. Return an empty list
        if namespace:
            return []

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

        try:
            page_iterator = paginator.paginate(
                TableName=self.dynamodb_table_name,
                ConsistentRead=True,
                KeyConditionExpression=f"{DYNAMODB_COL_IDENTIFIER} = :identifier",
                ExpressionAttributeValues={
                    ":identifier": {
                        "S": DYNAMODB_NAMESPACE,
                    }
                },
            )
        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

        database_identifiers = []
        for page in page_iterator:
            for item in page["Items"]:
                _dict = _convert_dynamo_item_to_regular_dict(item)
                namespace_col = _dict[DYNAMODB_COL_NAMESPACE]
                database_identifiers.append(self.identifier_to_tuple(namespace_col))

        return database_identifiers