|
| 1 | +from abc import ABCMeta |
| 2 | +from typing import TYPE_CHECKING, Any, Callable, TypeVar |
| 3 | + |
| 4 | +if TYPE_CHECKING: |
| 5 | + from pyiceberg.catalog import Catalog |
| 6 | + |
| 7 | +_T = TypeVar("_T", bound="Catalog") |
| 8 | + |
| 9 | + |
| 10 | +class NamespaceMeta(type): |
| 11 | + """Metaclass for managing namespace support configuration in catalog implementations. |
| 12 | +
|
| 13 | + This metaclass automatically handles the inheritance and initialization of namespace-related |
| 14 | + attributes for catalog classes. It ensures that namespace support configuration is properly |
| 15 | + propagated through class inheritance hierarchies. |
| 16 | +
|
| 17 | + Attributes: |
| 18 | + _support_namespaces (bool): Indicates whether the catalog supports nested namespaces. |
| 19 | + Defaults to False. When True, the catalog can handle hierarchical namespace |
| 20 | + structures beyond simple flat namespaces. |
| 21 | + _max_namespace_depth (int): Maximum depth allowed for nested namespaces. |
| 22 | + Defaults to -1 (unlimited depth). When set to a positive integer, |
| 23 | + restricts the nesting level of namespaces that can be created. |
| 24 | + """ |
| 25 | + |
| 26 | + _support_namespaces: bool = False |
| 27 | + _max_namespace_depth: int = -1 |
| 28 | + |
| 29 | + def __new__(mcls, name: str, bases: tuple[type, ...], atrrs: dict[str, Any], /, **kwargs: Any) -> type: |
| 30 | + cls = super().__new__(mcls, name, bases, atrrs, **kwargs) |
| 31 | + if "_support_namespaces" in atrrs: |
| 32 | + pass # Already defined in the class |
| 33 | + elif hasattr(bases[0], "_support_namespaces"): |
| 34 | + cls._support_namespaces = bases[0]._support_namespaces |
| 35 | + else: |
| 36 | + cls._support_namespaces = NamespaceMeta._support_namespaces |
| 37 | + return cls |
| 38 | + |
| 39 | + |
| 40 | +class CatalogABCMeta(NamespaceMeta, ABCMeta): |
| 41 | + """Metaclass for catalog implementations that combines namespace and abstract base class functionality. |
| 42 | +
|
| 43 | + This metaclass inherits from both NamespaceMeta and ABCMeta. |
| 44 | + """ |
| 45 | + |
| 46 | + |
| 47 | +def multiple_namespaces( |
| 48 | + _cls: type[_T] | None = None, /, disabled: bool = False, max_depth: int = -1 |
| 49 | +) -> type[_T] | Callable[[type[_T]], type[_T]]: |
| 50 | + def wrapper(cls: type[_T]) -> type[_T]: |
| 51 | + if not hasattr(cls, "_support_namespaces"): |
| 52 | + raise ValueError(f"{cls.__name__} must inherit Catalog with CatalogABCMeta and NamespaceMeta to use this decorator") |
| 53 | + if max_depth >= 0 and max_depth <= 1 or disabled: |
| 54 | + cls._support_namespaces = False |
| 55 | + else: |
| 56 | + cls._support_namespaces = True |
| 57 | + cls._max_namespace_depth = max_depth |
| 58 | + return cls |
| 59 | + |
| 60 | + return wrapper if _cls is None else wrapper(_cls) |
0 commit comments