Skip to content

motorhead.operator

All

Bases: KeyValueOperator

$all MongoDB operator.

Source code in motorhead/operator.py
class All(KeyValueOperator):
    """`$all` MongoDB operator."""

    __slots__ = ()

    def __init__(self, key: str | Field, value: list[Any]) -> None:
        if not isinstance(value, list):
            raise ValueError("All field only accepts list values.")
        super().__init__(key, value)

And

Bases: ClauseOperator

$and MongoDB operator.

Source code in motorhead/operator.py
class And(ClauseOperator):
    """`$and` MongoDB operator."""

    __slots__ = ()

ClauseOperator

Base class for clause sequence based operators.

Source code in motorhead/operator.py
class ClauseOperator:
    """
    Base class for clause sequence based operators.
    """

    __slots__ = ("_clauses",)

    _operator: str = None  # type: ignore[assignment]

    def __init__(self, *clauses: Clause) -> None:
        self._clauses = clauses

    def __init_subclass__(cls) -> None:
        if cls._operator is None:
            cls._operator = f"${cls.__name__.lower()}"

    @property
    def clauses(self) -> Generator[Clause, None, None]:
        for clause in self._clauses:
            yield clause

    def to_mongo(self) -> dict[str, Any]:
        return {self._operator: [c.to_mongo() for c in self._clauses]}

DirectEq

Bases: KeyValueOperator

Plain equality operator, {key: value}.

Source code in motorhead/operator.py
class DirectEq(KeyValueOperator):
    """Plain equality operator, `{key: value}`."""

    __slots__ = ()
    _operator = ""

    def to_mongo(self) -> dict[str, Any]:
        return {self._key: self._value}

ElemMatch

Bases: KeyValueOperator

$elemMatch MongoDB operator.

Source code in motorhead/operator.py
class ElemMatch(KeyValueOperator):
    """`$elemMatch` MongoDB operator."""

    __slots__ = ()
    _operator = "$elemMatch"

    def __init__(self, key: str | Field, value: dict[str, Any]) -> None:
        if not isinstance(value, dict):
            raise ValueError("ElemMatch field only accepts dict values.")
        super().__init__(key, value)

Eq

Bases: KeyValueOperator

$eq MongoDB operator.

Source code in motorhead/operator.py
class Eq(KeyValueOperator):
    """`$eq` MongoDB operator."""

    __slots__ = ()

Exists

Bases: KeyValueOperator

$exists MongoDB operator.

Source code in motorhead/operator.py
class Exists(KeyValueOperator):
    """`$exists` MongoDB operator."""

    __slots__ = ()

    def __init__(self, key: str | Field, value: bool) -> None:
        if not isinstance(value, bool):
            raise ValueError("Exists field only accepts bool values.")

        super().__init__(key, value)

Gt

Bases: KeyValueOperator

$gt MongoDB operator.

Source code in motorhead/operator.py
class Gt(KeyValueOperator):
    """`$gt` MongoDB operator."""

    __slots__ = ()

Gte

Bases: KeyValueOperator

$gte MongoDB operator.

Source code in motorhead/operator.py
class Gte(KeyValueOperator):
    """`$gte` MongoDB operator."""

    __slots__ = ()

In

Bases: KeyValueOperator

$in MongoDB operator.

Source code in motorhead/operator.py
class In(KeyValueOperator):
    """`$in` MongoDB operator."""

    __slots__ = ()

KeyValueOperator

Base class for key-value pair based operators.

Source code in motorhead/operator.py
class KeyValueOperator:
    """
    Base class for key-value pair based operators.
    """

    __slots__ = ("_key", "_value")

    _operator: str = None  # type: ignore[assignment]

    def __init__(self, key: str | Field, value: Any) -> None:
        self._key = key if isinstance(key, str) else key.name
        self._value = value

    def __init_subclass__(cls) -> None:
        if cls._operator is None:
            cls._operator = f"${cls.__name__.lower()}"

    @property
    def key(self) -> str:
        return self._key

    @property
    def value(self) -> Any:
        return self._value

    def to_mongo(self) -> dict[str, Any]:
        return {self._key: {self._operator: self._value}}

Lt

Bases: KeyValueOperator

$lt MongoDB operator.

Source code in motorhead/operator.py
class Lt(KeyValueOperator):
    """`$lt` MongoDB operator."""

    __slots__ = ()

Lte

Bases: KeyValueOperator

$lte MongoDB operator.

Source code in motorhead/operator.py
class Lte(KeyValueOperator):
    """`$lte` MongoDB operator."""

    __slots__ = ()

Ne

Bases: KeyValueOperator

$ne MongoDB operator.

Source code in motorhead/operator.py
class Ne(KeyValueOperator):
    """`$ne` MongoDB operator."""

    __slots__ = ()

Nor

Bases: ClauseOperator

$nor MongoDB operator.

Source code in motorhead/operator.py
class Nor(ClauseOperator):
    """`$nor` MongoDB operator."""

    __slots__ = ()

Not

Bases: ClauseOperator

$not MongoDB operator.

Source code in motorhead/operator.py
class Not(ClauseOperator):
    """`$not` MongoDB operator."""

    __slots__ = ()

NotIn

Bases: KeyValueOperator

$nin MongoDB operator.

Source code in motorhead/operator.py
class NotIn(KeyValueOperator):
    """`$nin` MongoDB operator."""

    __slots__ = ()
    _operator = "$nin"

Or

Bases: ClauseOperator

$or MongoDB operator.

Source code in motorhead/operator.py
class Or(ClauseOperator):
    """`$or` MongoDB operator."""

    __slots__ = ()

Raw

Clause that wraps a raw, MongoDB-compatible dict.

Source code in motorhead/operator.py
class Raw:
    """
    Clause that wraps a raw, MongoDB-compatible dict.
    """

    __slots__ = ("_data",)

    def __init__(self, value: dict[str, Any]) -> None:
        self._data = value

    def to_mongo(self) -> dict[str, Any]:
        return {**self._data}  # Just a shallow copy

Size

Bases: KeyValueOperator

$size MongoDB operator.

Source code in motorhead/operator.py
class Size(KeyValueOperator):
    """`$size` MongoDB operator."""

    __slots__ = ()

    def __init__(self, key: str | Field, value: int) -> None:
        if not isinstance(value, int):
            raise ValueError("Size field only accepts int values.")
        super().__init__(key, value)

Type

Bases: KeyValueOperator

$type MongoDB operator.

Source code in motorhead/operator.py
class Type(KeyValueOperator):
    """`$type` MongoDB operator."""

    __slots__ = ()

    def __init__(self, key: str | Field, value: str) -> None:
        if not isinstance(value, str):
            raise ValueError("Type field only accepts string values.")

        super().__init__(key, value)