mongo

This submodule implements the MongoTransformer, which takes the parsed filter and converts it to a valid pymongo/BSON query.

__all__ = ('MongoTransformer') module-attribute

BadRequest

Bases: OptimadeHTTPException

400 Bad Request

Source code in optimade/exceptions.py

class BadRequest(OptimadeHTTPException):
    """400 Bad Request"""

    status_code: int = 400
    title: str = "Bad Request"

BaseTransformer

Bases: Transformer, ABC

Generic filter transformer that handles various parts of the grammar in a backend non-specific way.

Attributes:

Name	Type	Description
operator_map	dict[str, Optional[str]]	A map from comparison operators to their backend-specific versions.
mapper	Optional[type[BaseResourceMapper]]	A resource mapper object that defines the expected fields and acts as a container for various field-related configuration.

Source code in optimade/filtertransformers/base_transformer.py

class BaseTransformer(Transformer, abc.ABC):
    """Generic filter transformer that handles various
    parts of the grammar in a backend non-specific way.

    Attributes:
        operator_map: A map from comparison operators
            to their backend-specific versions.
        mapper: A resource mapper object that defines the
            expected fields and acts as a container for
            various field-related configuration.

    """

    mapper: Optional[type[BaseResourceMapper]] = None
    operator_map: dict[str, Optional[str]] = {
        "<": None,
        "<=": None,
        ">": None,
        ">=": None,
        "!=": None,
        "=": None,
    }

    # map from operators to their syntactic (as opposed to logical) inverse to handle
    # equivalence between cases like "A > 3" and "3 < A".
    _reversed_operator_map = {
        ">": "<",
        ">=": "<=",
        "<": ">",
        "<=": ">=",
        "=": "=",
        "!=": "!=",
    }

    _quantity_type: type[Quantity] = Quantity
    _quantities = None

    def __init__(
        self, mapper: Optional[type[BaseResourceMapper]] = None
    ):  # pylint: disable=super-init-not-called
        """Initialise the transformer object, optionally loading in a
        resource mapper for use when post-processing.

        """
        self.mapper = mapper

    @property
    def backend_mapping(self) -> dict[str, Quantity]:
        """A mapping between backend field names (aliases) and the corresponding
        [`Quantity`][optimade.filtertransformers.base_transformer.Quantity] object.
        """
        return {
            quantity.backend_field: quantity for _, quantity in self.quantities.items()  # type: ignore[misc]
        }

    @property
    def quantities(self) -> dict[str, Quantity]:
        """A mapping from the OPTIMADE field name to the corresponding
        [`Quantity`][optimade.filtertransformers.base_transformer.Quantity] objects.
        """
        if self._quantities is None:
            self._quantities = self._build_quantities()

        return self._quantities

    @quantities.setter
    def quantities(self, quantities: dict[str, Quantity]) -> None:
        self._quantities = quantities

    def _build_quantities(self) -> dict[str, Quantity]:
        """Creates a dictionary of field names mapped to
        [`Quantity`][optimade.filtertransformers.base_transformer.Quantity] objects from the
        fields registered by the mapper.

        """

        quantities = {}

        if self.mapper is not None:
            for field in self.mapper.ALL_ATTRIBUTES:
                alias = self.mapper.get_backend_field(field)
                # Allow length aliases to be defined relative to either backend fields or OPTIMADE fields,
                # with preference for those defined from OPTIMADE fields
                length_alias = self.mapper.length_alias_for(
                    field
                ) or self.mapper.length_alias_for(alias)

                if field not in quantities:
                    quantities[field] = self._quantity_type(
                        name=field, backend_field=alias
                    )

                if length_alias:
                    if length_alias not in quantities:
                        quantities[length_alias] = self._quantity_type(
                            name=length_alias,
                            backend_field=self.mapper.get_backend_field(length_alias),
                        )
                    quantities[field].length_quantity = quantities[length_alias]

        return quantities

    def postprocess(self, query) -> Any:
        """Post-process the query according to the rules defined for
        the backend, returning the backend-specific query.

        """
        return query

    def transform(self, tree: Tree) -> Any:
        """Transform the query using the Lark `Transformer` then run the
        backend-specific post-processing methods.

        """
        return self.postprocess(super().transform(tree))

    def __default__(self, data, children, meta):
        """The default rule to call when no definition is found for a particular construct."""
        raise NotImplementedError(
            f"Calling __default__, i.e., unknown grammar concept. data: {data}, children: {children}, meta: {meta}"
        )

    def filter(self, arg):
        """filter: expression*"""
        return arg[0] if arg else None

    @v_args(inline=True)
    def constant(self, value):
        """constant: string | number"""
        # Note: Return as is.
        return value

    @v_args(inline=True)
    def value(self, value):
        """value: string | number | property"""
        # Note: Return as is.
        return value

    @v_args(inline=True)
    def non_string_value(self, value):
        """non_string_value: number | property"""
        # Note: Return as is.
        return value

    @v_args(inline=True)
    def not_implemented_string(self, value):
        """not_implemented_string: value

        Raises:
            NotImplementedError: For further information, see Materials-Consortia/OPTIMADE issue 157:
                https://github.com/Materials-Consortia/OPTIMADE/issues/157

        """
        raise NotImplementedError("Comparing strings is not yet implemented.")

    def property(self, args: list) -> Any:
        """property: IDENTIFIER ( "." IDENTIFIER )*

        If this transformer has an associated mapper, the property
        will be compared to possible relationship entry types and
        for any supported provider prefixes. If there is a match,
        this rule will return a string and not a dereferenced
        [`Quantity`][optimade.filtertransformers.base_transformer.Quantity].

        Raises:
            BadRequest: If the property does not match any
                of the above rules.

        """
        quantity_name = str(args[0])

        # If the quantity name matches an entry type (indicating a relationship filter)
        # then simply return the quantity name; the inherited property
        # must then handle any further nested identifiers
        if self.mapper:
            if quantity_name in self.mapper.RELATIONSHIP_ENTRY_TYPES:
                return quantity_name

        if self.quantities and quantity_name not in self.quantities:
            # If the quantity is provider-specific, but does not match this provider,
            # then return the quantity name such that it can be treated as unknown.
            # If the prefix does not match another known provider, also emit a warning
            # If the prefix does match a known provider, do not return a warning.
            # Following [Handling unknown property names](https://github.com/Materials-Consortia/OPTIMADE/blob/master/optimade.rst#handling-unknown-property-names)
            if self.mapper and quantity_name.startswith("_"):
                prefix = quantity_name.split("_")[1]
                if prefix not in self.mapper.SUPPORTED_PREFIXES:
                    if prefix not in self.mapper.KNOWN_PROVIDER_PREFIXES:
                        warnings.warn(
                            UnknownProviderProperty(
                                f"Field {quantity_name!r} has an unrecognised prefix: this property has been treated as UNKNOWN."
                            )
                        )

                    return quantity_name

            raise BadRequest(
                detail=f"'{quantity_name}' is not a known or searchable quantity"
            )

        quantity = self.quantities.get(quantity_name, None)
        if quantity is None:
            quantity = self._quantity_type(name=str(quantity_name))

        return quantity

    @v_args(inline=True)
    def string(self, string):
        """string: ESCAPED_STRING"""
        return string.strip('"')

    @v_args(inline=True)
    def signed_int(self, number):
        """signed_int : SIGNED_INT"""
        return int(number)

    @v_args(inline=True)
    def number(self, number):
        """number: SIGNED_INT | SIGNED_FLOAT"""
        if TYPE_CHECKING:  # pragma: no cover
            type_: Union[type[int], type[float]]

        if number.type == "SIGNED_INT":
            type_ = int
        elif number.type == "SIGNED_FLOAT":
            type_ = float
        return type_(number)

    @v_args(inline=True)
    def comparison(self, value):
        """comparison: constant_first_comparison | property_first_comparison"""
        # Note: Return as is.
        return value

    def value_list(self, arg):
        """value_list: [ OPERATOR ] value ( "," [ OPERATOR ] value )*"""

    def value_zip(self, arg):
        """value_zip: [ OPERATOR ] value ":" [ OPERATOR ] value (":" [ OPERATOR ] value)*"""
        pass

    def value_zip_list(self, arg):
        """value_zip_list: value_zip ( "," value_zip )*"""

    def expression(self, arg):
        """expression: expression_clause ( OR expression_clause )"""

    def expression_clause(self, arg):
        """expression_clause: expression_phrase ( AND expression_phrase )*"""

    def expression_phrase(self, arg):
        """expression_phrase: [ NOT ] ( comparison | "(" expression ")" )"""

    def property_first_comparison(self, arg):
        """property_first_comparison:
        property ( value_op_rhs
                 | known_op_rhs
                 | fuzzy_string_op_rhs
                 | set_op_rhs
                 | set_zip_op_rhs
                 | length_op_rhs )

        """

    def constant_first_comparison(self, arg):
        """constant_first_comparison: constant OPERATOR ( non_string_value | not_implemented_string )"""

    @v_args(inline=True)
    def value_op_rhs(self, operator, value):
        """value_op_rhs: OPERATOR value"""

    def known_op_rhs(self, arg):
        """known_op_rhs: IS ( KNOWN | UNKNOWN )"""

    def fuzzy_string_op_rhs(self, arg):
        """fuzzy_string_op_rhs: CONTAINS value | STARTS [ WITH ] value | ENDS [ WITH ] value"""

    def set_op_rhs(self, arg):
        """set_op_rhs: HAS ( [ OPERATOR ] value | ALL value_list | ANY value_list | ONLY value_list )"""

    def length_op_rhs(self, arg):
        """length_op_rhs: LENGTH [ OPERATOR ] value"""

    def set_zip_op_rhs(self, arg):
        """set_zip_op_rhs: property_zip_addon HAS ( value_zip
        | ONLY value_zip_list
        | ALL value_zip_list
        | ANY value_zip_list )

        """

    def property_zip_addon(self, arg):
        """property_zip_addon: ":" property (":" property)*"""

backend_mapping: dict[str, Quantity] property

A mapping between backend field names (aliases) and the corresponding Quantity object.

quantities: dict[str, Quantity] property writable

A mapping from the OPTIMADE field name to the corresponding Quantity objects.

__default__(data, children, meta)

The default rule to call when no definition is found for a particular construct.

Source code in optimade/filtertransformers/base_transformer.py

def __default__(self, data, children, meta):
    """The default rule to call when no definition is found for a particular construct."""
    raise NotImplementedError(
        f"Calling __default__, i.e., unknown grammar concept. data: {data}, children: {children}, meta: {meta}"
    )

__init__(mapper=None)

Initialise the transformer object, optionally loading in a resource mapper for use when post-processing.

Source code in optimade/filtertransformers/base_transformer.py

def __init__(
    self, mapper: Optional[type[BaseResourceMapper]] = None
):  # pylint: disable=super-init-not-called
    """Initialise the transformer object, optionally loading in a
    resource mapper for use when post-processing.

    """
    self.mapper = mapper

comparison(value)

comparison: constant_first_comparison | property_first_comparison

Source code in optimade/filtertransformers/base_transformer.py

@v_args(inline=True)
def comparison(self, value):
    """comparison: constant_first_comparison | property_first_comparison"""
    # Note: Return as is.
    return value

constant(value)

constant: string | number

Source code in optimade/filtertransformers/base_transformer.py

@v_args(inline=True)
def constant(self, value):
    """constant: string | number"""
    # Note: Return as is.
    return value

constant_first_comparison(arg)

constant_first_comparison: constant OPERATOR ( non_string_value | not_implemented_string )

Source code in optimade/filtertransformers/base_transformer.py

def constant_first_comparison(self, arg):
    """constant_first_comparison: constant OPERATOR ( non_string_value | not_implemented_string )"""

expression(arg)

expression: expression_clause ( OR expression_clause )

Source code in optimade/filtertransformers/base_transformer.py

def expression(self, arg):
    """expression: expression_clause ( OR expression_clause )"""

expression_clause(arg)

expression_clause: expression_phrase ( AND expression_phrase )*

Source code in optimade/filtertransformers/base_transformer.py

def expression_clause(self, arg):
    """expression_clause: expression_phrase ( AND expression_phrase )*"""

expression_phrase(arg)

expression_phrase: [ NOT ] ( comparison | "(" expression ")" )

Source code in optimade/filtertransformers/base_transformer.py

def expression_phrase(self, arg):
    """expression_phrase: [ NOT ] ( comparison | "(" expression ")" )"""

filter(arg)

filter: expression*

Source code in optimade/filtertransformers/base_transformer.py

def filter(self, arg):
    """filter: expression*"""
    return arg[0] if arg else None

fuzzy_string_op_rhs(arg)

fuzzy_string_op_rhs: CONTAINS value | STARTS [ WITH ] value | ENDS [ WITH ] value

Source code in optimade/filtertransformers/base_transformer.py

def fuzzy_string_op_rhs(self, arg):
    """fuzzy_string_op_rhs: CONTAINS value | STARTS [ WITH ] value | ENDS [ WITH ] value"""

known_op_rhs(arg)

known_op_rhs: IS ( KNOWN | UNKNOWN )

Source code in optimade/filtertransformers/base_transformer.py

def known_op_rhs(self, arg):
    """known_op_rhs: IS ( KNOWN | UNKNOWN )"""

length_op_rhs(arg)

length_op_rhs: LENGTH [ OPERATOR ] value

Source code in optimade/filtertransformers/base_transformer.py

def length_op_rhs(self, arg):
    """length_op_rhs: LENGTH [ OPERATOR ] value"""

non_string_value(value)

non_string_value: number | property

Source code in optimade/filtertransformers/base_transformer.py

@v_args(inline=True)
def non_string_value(self, value):
    """non_string_value: number | property"""
    # Note: Return as is.
    return value

not_implemented_string(value)

not_implemented_string: value

Raises:

Type	Description
NotImplementedError	For further information, see Materials-Consortia/OPTIMADE issue 157: https://github.com/Materials-Consortia/OPTIMADE/issues/157

Source code in optimade/filtertransformers/base_transformer.py

@v_args(inline=True)
def not_implemented_string(self, value):
    """not_implemented_string: value

    Raises:
        NotImplementedError: For further information, see Materials-Consortia/OPTIMADE issue 157:
            https://github.com/Materials-Consortia/OPTIMADE/issues/157

    """
    raise NotImplementedError("Comparing strings is not yet implemented.")

number(number)

number: SIGNED_INT | SIGNED_FLOAT

Source code in optimade/filtertransformers/base_transformer.py

@v_args(inline=True)
def number(self, number):
    """number: SIGNED_INT | SIGNED_FLOAT"""
    if TYPE_CHECKING:  # pragma: no cover
        type_: Union[type[int], type[float]]

    if number.type == "SIGNED_INT":
        type_ = int
    elif number.type == "SIGNED_FLOAT":
        type_ = float
    return type_(number)

postprocess(query)

Post-process the query according to the rules defined for the backend, returning the backend-specific query.

Source code in optimade/filtertransformers/base_transformer.py

def postprocess(self, query) -> Any:
    """Post-process the query according to the rules defined for
    the backend, returning the backend-specific query.

    """
    return query

property(args)

property: IDENTIFIER ( "." IDENTIFIER )*

If this transformer has an associated mapper, the property will be compared to possible relationship entry types and for any supported provider prefixes. If there is a match, this rule will return a string and not a dereferenced Quantity.

Raises:

Type	Description
BadRequest	If the property does not match any of the above rules.

Source code in optimade/filtertransformers/base_transformer.py

def property(self, args: list) -> Any:
    """property: IDENTIFIER ( "." IDENTIFIER )*

    If this transformer has an associated mapper, the property
    will be compared to possible relationship entry types and
    for any supported provider prefixes. If there is a match,
    this rule will return a string and not a dereferenced
    [`Quantity`][optimade.filtertransformers.base_transformer.Quantity].

    Raises:
        BadRequest: If the property does not match any
            of the above rules.

    """
    quantity_name = str(args[0])

    # If the quantity name matches an entry type (indicating a relationship filter)
    # then simply return the quantity name; the inherited property
    # must then handle any further nested identifiers
    if self.mapper:
        if quantity_name in self.mapper.RELATIONSHIP_ENTRY_TYPES:
            return quantity_name

    if self.quantities and quantity_name not in self.quantities:
        # If the quantity is provider-specific, but does not match this provider,
        # then return the quantity name such that it can be treated as unknown.
        # If the prefix does not match another known provider, also emit a warning
        # If the prefix does match a known provider, do not return a warning.
        # Following [Handling unknown property names](https://github.com/Materials-Consortia/OPTIMADE/blob/master/optimade.rst#handling-unknown-property-names)
        if self.mapper and quantity_name.startswith("_"):
            prefix = quantity_name.split("_")[1]
            if prefix not in self.mapper.SUPPORTED_PREFIXES:
                if prefix not in self.mapper.KNOWN_PROVIDER_PREFIXES:
                    warnings.warn(
                        UnknownProviderProperty(
                            f"Field {quantity_name!r} has an unrecognised prefix: this property has been treated as UNKNOWN."
                        )
                    )

                return quantity_name

        raise BadRequest(
            detail=f"'{quantity_name}' is not a known or searchable quantity"
        )

    quantity = self.quantities.get(quantity_name, None)
    if quantity is None:
        quantity = self._quantity_type(name=str(quantity_name))

    return quantity

property_first_comparison(arg)

property_first_comparison: property ( value_op_rhs | known_op_rhs | fuzzy_string_op_rhs | set_op_rhs | set_zip_op_rhs | length_op_rhs )

Source code in optimade/filtertransformers/base_transformer.py

def property_first_comparison(self, arg):
    """property_first_comparison:
    property ( value_op_rhs
             | known_op_rhs
             | fuzzy_string_op_rhs
             | set_op_rhs
             | set_zip_op_rhs
             | length_op_rhs )

    """

property_zip_addon(arg)

property_zip_addon: ":" property (":" property)*

Source code in optimade/filtertransformers/base_transformer.py

def property_zip_addon(self, arg):
    """property_zip_addon: ":" property (":" property)*"""

set_op_rhs(arg)

set_op_rhs: HAS ( [ OPERATOR ] value | ALL value_list | ANY value_list | ONLY value_list )

Source code in optimade/filtertransformers/base_transformer.py

def set_op_rhs(self, arg):
    """set_op_rhs: HAS ( [ OPERATOR ] value | ALL value_list | ANY value_list | ONLY value_list )"""

set_zip_op_rhs(arg)

set_zip_op_rhs: property_zip_addon HAS ( value_zip | ONLY value_zip_list | ALL value_zip_list | ANY value_zip_list )

Source code in optimade/filtertransformers/base_transformer.py

def set_zip_op_rhs(self, arg):
    """set_zip_op_rhs: property_zip_addon HAS ( value_zip
    | ONLY value_zip_list
    | ALL value_zip_list
    | ANY value_zip_list )

    """

signed_int(number)

signed_int : SIGNED_INT

Source code in optimade/filtertransformers/base_transformer.py

@v_args(inline=True)
def signed_int(self, number):
    """signed_int : SIGNED_INT"""
    return int(number)

string(string)

string: ESCAPED_STRING

Source code in optimade/filtertransformers/base_transformer.py

@v_args(inline=True)
def string(self, string):
    """string: ESCAPED_STRING"""
    return string.strip('"')

transform(tree)

Transform the query using the Lark Transformer then run the backend-specific post-processing methods.

Source code in optimade/filtertransformers/base_transformer.py

def transform(self, tree: Tree) -> Any:
    """Transform the query using the Lark `Transformer` then run the
    backend-specific post-processing methods.

    """
    return self.postprocess(super().transform(tree))

value(value)

value: string | number | property

Source code in optimade/filtertransformers/base_transformer.py

@v_args(inline=True)
def value(self, value):
    """value: string | number | property"""
    # Note: Return as is.
    return value

value_list(arg)

value_list: [ OPERATOR ] value ( "," [ OPERATOR ] value )*

Source code in optimade/filtertransformers/base_transformer.py

def value_list(self, arg):
    """value_list: [ OPERATOR ] value ( "," [ OPERATOR ] value )*"""

value_op_rhs(operator, value)

value_op_rhs: OPERATOR value

Source code in optimade/filtertransformers/base_transformer.py

@v_args(inline=True)
def value_op_rhs(self, operator, value):
    """value_op_rhs: OPERATOR value"""

value_zip(arg)

value_zip: [ OPERATOR ] value ":" [ OPERATOR ] value (":" [ OPERATOR ] value)*

Source code in optimade/filtertransformers/base_transformer.py

def value_zip(self, arg):
    """value_zip: [ OPERATOR ] value ":" [ OPERATOR ] value (":" [ OPERATOR ] value)*"""
    pass

value_zip_list(arg)

value_zip_list: value_zip ( "," value_zip )*

Source code in optimade/filtertransformers/base_transformer.py

def value_zip_list(self, arg):
    """value_zip_list: value_zip ( "," value_zip )*"""

MongoTransformer

Bases: BaseTransformer

A filter transformer for the MongoDB backend.

Parses a lark tree into a dictionary representation to be used by pymongo or mongomock. Uses post-processing functions to handle some specific edge-cases for MongoDB.

Attributes:

Name	Type	Description
operator_map		A map from comparison operators to the mongoDB specific versions.
inverse_operator_map		A map from operators to their logical inverse.
mapper		A resource mapper object that defines the expected fields and acts as a container for various field-related configuration.

Source code in optimade/filtertransformers/mongo.py

class MongoTransformer(BaseTransformer):
    """A filter transformer for the MongoDB backend.

    Parses a lark tree into a dictionary representation to be
    used by pymongo or mongomock. Uses post-processing functions
    to handle some specific edge-cases for MongoDB.

    Attributes:
        operator_map: A map from comparison operators
            to the mongoDB specific versions.
        inverse_operator_map: A map from operators to their
            logical inverse.
        mapper: A resource mapper object that defines the
            expected fields and acts as a container for
            various field-related configuration.

    """

    operator_map = {
        "<": "$lt",
        "<=": "$lte",
        ">": "$gt",
        ">=": "$gte",
        "!=": "$ne",
        "=": "$eq",
    }

    inverse_operator_map = {
        "$lt": "$gte",
        "$lte": "$gt",
        "$gt": "$lte",
        "$gte": "$lt",
        "$ne": "$eq",
        "$eq": "$ne",
        "$in": "$nin",
        "$nin": "$in",
    }

    def postprocess(self, query: dict[str, Any]):
        """Used to post-process the nested dictionary of the parsed query."""
        query = self._apply_relationship_filtering(query)
        query = self._apply_length_operators(query)
        query = self._apply_unknown_or_null_filter(query)
        query = self._apply_has_only_filter(query)
        query = self._apply_mongo_id_filter(query)
        query = self._apply_mongo_date_filter(query)
        return query

    def value_list(self, arg):
        # value_list: [ OPERATOR ] value ( "," [ OPERATOR ] value )*
        # NOTE: no support for optional OPERATOR, yet, so this takes the
        # parsed values and returns an error if that is being attempted
        for value in arg:
            if str(value) in self.operator_map.keys():
                raise NotImplementedError(
                    f"OPERATOR {value} inside value_list {arg} not implemented."
                )

        return arg

    def value_zip(self, arg):
        # value_zip: [ OPERATOR ] value ":" [ OPERATOR ] value (":" [ OPERATOR ] value)*
        raise NotImplementedError("Correlated list queries are not supported.")

    def value_zip_list(self, arg):
        # value_zip_list: value_zip ( "," value_zip )*
        raise NotImplementedError("Correlated list queries are not supported.")

    def expression(self, arg):
        # expression: expression_clause ( OR expression_clause )
        # expression with and without 'OR'
        return {"$or": arg} if len(arg) > 1 else arg[0]

    def expression_clause(self, arg):
        # expression_clause: expression_phrase ( AND expression_phrase )*
        # expression_clause with and without 'AND'
        return {"$and": arg} if len(arg) > 1 else arg[0]

    def expression_phrase(self, arg):
        # expression_phrase: [ NOT ] ( comparison | "(" expression ")" )
        return self._recursive_expression_phrase(arg)

    @v_args(inline=True)
    def property_first_comparison(self, quantity, query):
        # property_first_comparison: property ( value_op_rhs | known_op_rhs | fuzzy_string_op_rhs | set_op_rhs |
        # set_zip_op_rhs | length_op_rhs )

        # Awkwardly, MongoDB will match null fields in $ne filters,
        # so we need to add a check for null equality in evey $ne query.
        if "$ne" in query:
            return {"$and": [{quantity: query}, {quantity: {"$ne": None}}]}

        # Check if a $size query is being made (indicating a length_op_rhs filter); if so, check for
        # a defined length alias to replace the $size call with the corresponding filter on the
        # length quantity then carefully merge the two queries.
        #
        # e.g. `("elements", {"$size": 2, "$all": ["Ag", "Au"]})` should become
        # `{"elements": {"$all": ["Ag", "Au"]}, "nelements": 2}` if the `elements` -> `nelements`
        # length alias is defined.
        if "$size" in query:
            if (
                getattr(self.backend_mapping.get(quantity), "length_quantity", None)
                is not None
            ):
                size_query = {
                    self.backend_mapping[  # type: ignore[union-attr]
                        quantity
                    ].length_quantity.backend_field: query.pop("$size")
                }

                final_query = {}
                if query:
                    final_query = {quantity: query}
                for q in size_query:
                    if q in final_query:
                        final_query[q].update(size_query[q])
                    else:
                        final_query[q] = size_query[q]

                return final_query

        return {quantity: query}

    def constant_first_comparison(self, arg):
        # constant_first_comparison: constant OPERATOR ( non_string_value | not_implemented_string )
        return self.property_first_comparison(
            arg[2], {self.operator_map[self._reversed_operator_map[arg[1]]]: arg[0]}
        )

    @v_args(inline=True)
    def value_op_rhs(self, operator, value):
        # value_op_rhs: OPERATOR value
        return {self.operator_map[operator]: value}

    def known_op_rhs(self, arg):
        # known_op_rhs: IS ( KNOWN | UNKNOWN )
        # The OPTIMADE spec also required a type comparison with null, this must be post-processed
        # so here we use a special key "#known" which will get replaced in post-processing with the
        # expanded dict
        return {"#known": arg[1] == "KNOWN"}

    def fuzzy_string_op_rhs(self, arg):
        # fuzzy_string_op_rhs: CONTAINS value | STARTS [ WITH ] value | ENDS [ WITH ] value

        # The WITH keyword may be omitted.
        if isinstance(arg[1], Token) and arg[1].type == "WITH":
            pattern = arg[2]
        else:
            pattern = arg[1]

        # CONTAINS
        if arg[0] == "CONTAINS":
            regex = f"{pattern}"
        elif arg[0] == "STARTS":
            regex = f"^{pattern}"
        elif arg[0] == "ENDS":
            regex = f"{pattern}$"
        return {"$regex": regex}

    def set_op_rhs(self, arg):
        # set_op_rhs: HAS ( [ OPERATOR ] value | ALL value_list | ANY value_list | ONLY value_list )

        if len(arg) == 2:
            # only value without OPERATOR
            return {"$in": arg[1:]}

        if arg[1] == "ALL":
            return {"$all": arg[2]}

        if arg[1] == "ANY":
            return {"$in": arg[2]}

        if arg[1] == "ONLY":
            return {"#only": arg[2]}

        # value with OPERATOR
        raise NotImplementedError(
            f"set_op_rhs not implemented for use with OPERATOR. Given: {arg}"
        )

    def property(self, args):
        # property: IDENTIFIER ( "." IDENTIFIER )*
        quantity = super().property(args)
        if isinstance(quantity, Quantity):
            quantity = quantity.backend_field

        return ".".join([quantity] + args[1:])

    def length_op_rhs(self, arg):
        # length_op_rhs: LENGTH [ OPERATOR ] value
        if len(arg) == 2 or (len(arg) == 3 and arg[1] == "="):
            return {"$size": arg[-1]}

        if arg[1] in self.operator_map and arg[1] != "!=":
            # create an invalid query that needs to be post-processed
            # e.g. {'$size': {'$gt': 2}}, which is not allowed by Mongo.
            return {"$size": {self.operator_map[arg[1]]: arg[-1]}}

        raise NotImplementedError(
            f"Operator {arg[1]} not implemented for LENGTH filter."
        )

    def set_zip_op_rhs(self, arg):
        # set_zip_op_rhs: property_zip_addon HAS ( value_zip | ONLY value_zip_list | ALL value_zip_list |
        # ANY value_zip_list )
        raise NotImplementedError("Correlated list queries are not supported.")

    def property_zip_addon(self, arg):
        # property_zip_addon: ":" property (":" property)*
        raise NotImplementedError("Correlated list queries are not supported.")

    def _recursive_expression_phrase(self, arg: list) -> dict[str, Any]:
        """Helper function for parsing `expression_phrase`. Recursively sorts out
        the correct precedence for `$not`, `$and` and `$or`.

        Parameters:
            arg: A list containing the expression to be evaluated and whether it
                is negated, e.g., `["NOT", expr]` or just `[expr]`.

        Returns:
             The evaluated filter as a nested dictionary.

        """

        def handle_not_and(arg: dict[str, list]) -> dict[str, list]:
            """Handle the case of `~(A & B) -> (~A | ~B)`.

            We have to check for the special case in which the "and" was created
            by a previous NOT, e.g.,
            `NOT (NOT ({"a": {"$eq": 6}})) -> NOT({"$and": [{"a": {"$ne": 6}},{"a": {"$ne": None}}]})`

            Parameters:
                arg: A dictionary with key `"$and"` containing a list of expressions.

            Returns:
                A dictionary with key `"$or"` containing a list of the appropriate negated expressions.
            """

            expr1 = arg["$and"][0]
            expr2 = arg["$and"][1]
            if expr1.keys() == expr2.keys():
                key = list(expr1.keys())[0]
                for e, f in itertools.permutations((expr1, expr2)):
                    if e.get(key) == {"$ne": None}:
                        return self._recursive_expression_phrase(["NOT", f])

            return {
                "$or": [
                    self._recursive_expression_phrase(["NOT", subdict])
                    for subdict in arg["$and"]
                ]
            }

        def handle_not_or(arg: dict[str, list]) -> dict[str, list]:
            """Handle the case of ~(A | B) -> (~A & ~B).

            !!! note
            Although the MongoDB `$nor` could be used here, it is not convenient as it
            will also return documents where the filtered field is missing when testing
            for inequality.

            Parameters:
                arg: A dictionary with key `"$or"` containing a list of expressions.

            Returns:
                A dictionary with key `"$and"` that lists the appropriate negated expressions.
            """

            return {
                "$and": [
                    self._recursive_expression_phrase(["NOT", subdict])
                    for subdict in arg["$or"]
                ]
            }

        if len(arg) == 1:
            # without NOT
            return arg[0]

        if "$or" in arg[1]:
            return handle_not_or(arg[1])

        if "$and" in arg[1]:
            return handle_not_and(arg[1])

        prop, expr = next(iter(arg[1].items()))
        operator, value = next(iter(expr.items()))
        if operator == "$not":  # Case of double negation e.g. NOT("$not":{ ...})
            return {prop: value}

        # If the NOT operator occurs at the lowest nesting level,
        # the expression can be simplified by using the opposite operator and removing the not.
        if operator in self.inverse_operator_map:
            filter_ = {prop: {self.inverse_operator_map[operator]: value}}
            if operator in ("$in", "$eq"):
                filter_ = {"$and": [filter_, {prop: {"$ne": None}}]}  # type: ignore[dict-item]
            return filter_

        filter_ = {prop: {"$not": expr}}
        if "#known" in expr:
            return filter_
        return {"$and": [filter_, {prop: {"$ne": None}}]}

    def _apply_length_operators(self, filter_: dict) -> dict:
        """Check for any invalid pymongo queries that involve applying a
        comparison operator to the length of a field, and transform
        them into a test for existence of the relevant entry, e.g.
        "list LENGTH > 3" becomes "does the 4th list entry exist?".

        """

        def check_for_length_op_filter(_, expr):
            return (
                isinstance(expr, dict)
                and "$size" in expr
                and isinstance(expr["$size"], dict)
            )

        def apply_length_op(subdict, prop, expr):
            # assumes that the dictionary only has one element by design
            # (we just made it above in the transformer)
            operator, value = list(expr["$size"].items())[0]
            if operator in self.operator_map.values() and operator != "$ne":
                # worth being explicit here, I think
                _prop = None
                existence = None
                if operator == "$gt":
                    _prop = f"{prop}.{value + 1}"
                    existence = True
                elif operator == "$gte":
                    _prop = f"{prop}.{value}"
                    existence = True
                elif operator == "$lt":
                    _prop = f"{prop}.{value}"
                    existence = False
                elif operator == "$lte":
                    _prop = f"{prop}.{value + 1}"
                    existence = False
                if _prop is not None:
                    subdict.pop(prop)
                    subdict[_prop] = {"$exists": existence}

            return subdict

        return recursive_postprocessing(
            filter_,
            check_for_length_op_filter,
            apply_length_op,
        )

    def _apply_relationship_filtering(self, filter_: dict) -> dict:
        """Check query for property names that match the entry
        types, and transform them as relationship filters rather than
        property filters.

        """

        def check_for_entry_type(prop, _):
            return str(prop).count(".") == 1 and str(prop).split(".")[0] in (
                "structures",
                "references",
            )

        def replace_with_relationship(subdict, prop, expr):
            _prop, _field = str(prop).split(".")
            if _field != "id":
                raise NotImplementedError(
                    f'Cannot filter relationships by field "{_field}", only "id" is supported.'
                )

            subdict[f"relationships.{_prop}.data.{_field}"] = expr
            subdict.pop(prop)
            return subdict

        return recursive_postprocessing(
            filter_, check_for_entry_type, replace_with_relationship
        )

    def _apply_has_only_filter(self, filter_: dict) -> dict:
        """This method loops through the query and replaces the magic key `"#only"`
        with the proper 'HAS ONLY' query.
        """

        def check_for_only_filter(_, expr):
            """Find cases where the magic key `"#only"` is in the query."""
            return isinstance(expr, dict) and ("#only" in expr)

        def replace_only_filter(subdict: dict, prop: str, expr: dict):
            """Replace the magic key `"#only"` (added by this transformer) with an `$elemMatch`-based query.

            The first part of the query selects all the documents that contain any value that does not
            match any target values for the property `prop`.
            Subsequently, this selection is inverted, to get the documents that only have
            the allowed values.
            This inversion also selects documents with edge-case values such as null or empty lists;
            these are removed in the second part of the query that makes sure that only documents
            with lists that have at least one value are selected.

            """

            if "$and" not in subdict:
                subdict["$and"] = []

            if prop.startswith("relationships."):
                if prop not in (
                    "relationships.references.data.id",
                    "relationships.structures.data.id",
                ):
                    raise BadRequest(f"Unable to query on unrecognised field {prop}.")
                first_part_prop = ".".join(prop.split(".")[:-1])
                subdict["$and"].append(
                    {
                        first_part_prop: {
                            "$not": {"$elemMatch": {"id": {"$nin": expr["#only"]}}}
                        }
                    }
                )
                subdict["$and"].append({first_part_prop + ".0": {"$exists": True}})

            else:
                subdict["$and"].append(
                    {prop: {"$not": {"$elemMatch": {"$nin": expr["#only"]}}}}
                )
                subdict["$and"].append({prop + ".0": {"$exists": True}})

            subdict.pop(prop)
            return subdict

        return recursive_postprocessing(
            filter_, check_for_only_filter, replace_only_filter
        )

    def _apply_unknown_or_null_filter(self, filter_: dict) -> dict:
        """This method loops through the query and replaces the check for
        KNOWN with a check for existence and a check for not null, and the
        inverse for UNKNOWN.

        """

        def check_for_known_filter(_, expr):
            """Find cases where the query dict looks like
            `{"field": {"#known": T/F}}` or
            `{"field": "$not": {"#known": T/F}}`, which is a magic word
            for KNOWN/UNKNOWN filters in this transformer.

            """
            return isinstance(expr, dict) and (
                "#known" in expr or "#known" in expr.get("$not", {})
            )

        def replace_known_filter_with_or(subdict, prop, expr):
            """Replace magic key `"#known"` (added by this transformer) with the appropriate
            combination of `$exists` and/or test for nullity.
            combination of $exists and/or $eq/$ne null.

            """
            not_ = set(expr.keys()) == {"$not"}
            if not_:
                expr = expr["$not"]

            exists = expr["#known"] ^ not_

            top_level_key = "$or"
            comparison_operator = "$eq"
            if exists:
                top_level_key = "$and"
                comparison_operator = "$ne"

            if top_level_key not in subdict:
                subdict[top_level_key] = []

            subdict[top_level_key].append({prop: {"$exists": exists}})
            subdict[top_level_key].append({prop: {comparison_operator: None}})

            subdict.pop(prop)

            return subdict

        return recursive_postprocessing(
            filter_, check_for_known_filter, replace_known_filter_with_or
        )

    def _apply_mongo_id_filter(self, filter_: dict) -> dict:
        """This method loops through the query and replaces any operations
        on the special Mongodb `_id` key with the corresponding operation
        on a BSON `ObjectId` type.
        """

        def check_for_id_key(prop, _):
            """Find cases where the query dict is operating on the `_id` field."""
            return prop == "_id"

        def replace_str_id_with_objectid(subdict, prop, expr):
            from bson import ObjectId

            for operator in subdict[prop]:
                val = subdict[prop][operator]
                if operator not in ("$eq", "$ne"):
                    if self.mapper is not None:
                        prop = self.mapper.get_optimade_field(prop)
                    raise NotImplementedError(
                        f"Operator {operator} not supported for query on field {prop!r}, can only test for equality"
                    )
                if isinstance(val, str):
                    subdict[prop][operator] = ObjectId(val)
            return subdict

        return recursive_postprocessing(
            filter_, check_for_id_key, replace_str_id_with_objectid
        )

    def _apply_mongo_date_filter(self, filter_: dict) -> dict:
        """This method loops through the query and replaces any operations
        on suspected timestamp properties with the corresponding operation
        on a BSON `DateTime` type.
        """

        def check_for_timestamp_field(prop, _):
            """Find cases where the query dict is operating on a timestamp field."""
            if self.mapper is not None:
                prop = self.mapper.get_optimade_field(prop)
            return prop == "last_modified"

        def replace_str_date_with_datetime(subdict, prop, expr):
            """Encode suspected dates in with BSON."""
            import bson.json_util

            for operator in subdict[prop]:
                query_datetime = bson.json_util.loads(
                    bson.json_util.dumps({"$date": subdict[prop][operator]}),
                    json_options=bson.json_util.DEFAULT_JSON_OPTIONS.with_options(
                        tz_aware=True, tzinfo=bson.tz_util.utc
                    ),
                )
                if query_datetime.microsecond != 0:
                    warnings.warn(
                        f"Query for timestamp {subdict[prop][operator]!r} for field {prop!r} contained microseconds, which is not RFC3339 compliant. "
                        "This may cause undefined behaviour for the underlying database.",
                        TimestampNotRFCCompliant,
                    )

                subdict[prop][operator] = query_datetime

            return subdict

        return recursive_postprocessing(
            filter_, check_for_timestamp_field, replace_str_date_with_datetime
        )

backend_mapping: dict[str, Quantity] property

A mapping between backend field names (aliases) and the corresponding Quantity object.

quantities: dict[str, Quantity] property writable

A mapping from the OPTIMADE field name to the corresponding Quantity objects.

__default__(data, children, meta)

The default rule to call when no definition is found for a particular construct.

Source code in optimade/filtertransformers/base_transformer.py

def __default__(self, data, children, meta):
    """The default rule to call when no definition is found for a particular construct."""
    raise NotImplementedError(
        f"Calling __default__, i.e., unknown grammar concept. data: {data}, children: {children}, meta: {meta}"
    )

__init__(mapper=None)

Initialise the transformer object, optionally loading in a resource mapper for use when post-processing.

Source code in optimade/filtertransformers/base_transformer.py

def __init__(
    self, mapper: Optional[type[BaseResourceMapper]] = None
):  # pylint: disable=super-init-not-called
    """Initialise the transformer object, optionally loading in a
    resource mapper for use when post-processing.

    """
    self.mapper = mapper

comparison(value)

comparison: constant_first_comparison | property_first_comparison

Source code in optimade/filtertransformers/base_transformer.py

@v_args(inline=True)
def comparison(self, value):
    """comparison: constant_first_comparison | property_first_comparison"""
    # Note: Return as is.
    return value

constant(value)

constant: string | number

Source code in optimade/filtertransformers/base_transformer.py

@v_args(inline=True)
def constant(self, value):
    """constant: string | number"""
    # Note: Return as is.
    return value

filter(arg)

filter: expression*

Source code in optimade/filtertransformers/base_transformer.py

def filter(self, arg):
    """filter: expression*"""
    return arg[0] if arg else None

non_string_value(value)

non_string_value: number | property

Source code in optimade/filtertransformers/base_transformer.py

@v_args(inline=True)
def non_string_value(self, value):
    """non_string_value: number | property"""
    # Note: Return as is.
    return value

not_implemented_string(value)

not_implemented_string: value

Raises:

Type	Description
NotImplementedError	For further information, see Materials-Consortia/OPTIMADE issue 157: https://github.com/Materials-Consortia/OPTIMADE/issues/157

Source code in optimade/filtertransformers/base_transformer.py

@v_args(inline=True)
def not_implemented_string(self, value):
    """not_implemented_string: value

    Raises:
        NotImplementedError: For further information, see Materials-Consortia/OPTIMADE issue 157:
            https://github.com/Materials-Consortia/OPTIMADE/issues/157

    """
    raise NotImplementedError("Comparing strings is not yet implemented.")

number(number)

number: SIGNED_INT | SIGNED_FLOAT

Source code in optimade/filtertransformers/base_transformer.py

@v_args(inline=True)
def number(self, number):
    """number: SIGNED_INT | SIGNED_FLOAT"""
    if TYPE_CHECKING:  # pragma: no cover
        type_: Union[type[int], type[float]]

    if number.type == "SIGNED_INT":
        type_ = int
    elif number.type == "SIGNED_FLOAT":
        type_ = float
    return type_(number)

postprocess(query)

Used to post-process the nested dictionary of the parsed query.

Source code in optimade/filtertransformers/mongo.py

def postprocess(self, query: dict[str, Any]):
    """Used to post-process the nested dictionary of the parsed query."""
    query = self._apply_relationship_filtering(query)
    query = self._apply_length_operators(query)
    query = self._apply_unknown_or_null_filter(query)
    query = self._apply_has_only_filter(query)
    query = self._apply_mongo_id_filter(query)
    query = self._apply_mongo_date_filter(query)
    return query

signed_int(number)

signed_int : SIGNED_INT

Source code in optimade/filtertransformers/base_transformer.py

@v_args(inline=True)
def signed_int(self, number):
    """signed_int : SIGNED_INT"""
    return int(number)

string(string)

string: ESCAPED_STRING

Source code in optimade/filtertransformers/base_transformer.py

@v_args(inline=True)
def string(self, string):
    """string: ESCAPED_STRING"""
    return string.strip('"')

transform(tree)

Transform the query using the Lark Transformer then run the backend-specific post-processing methods.

Source code in optimade/filtertransformers/base_transformer.py

def transform(self, tree: Tree) -> Any:
    """Transform the query using the Lark `Transformer` then run the
    backend-specific post-processing methods.

    """
    return self.postprocess(super().transform(tree))

value(value)

value: string | number | property

Source code in optimade/filtertransformers/base_transformer.py

@v_args(inline=True)
def value(self, value):
    """value: string | number | property"""
    # Note: Return as is.
    return value

Quantity

Class to provide information about available quantities to the transformer.

The transformer can use Quantity's to

• do some semantic checks,
• map quantities to the underlying backend field name.

Attributes:

Name	Type	Description
name	str	The name of the quantity as used in the filter expressions.
backend_field	Optional[str]	The name of the field for this quantity in the backend database, will be name by default.
length_quantity	Optional[Quantity]	Another (typically integer) Quantity that can be queried as the length of this quantity, e.g. elements and nelements. Backends can then decide whether to use this for all "LENGTH" queries.

Source code in optimade/filtertransformers/base_transformer.py

class Quantity:
    """Class to provide information about available quantities to the transformer.

    The transformer can use [`Quantity`][optimade.filtertransformers.base_transformer.Quantity]'s to

    * do some semantic checks,
    * map quantities to the underlying backend field name.

    Attributes:
        name: The name of the quantity as used in the filter expressions.
        backend_field: The name of the field for this quantity in the backend database, will be
            `name` by default.
        length_quantity: Another (typically integer) [`Quantity`][optimade.filtertransformers.base_transformer.Quantity]
            that can be queried as the length of this quantity, e.g. `elements` and `nelements`. Backends
            can then decide whether to use this for all "LENGTH" queries.

    """

    name: str
    backend_field: Optional[str]
    length_quantity: Optional["Quantity"]

    def __init__(
        self,
        name: str,
        backend_field: Optional[str] = None,
        length_quantity: Optional["Quantity"] = None,
    ):
        """Initialise the `quantity` from it's name and aliases.

        Parameters:
            name: The name of the quantity as used in the filter expressions.
            backend_field: The name of the field for this quantity in the backend database, will be
                `name` by default.
            length_quantity: Another (typically integer) [`Quantity`][optimade.filtertransformers.base_transformer.Quantity]
                that can be queried as the length of this quantity, e.g. `elements` and `nelements`. Backends
                can then decide whether to use this for all "LENGTH" queries.

        """

        self.name = name
        self.backend_field = backend_field if backend_field is not None else name
        self.length_quantity = length_quantity

__init__(name, backend_field=None, length_quantity=None)

Initialise the quantity from it's name and aliases.

Parameters:

Name	Type	Description	Default
name	str	The name of the quantity as used in the filter expressions.	required
backend_field	Optional[str]	The name of the field for this quantity in the backend database, will be name by default.	None
length_quantity	Optional[Quantity]	Another (typically integer) Quantity that can be queried as the length of this quantity, e.g. elements and nelements. Backends can then decide whether to use this for all "LENGTH" queries.	None

Source code in optimade/filtertransformers/base_transformer.py

def __init__(
    self,
    name: str,
    backend_field: Optional[str] = None,
    length_quantity: Optional["Quantity"] = None,
):
    """Initialise the `quantity` from it's name and aliases.

    Parameters:
        name: The name of the quantity as used in the filter expressions.
        backend_field: The name of the field for this quantity in the backend database, will be
            `name` by default.
        length_quantity: Another (typically integer) [`Quantity`][optimade.filtertransformers.base_transformer.Quantity]
            that can be queried as the length of this quantity, e.g. `elements` and `nelements`. Backends
            can then decide whether to use this for all "LENGTH" queries.

    """

    self.name = name
    self.backend_field = backend_field if backend_field is not None else name
    self.length_quantity = length_quantity

TimestampNotRFCCompliant

Bases: OptimadeWarning

A timestamp has been used in a filter that contains microseconds and is thus not RFC 3339 compliant. This may cause undefined behaviour in the query results.

Source code in optimade/warnings.py

class TimestampNotRFCCompliant(OptimadeWarning):
    """A timestamp has been used in a filter that contains microseconds and is thus not
    RFC 3339 compliant. This may cause undefined behaviour in the query results.

    """

recursive_postprocessing(filter_, condition, replacement)

Recursively descend into the query, checking each dictionary (contained in a list, or as an entry in another dictionary) for the condition passed. If the condition is true, apply the replacement to the dictionary.

Parameters:

Name	Type	Description	Default
filter_		the filter_ to process.	required
condition	callable	a function that returns True if the replacement function should be applied. It should take as arguments the property and expression from the filter_, as would be returned by iterating over filter_.items().	required
replacement	callable	a function that returns the processed dictionary. It should take as arguments the dictionary to modify, the property and the expression (as described above).	required

Example

For the simple case of replacing one field name with another, the following functions could be used:

def condition(prop, expr):
    return prop == "field_name_old"

def replacement(d, prop, expr):
    d["field_name_old"] = d.pop(prop)

filter_ = recursive_postprocessing(
    filter_, condition, replacement
)

Source code in optimade/filtertransformers/mongo.py

def recursive_postprocessing(filter_: Union[dict, list], condition, replacement):
    """Recursively descend into the query, checking each dictionary
    (contained in a list, or as an entry in another dictionary) for
    the condition passed. If the condition is true, apply the
    replacement to the dictionary.

    Parameters:
        filter_ : the filter_ to process.
        condition (callable): a function that returns True if the
            replacement function should be applied. It should take
            as arguments the property and expression from the filter_,
            as would be returned by iterating over `filter_.items()`.
        replacement (callable): a function that returns the processed
            dictionary. It should take as arguments the dictionary
            to modify, the property and the expression (as described
            above).

    Example:
        For the simple case of replacing one field name with
        another, the following functions could be used:

        ```python
        def condition(prop, expr):
            return prop == "field_name_old"

        def replacement(d, prop, expr):
            d["field_name_old"] = d.pop(prop)

        filter_ = recursive_postprocessing(
            filter_, condition, replacement
        )

        ```

    """
    if isinstance(filter_, list):
        result = [recursive_postprocessing(q, condition, replacement) for q in filter_]
        return result

    if isinstance(filter_, dict):
        # this could potentially lead to memory leaks if the filter_ is *heavily* nested
        _cached_filter = copy.deepcopy(filter_)
        for prop, expr in filter_.items():
            if condition(prop, expr):
                _cached_filter = replacement(_cached_filter, prop, expr)
            elif isinstance(expr, list):
                _cached_filter[prop] = [
                    recursive_postprocessing(q, condition, replacement) for q in expr
                ]
        return _cached_filter

    return filter_