Skip to content

motorhead.query

Field

Queryable field with support for MongoDB operators in the form of overridden built-in operators or MongoDB-specific methods.

Source code in motorhead/query.py 
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
class Field:
    """
    Queryable field with support for MongoDB operators in the form of
    overridden built-in operators or MongoDB-specific methods.
    """

    __slots__ = ("_name",)

    def __init__(self, name: str) -> None:
        self._name = name

    @property
    def name(self) -> str:
        """
        The name of the field as well as the name of the corresponding property
        in the MongoDB model.
        """
        return self._name

    # -- Operators

    def __lt__(self, value: Any) -> Query:
        return Query(Lt(self._name, value))

    def __le__(self, value: Any) -> Query:
        return Query(Lte(self._name, value))

    def __eq__(self, value: Any) -> Query:  # type: ignore[override]
        if isinstance(value, (list, set, tuple, dict)):
            return Query(Eq(self._name, value))

        return Query(DirectEq(self._name, value))

    def __ne__(self, value: Any) -> Query:  # type: ignore[override]
        return Query(Ne(self._name, value))

    def __gt__(self, value: Any) -> Query:
        return Query(Gt(self._name, value))

    def __ge__(self, value: Any) -> Query:
        return Query(Gte(self._name, value))

    # -- Methods

    def All(self, value: list[Any]) -> Query:
        """
        `$all` operator.

        Arguments:
            value: The operator's argument.
        """
        return Query(All(self._name, value))

    def ElemMatch(self, value: dict[str, Any]) -> Query:
        """
        `$elemMatch` operator.

        Arguments:
            value: The operator's argument.
        """
        return Query(ElemMatch(self._name, value))

    def Exists(self, value: bool) -> Query:
        """
        `$exists` operator.

        Arguments:
            value: The operator's argument.
        """
        return Query(Exists(self._name, value))

    def In(self, value: Any) -> Query:
        """
        `$in` operator.

        Arguments:
            value: The operator's argument.
        """
        return Query(In(self._name, value))

    def NotIn(self, value: Any) -> Query:
        """
        `$nin` operator.

        Arguments:
            value: The operator's argument.
        """
        return Query(NotIn(self._name, value))

    def Regex(self, value: str) -> Query:
        """
        `$regex` operator.

        Arguments:
            value: The regular expression.
        """
        return Query(Regex(self._name, value))

    def Size(self, value: int) -> Query:
        """
        `$size` operator.

        Arguments:
            value: The operator's argument.
        """
        return Query(Size(self._name, value))

    def Type(self, value: str) -> Query:
        """
        `$type` operator.

        Arguments:
            value: The operator's argument.
        """
        return Query(Type(self._name, value))

name property

The name of the field as well as the name of the corresponding property in the MongoDB model.

All(value)

$all operator.

Parameters:

Name Type Description Default
value list[Any]

The operator's argument.

 required
Source code in motorhead/query.py 
80
81
82
83
84
85
86
87
def All(self, value: list[Any]) -> Query:
    """
    `$all` operator.

    Arguments:
        value: The operator's argument.
    """
    return Query(All(self._name, value))

ElemMatch(value)

$elemMatch operator.

Parameters:

Name Type Description Default
value dict[str, Any]

The operator's argument.

 required
Source code in motorhead/query.py 
89
90
91
92
93
94
95
96
def ElemMatch(self, value: dict[str, Any]) -> Query:
    """
    `$elemMatch` operator.

    Arguments:
        value: The operator's argument.
    """
    return Query(ElemMatch(self._name, value))

Exists(value)

$exists operator.

Parameters:

Name Type Description Default
value bool

The operator's argument.

 required
Source code in motorhead/query.py 
 98
 99
100
101
102
103
104
105
def Exists(self, value: bool) -> Query:
    """
    `$exists` operator.

    Arguments:
        value: The operator's argument.
    """
    return Query(Exists(self._name, value))

In(value)

$in operator.

Parameters:

Name Type Description Default
value Any

The operator's argument.

 required
Source code in motorhead/query.py 
107
108
109
110
111
112
113
114
def In(self, value: Any) -> Query:
    """
    `$in` operator.

    Arguments:
        value: The operator's argument.
    """
    return Query(In(self._name, value))

NotIn(value)

$nin operator.

Parameters:

Name Type Description Default
value Any

The operator's argument.

 required
Source code in motorhead/query.py 
116
117
118
119
120
121
122
123
def NotIn(self, value: Any) -> Query:
    """
    `$nin` operator.

    Arguments:
        value: The operator's argument.
    """
    return Query(NotIn(self._name, value))

Regex(value)

$regex operator.

Parameters:

Name Type Description Default
value str

The regular expression.

 required
Source code in motorhead/query.py 
125
126
127
128
129
130
131
132
def Regex(self, value: str) -> Query:
    """
    `$regex` operator.

    Arguments:
        value: The regular expression.
    """
    return Query(Regex(self._name, value))

Size(value)

$size operator.

Parameters:

Name Type Description Default
value int

The operator's argument.

 required
Source code in motorhead/query.py 
134
135
136
137
138
139
140
141
def Size(self, value: int) -> Query:
    """
    `$size` operator.

    Arguments:
        value: The operator's argument.
    """
    return Query(Size(self._name, value))

Type(value)

$type operator.

Parameters:

Name Type Description Default
value str

The operator's argument.

 required
Source code in motorhead/query.py 
143
144
145
146
147
148
149
150
def Type(self, value: str) -> Query:
    """
    `$type` operator.

    Arguments:
        value: The operator's argument.
    """
    return Query(Type(self._name, value))

Query

Query implementation.

Source code in motorhead/query.py 
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
class Query:
    """
    Query implementation.
    """

    __slots__ = ("_clause",)

    def __init__(self, clause: Clause | None = None) -> None:
        """
        Initialization.

        Arguments:
            clause: The default clause of the query.
        """
        self._clause = clause

    def __and__(self, value: Clause) -> Query:
        """The `&` operator, produces a `Query`."""
        result = self.clone()
        clause = value._clause if isinstance(value, Query) else value
        if clause is None:
            result._clause = self._clause
        elif isinstance(self._clause, And):
            result._clause = And(*self._clause.clauses, clause)
        else:
            result._clause = clause if self._clause is None else And(self._clause, clause)

        return result

    def __or__(self, value: Clause) -> Query:
        """The `|` operator, produces a `Query`."""
        result = self.clone()
        clause = value._clause if isinstance(value, Query) else value
        if clause is None:
            result._clause = self._clause
        elif isinstance(self._clause, Or):
            result._clause = Or(*self._clause.clauses, clause)
        else:
            result._clause = clause if self._clause is None else Or(self._clause, clause)

        return result

    def clone(self) -> Query:
        """
        Returns a clone of the query.
        """
        result = type(self)()
        self._clone_attributes(result)
        return result

    def _clone_attributes(self, item: Query) -> None:
        """
        Clones the attributes of the query.

        Arguments:
            item: The query on which clonable properties must be set.
        """
        item._clause = self._clause

    def to_mongo(self) -> MongoQuery:
        """
        Returns the `MongoQuery` representation of the query.
        """
        return {} if self._clause is None else self._clause.to_mongo()

__and__(value)

The & operator, produces a Query.

Source code in motorhead/query.py 
169
170
171
172
173
174
175
176
177
178
179
180
def __and__(self, value: Clause) -> Query:
    """The `&` operator, produces a `Query`."""
    result = self.clone()
    clause = value._clause if isinstance(value, Query) else value
    if clause is None:
        result._clause = self._clause
    elif isinstance(self._clause, And):
        result._clause = And(*self._clause.clauses, clause)
    else:
        result._clause = clause if self._clause is None else And(self._clause, clause)

    return result

__init__(clause=None)

Initialization.

Parameters:

Name Type Description Default
clause Clause | None

The default clause of the query.

 None
Source code in motorhead/query.py 
160
161
162
163
164
165
166
167
def __init__(self, clause: Clause | None = None) -> None:
    """
    Initialization.

    Arguments:
        clause: The default clause of the query.
    """
    self._clause = clause

__or__(value)

The | operator, produces a Query.

Source code in motorhead/query.py 
182
183
184
185
186
187
188
189
190
191
192
193
def __or__(self, value: Clause) -> Query:
    """The `|` operator, produces a `Query`."""
    result = self.clone()
    clause = value._clause if isinstance(value, Query) else value
    if clause is None:
        result._clause = self._clause
    elif isinstance(self._clause, Or):
        result._clause = Or(*self._clause.clauses, clause)
    else:
        result._clause = clause if self._clause is None else Or(self._clause, clause)

    return result

clone()

Returns a clone of the query.

Source code in motorhead/query.py 
195
196
197
198
199
200
201
def clone(self) -> Query:
    """
    Returns a clone of the query.
    """
    result = type(self)()
    self._clone_attributes(result)
    return result

to_mongo()

Returns the MongoQuery representation of the query.

Source code in motorhead/query.py 
212
213
214
215
216
def to_mongo(self) -> MongoQuery:
    """
    Returns the `MongoQuery` representation of the query.
    """
    return {} if self._clause is None else self._clause.to_mongo()

Queryable

Base class for queryable models.

It expects a Pydantic BaseModel class during subclassing in its model argument and adds a queryable Field descriptor to the subclass for each field of the Pydantic model.

Example:

from pydantic import BaseModel

class Person(BaseModel):
    name: str
    lucky_number: int

class QPerson(Queryable, model=Person):
    __slots__ = ()

It's recommended to use the Q helper to achieve the above result, so you can have at least ORM-level type hints and autocompletion for the created queryable class.

from pydantic import BaseModel

class Person(BaseModel):
    name: str
    lucky_number: int

QPerson = Q(Person)
Source code in motorhead/query.py 
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
class Queryable:
    """
    Base class for queryable models.

    It expects a Pydantic `BaseModel` class during subclassing in its `model` argument and
    adds a queryable `Field` descriptor to the subclass for each field of the Pydantic model.

    Example:

    ```python
    from pydantic import BaseModel

    class Person(BaseModel):
        name: str
        lucky_number: int

    class QPerson(Queryable, model=Person):
        __slots__ = ()
    ```

    It's recommended to use the `Q` helper to achieve the above result, so you can have
    at least ORM-level type hints and autocompletion for the created queryable class.

    ```python
    from pydantic import BaseModel

    class Person(BaseModel):
        name: str
        lucky_number: int

    QPerson = Q(Person)
    ```
    """

    __slots__ = ()

    def __init_subclass__(cls, *, model: type[BaseModel]) -> None:
        """
        Subclass initialization.

        Arguments:
            model: The Pydantic model the `Queryable` is created for.
        """
        super().__init_subclass__()
        for name, info in model.model_fields.items():
            field_name = name
            if name == "id" and info.alias == "_id":
                field_name = "_id"

            setattr(cls, name, Field(field_name))

__init_subclass__(*, model)

Subclass initialization.

Parameters:

Name Type Description Default
model type[BaseModel]

The Pydantic model the Queryable is created for.

 required
Source code in motorhead/query.py 
255
256
257
258
259
260
261
262
263
264
265
266
267
268
def __init_subclass__(cls, *, model: type[BaseModel]) -> None:
    """
    Subclass initialization.

    Arguments:
        model: The Pydantic model the `Queryable` is created for.
    """
    super().__init_subclass__()
    for name, info in model.model_fields.items():
        field_name = name
        if name == "id" and info.alias == "_id":
            field_name = "_id"

        setattr(cls, name, Field(field_name))

Q(model)

Creates a new Queryable class that can be used to construct queries for documents with the given Pydantic model.

The created Queryable class will have a property (Field) for all properties that were declared in the given Pydantic class. For usage examples and more details, plase see Queryable.

Parameters:

Name Type Description Default
model type[_T]

The Pydantic model the Queryable class should be created for.

 required
Source code in motorhead/query.py 
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
def Q(model: type[_T]) -> type[_T]:
    """
    Creates a new `Queryable` class that can be used to construct queries for documents
    with the given Pydantic model.

    The created `Queryable` class will have a property (`Field`) for all properties
    that were declared in the given Pydantic class. For usage examples and more details,
    plase see `Queryable`.

    Arguments:
        model: The Pydantic model the `Queryable` class should be created for.
    """

    class Result(Queryable, model=model):
        __slots__ = ()

    return Result  # type: ignore[return-value]