"""Arbitrary entity management for the SDP Configuration Database."""
import warnings
from typing import Iterable
from ..base_transaction import BaseTransaction
from .entity_operations import EntityOperations
_UNSUPPORTED_PATH_MSG_TPL = """
The path '%s' is not part of any of the root paths hosting entities known to
the SDP Configuration DB, and therefore you seem to be dealing with unsupported
entities. If you are experimenting with storing such entities into the SDP
Configuration DB, you should consider bringng support into the ska-sdp-config
package instead.
"""
# pylint: disable-next=too-few-public-methods
[docs]
class ArbitraryOperations:
"""
Database operations related to arbitrary entities. Entities are selected at
runtime with a given path. If the given path doesn't belong to the set of
known paths given at construction time, a warning is issued.
"""
def __init__(self, txn: BaseTransaction, known_roots: Iterable[str]):
self._txn = txn
self._known_roots = set(known_roots)
[docs]
def __call__(self, path: str) -> EntityOperations:
"""Returns a set of operations over the requested arbitrary path."""
if not any(path.startswith(root) for root in self._known_roots):
warnings.warn(
_UNSUPPORTED_PATH_MSG_TPL % path, UserWarning, stacklevel=2
)
return EntityOperations(self._txn, path)