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:
        """
        Initialization.

        Arguments:
            key: The operator's key (the document attribute name).
            value: The operator's value.
        """
        if not isinstance(value, list):
            raise ValueError("All field only accepts list values.")
        super().__init__(key, value)

__init__(key, value)

Initialization.

Parameters:

Name	Type	Description	Default
key	str | Field	The operator's key (the document attribute name).	required
value	list[Any]	The operator's value.	required

Source code in motorhead/operator.py

def __init__(self, key: str | Field, value: list[Any]) -> None:
    """
    Initialization.

    Arguments:
        key: The operator's key (the document attribute name).
        value: The operator's value.
    """
    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:
        """
        Initialization.

        Arguments:
            *clauses: The operator's clauses.
        """
        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]:
        """Generator that yields the operator's clauses."""
        for clause in self._clauses:
            yield clause

    def to_mongo(self) -> dict[str, Any]:
        """Converts the operator to a MongoDB-compatible dict."""
        return {self._operator: [c.to_mongo() for c in self._clauses]}

clauses property

Generator that yields the operator's clauses.

__init__(*clauses)

Initialization.

Parameters:

Name	Type	Description	Default
*clauses	Clause	The operator's clauses.	()

Source code in motorhead/operator.py

def __init__(self, *clauses: Clause) -> None:
    """
    Initialization.

    Arguments:
        *clauses: The operator's clauses.
    """
    self._clauses = clauses

to_mongo()

Converts the operator to a MongoDB-compatible dict.

Source code in motorhead/operator.py

def to_mongo(self) -> dict[str, Any]:
    """Converts the operator to a MongoDB-compatible dict."""
    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:
        """
        Initialization.

        Arguments:
            key: The operator's key (the document attribute name).
            value: The operator's value.
        """
        if not isinstance(value, dict):
            raise ValueError("ElemMatch field only accepts dict values.")
        super().__init__(key, value)

__init__(key, value)

Initialization.

Parameters:

Name	Type	Description	Default
key	str | Field	The operator's key (the document attribute name).	required
value	dict[str, Any]	The operator's value.	required

Source code in motorhead/operator.py

def __init__(self, key: str | Field, value: dict[str, Any]) -> None:
    """
    Initialization.

    Arguments:
        key: The operator's key (the document attribute name).
        value: The operator's value.
    """
    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:
        """
        Initialization.

        Arguments:
            key: The operator's key (the document attribute name).
            value: The operator's value.
        """
        if not isinstance(value, bool):
            raise ValueError("Exists field only accepts bool values.")
        super().__init__(key, value)

__init__(key, value)

Initialization.

Parameters:

Name	Type	Description	Default
key	str | Field	The operator's key (the document attribute name).	required
value	bool	The operator's value.	required

Source code in motorhead/operator.py

def __init__(self, key: str | Field, value: bool) -> None:
    """
    Initialization.

    Arguments:
        key: The operator's key (the document attribute name).
        value: The operator's value.
    """
    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:
        """
        Initialization.

        Arguments:
            key: The operator's key (the document attribute name).
            value: The operator's value.
        """
        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:
        """The operator's key (the document attribute name)."""
        return self._key

    @property
    def value(self) -> Any:
        """The operator's value."""
        return self._value

    def to_mongo(self) -> dict[str, Any]:
        """Converts the operator to a MongoDB-compatible dict."""
        return {self._key: {self._operator: self._value}}

key property

The operator's key (the document attribute name).

value property

The operator's value.

__init__(key, value)

Initialization.

Parameters:

Name	Type	Description	Default
key	str | Field	The operator's key (the document attribute name).	required
value	Any	The operator's value.	required

Source code in motorhead/operator.py

def __init__(self, key: str | Field, value: Any) -> None:
    """
    Initialization.

    Arguments:
        key: The operator's key (the document attribute name).
        value: The operator's value.
    """
    self._key = key if isinstance(key, str) else key.name
    self._value = value

to_mongo()

Converts the operator to a MongoDB-compatible dict.

Source code in motorhead/operator.py

def to_mongo(self) -> dict[str, Any]:
    """Converts the operator to a MongoDB-compatible dict."""
    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 query dict.

Source code in motorhead/operator.py

class Raw:
    """Clause that wraps a raw, MongoDB query dict."""

    __slots__ = ("_data",)

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

        Arguments:
            value: The raw MongoDB query dict.
        """
        self._data = value

    def to_mongo(self) -> dict[str, Any]:
        """Converts the operator to a MongoDB-compatible dict."""
        return {**self._data}  # Just a shallow copy

__init__(value)

Initialization.

Parameters:

Name	Type	Description	Default
value	dict[str, Any]	The raw MongoDB query dict.	required

Source code in motorhead/operator.py

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

    Arguments:
        value: The raw MongoDB query dict.
    """
    self._data = value

to_mongo()

Converts the operator to a MongoDB-compatible dict.

Source code in motorhead/operator.py

def to_mongo(self) -> dict[str, Any]:
    """Converts the operator to a MongoDB-compatible dict."""
    return {**self._data}  # Just a shallow copy

Regex

Bases: KeyValueOperator

$regex MongoDB operator.

Source code in motorhead/operator.py

class Regex(KeyValueOperator):
    """`$regex` MongoDB operator."""

    __slots__ = ()

    def __init__(self, key: str | Field, value: str) -> None:
        """
        Initialization.

        Arguments:
            key: The operator's key (the document attribute name).
            value: The operator's value.
        """
        if not isinstance(value, str):
            raise ValueError("Regex field only accepts str values.")
        super().__init__(key, value)

__init__(key, value)

Initialization.

Parameters:

Name	Type	Description	Default
key	str | Field	The operator's key (the document attribute name).	required
value	str	The operator's value.	required

Source code in motorhead/operator.py

def __init__(self, key: str | Field, value: str) -> None:
    """
    Initialization.

    Arguments:
        key: The operator's key (the document attribute name).
        value: The operator's value.
    """
    if not isinstance(value, str):
        raise ValueError("Regex field only accepts str values.")
    super().__init__(key, value)

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:
        """
        Initialization.

        Arguments:
            key: The operator's key (the document attribute name).
            value: The operator's value.
        """
        if not isinstance(value, int):
            raise ValueError("Size field only accepts int values.")
        super().__init__(key, value)

__init__(key, value)

Initialization.

Parameters:

Name	Type	Description	Default
key	str | Field	The operator's key (the document attribute name).	required
value	int	The operator's value.	required

Source code in motorhead/operator.py

def __init__(self, key: str | Field, value: int) -> None:
    """
    Initialization.

    Arguments:
        key: The operator's key (the document attribute name).
        value: The operator's value.
    """
    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:
        """
        Initialization.

        Arguments:
            key: The operator's key (the document attribute name).
            value: The operator's value.
        """
        if not isinstance(value, str):
            raise ValueError("Type field only accepts string values.")
        super().__init__(key, value)

__init__(key, value)

Initialization.

Parameters:

Name	Type	Description	Default
key	str | Field	The operator's key (the document attribute name).	required
value	str	The operator's value.	required

Source code in motorhead/operator.py

def __init__(self, key: str | Field, value: str) -> None:
    """
    Initialization.

    Arguments:
        key: The operator's key (the document attribute name).
        value: The operator's value.
    """
    if not isinstance(value, str):
        raise ValueError("Type field only accepts string values.")
    super().__init__(key, value)

ensure_dict(data)

Converts the given value to a dict.

Source code in motorhead/operator.py

def ensure_dict(data: dict[str, Any] | Clause) -> dict[str, Any]:
    """Converts the given value to a dict."""
    return data.to_mongo() if hasattr(data, "to_mongo") else data