Source code for tiled.query_registration

These objects express high-level queries and translate them (when possible)
into concrete queries for specific storage backends.

This intentionally only uses built-in dataclasses, not pydantic models.
import inspect
from dataclasses import fields
from typing import Any

from .utils import DictView, UnsupportedQueryType

[docs] class QueryRegistry: """ Keep track of all known queries types, with names. When the server starts up, it uses this registry to populate the list of allowed URL query parameters to its /search route. There is a global instance of this, defined below. It is implemented as a class, rather than a module-scope singleton, for the sake of tests. """
[docs] def __init__(self): self._name_to_query_type_type = {} self._query_type_to_name = {}
@property def name_to_query_type(self): return DictView(self._name_to_query_type_type) @property def query_type_to_name(self): return DictView(self._query_type_to_name) def register(self, name=None, overwrite=False, must_revalidate=True): """ Register a new type of query. """ if "___" in name: raise Exception("Names must not contain triple underscores ('___').") # Why? This would create ambiguity in the server's handling of # search requests. Route signature parameters are named like # "filter___{name}___{field}". def inner(cls): if (name in self._name_to_query_type_type) and (not overwrite): if self._name_to_query_type_type[name] is cls: # redundant registration; do nothing return raise Exception( f"The class {self._name_to_query_type_type[name]} is registered to the " f"name {name}. To overwrite, set overwrite=True." ) if cls in self._name_to_query_type_type.values(): raise Exception( f"The class {cls} is already registered by another name." ) for field in fields(cls): if "___" in raise Exception( "Fields must not contain triple underscores ('___')." ) # Why? This would create ambiguity in the server's handling of # search requests. Route signature parameters are named like # "filter___{name}___{field}". if cls in self._query_type_to_name: raise Exception( f"The type {cls} is already registered to the name " f"{self._query_type_to_name[cls]} and cannot also be " f"registered to {name}." ) self._name_to_query_type_type[name] = cls self._query_type_to_name[cls] = name return cls return inner
# Make a global registry. query_registry = QueryRegistry() register = query_registry.register """Register a new type of query.""" class QueryTranslationRegistry: def __init__(self) -> None: self._lookup = {} self._lazy = {} def register(self, class_, translator) -> Any: self._lookup[class_] = translator return translator def register_lazy(self, toplevel, register): """ Register a registration function which will be called if the *toplevel* module (e.g. 'pandas') is ever loaded. """ self._lazy[toplevel] = register def dispatch(self, class_): """Return the function implementation for the given ``class_``""" # Fast path with direct lookup on cls lk = self._lookup try: impl = lk[class_] except KeyError: pass else: return impl # Is a lazy registration function present? toplevel, _, _ = class_.__module__.partition(".") try: register = self._lazy.pop(toplevel) except KeyError: pass else: register() return self.dispatch(class_) # recurse # Walk the MRO and cache the lookup result for base in inspect.getmro(class_)[1:]: if base in lk: lk[class_] = lk[base] return lk[base] raise UnsupportedQueryType(class_.__name__) def __call__(self, arg, *args, **kwargs): """ Call the corresponding method based on type of argument. """ meth = self.dispatch(type(arg)) return meth(arg, *args, **kwargs)