Source code for tiled.queries

"""
These objects describe queries, independent of how the actual querying is implemented.
They are used on the client side and the server side.

The are encoded into and decoded from URL query parameters.
"""

import enum
import json
from dataclasses import dataclass
from typing import Any, List

from .query_registration import register
from .structures.core import StructureFamily as StructureFamilyEnum

JSONSerializable = Any  # Feel free to refine this.


class NoBool:
    def __bool__(self):
        raise TypeError(
            """Queries are not "truth-y" or "false-y". They must be passed to search().

You may be seeing this error message because you tried to use a tiled query
with the Python keywords `and` or `or`. This does not (cannot) work.
To compose queries, chain search calls like:

    c.search(...).search(...).search(...)
"""
        )


[docs] @register(name="fulltext") @dataclass class FullText(NoBool): """ Search the full text of all metadata values for word matches. This matches *complete words*, so 'dog' would match 'cat dog elephant', but 'do' would not match. Parameters ---------- text : str """ text: str def encode(self): return {"text": self.text} @classmethod def decode(cls, *, text): return cls(text=text)
[docs] @register(name="lookup") @dataclass class KeyLookup(NoBool): """ Match a specific Entry by key. Mostly for internal use. This is necessary to support item lookup within search results, as in: >>> tree.search(...)["..."] The server handles this directly and generically, simply calling __getitem__ on the tree after apply all other queries. Implementations of search(...) do not need to handle it. Parameters ---------- key : str """ key: str def encode(self): return {"key": self.key} @classmethod def decode(cls, *, key): return cls(key=key)
@register(name="keys_filter") @dataclass class KeysFilter(NoBool): """ Filter entries that do not match one of these keys. This is used by the SimpleAccessPolicy. Parameters ---------- keys : List[str] """ keys: List[str] def encode(self): return {"keys": json.dumps(self.keys)} @classmethod def decode(cls, *, keys): return cls(keys=json.loads(keys))
[docs] @register(name="regex") @dataclass class Regex(NoBool): """ Match a key's value to a regular expression. Parameters ---------- key : str e.g. "color", "sample.name" pattern : str regular expression case_sensitive : bool, optional Default True (case-sensitive). Note that this is the opposite of the default for FullText; regex users generally expect case sensitivity by default. Examples -------- Search for color == "red" >>> c.search(Regex("sample.name", "Cu.*")) """ key: str pattern: str case_sensitive: bool = True def encode(self): return { "key": self.key, "pattern": self.pattern, "case_sensitive": json.dumps(self.case_sensitive), } @classmethod def decode(cls, *, key, pattern, case_sensitive=True): # Note: FastAPI decodes case_sensitive into a boolean for us. return cls( key=key, pattern=pattern, case_sensitive=case_sensitive, )
[docs] @register(name="eq") @dataclass class Eq(NoBool): """ Query equality of a given key's value to the specified value. See `Key` in this module for a more intuitive interface for equality. Parameters ---------- key : str e.g. "color", "sample.name" value : JSONSerializable May be a string, number, list, or dict. Examples -------- Search for color == "red" >>> c.search(Eq("color", "red")) """ key: str value: JSONSerializable def encode(self): return {"key": self.key, "value": json.dumps(self.value)} @classmethod def decode(cls, *, key, value): return cls(key=key, value=json.loads(value))
[docs] @register(name="noteq") @dataclass class NotEq(NoBool): """ Query inequality of a given key's value to the specified value. See `Key` in this module for a more intuitive interface for inequality. Parameters ---------- key : str e.g. "color", "sample.name" value : JSONSerializable May be a string, number, list, or dict. Examples -------- Search for color == "red" >>> c.search(Eq("color", "red")) """ key: str value: JSONSerializable def encode(self): return {"key": self.key, "value": json.dumps(self.value)} @classmethod def decode(cls, *, key, value): return cls(key=key, value=json.loads(value))
class Operator(str, enum.Enum): lt = "lt" gt = "gt" le = "le" ge = "ge"
[docs] @register(name="comparison") @dataclass class Comparison(NoBool): """ Query binary comparison between given key's value to the specified value. See `Key` in this module for a more intuitive interface for comparisons. Parameters ---------- operator : {"gt", "lt", "ge", "le"} key : str e.g. "temperature" value : JSONSerializable May be a string, number, list, or dict. Examples -------- Search for temperature > 300. >>> c.search(Comparison("gt", "temperature", 300)) """ operator: Operator key: str value: JSONSerializable
[docs] def __init__(self, operator, key, value): self.operator = Operator(operator) self.key = key self.value = value
def encode(self): return { "operator": self.operator.value, "key": self.key, "value": json.dumps(self.value), } @classmethod def decode(cls, *, operator, key, value): return cls(operator=Operator(operator), key=key, value=json.loads(value))
[docs] @register(name="contains") @dataclass class Contains(NoBool): """ Query where a given key's value contains the specified value. Parameters ---------- key : str e.g. "motors" value : JSONSerializable May be a string, number, list, or dict. Examples -------- Search for matches where "ccd" is including the list of detectors. >>> c.search(Contains("detectors", "ccd")) """ key: str value: JSONSerializable def encode(self): return {"key": self.key, "value": json.dumps(self.value)} @classmethod def decode(cls, *, key, value): return cls(key=key, value=json.loads(value))
[docs] @register(name="in") @dataclass class In: """ Query if a given key's value is present in the specified sequence of values. Parameters ---------- key : str e.g. "color", "sample.name" value : List[JSONSerializable] e.g. ["red", "blue"] Examples -------- Search for color in ["red", "blue"] >>> c.search(In("color", ["red", "blue"])) """ key: str value: List[JSONSerializable] def __post_init__(self): self.value = list(self.value) def encode(self): return {"key": self.key, "value": json.dumps(self.value)} @classmethod def decode(cls, *, key, value): return cls(key=key, value=json.loads(value))
[docs] @register(name="notin") @dataclass class NotIn: """ Query if a given key's value is not present in the specified sequence of values. Parameters ---------- key : str e.g. "color", "sample.name" value : List[JSONSerializable] e.g. ["red", "blue"] Examples -------- Search for color not in ["red", "blue"] >>> c.search(NotIn("color", ["red", "blue"])) """ key: str value: List[JSONSerializable] def __post_init__(self): self.value = list(self.value) def encode(self): return {"key": self.key, "value": json.dumps(self.value)} @classmethod def decode(cls, *, key, value): return cls(key=key, value=json.loads(value))
[docs] @register(name="specs") @dataclass(init=False) class SpecsQuery: """ Query if specs list matches all elements in include list and does not match any element in exclude list Parameters ---------- include : List[str] exclude : List[str] Examples -------- Search for specs ["foo", "bar"] and NOT "baz" >>> c.search(SpecsQuery(include=["foo", "bar"], exclude=["baz"])) """ include: List[str] exclude: List[str]
[docs] def __init__(self, include, exclude=None): exclude = exclude or [] if isinstance(include, str): raise TypeError("include must be a list not a str") if isinstance(exclude, str): raise TypeError("exclude must be a list not a str") self.include = list(include) self.exclude = list(exclude)
def encode(self): return { "include": json.dumps(self.include), "exclude": json.dumps(self.exclude), } @classmethod def decode(cls, *, include, exclude): return cls(include=json.loads(include), exclude=json.loads(exclude))
[docs] def SpecQuery(spec): """ Convenience function for querying if specs list contains a given spec Equivalent to SpecsQuery([spec]). Parameters ---------- spec: str Examples -------- Search for spec "foo" >>> c.search(Spec("foo")) """ return SpecsQuery([spec])
[docs] @register(name="structure_family") @dataclass(init=False) class StructureFamilyQuery: """ Query if structure_families match value Parameters ---------- value : StructureFamily Examples -------- Search for dataframes >>> c.search(StructureFamilies("dataframe")) """
[docs] def __init__(self, value): self.value = StructureFamilyEnum(value)
value: StructureFamilyEnum def encode(self): return {"value": self.value.value} @classmethod def decode(cls, *, value): return cls(value=value)
[docs] class Key: """ Compare a key in the metadata to a value using standard Python operators. This itself is not a query, but *comparing* it with a value, as shown in the examples below, produces a query. Parameters ---------- key : str Examples -------- Search for equality, comparison, or membership in a collection. >>> c.search(Key("color") == "red") >>> c.search(Key("temperature") >= 300) >>> c.search(Key("temperature") <= 300) >>> c.search(Key("position") > 5.0) >>> c.search(Key("position") < 5.0) """
[docs] def __init__(self, key): self.key = key
def __eq__(self, value): return Eq(self.key, value) def __ne__(self, value): return NotEq(self.key, value) def __lt__(self, value): return Comparison("lt", self.key, value) def __gt__(self, value): return Comparison("gt", self.key, value) def __le__(self, value): return Comparison("le", self.key, value) def __ge__(self, value): return Comparison("ge", self.key, value)
# Note: __contains__ cannot be supported because the language coerces # the result of __contains__ to be a literal boolean. We are not # allowed to return a custom type. class QueryValueError(ValueError): pass