Skip to content

ftmq.Query

See the query guide for a narrative introduction.

A filter over FtM entities, built from composable M / P / G nodes.

Examples:

from ftmq import Query, M, P, G

q = Query().where(M(schema="Person"), P(name__ilike="jane%"))
q = q.where(G(countries="de") | G(countries="at"))
q = q.order_by("name")[:10]
Source code in ftmq/query/main.py
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
151
152
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
217
218
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
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
class Query:
    """
    A filter over FtM entities, built from composable `M` / `P` / `G` nodes.

    Examples:
        ```python
        from ftmq import Query, M, P, G

        q = Query().where(M(schema="Person"), P(name__ilike="jane%"))
        q = q.where(G(countries="de") | G(countries="at"))
        q = q.order_by("name")[:10]
        ```
    """

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, Query):
            return NotImplemented
        return hash(self) == hash(other)

    def __init__(
        self,
        *nodes: Expr,
        q: Expr | None = None,
        aggregations: Iterable[Agg] | None = None,
        aggregator: Aggregator | None = None,
        sort: Sort | None = None,
        slice: slice | None = None,
    ):
        self.q: Expr | None = q if q is not None else combine(*nodes)
        self.aggregations: set[Agg] = set(aggregations or [])
        self.aggregator = aggregator
        self.sort = sort
        self.slice = slice

    def __getitem__(self, value: Any) -> Self:
        """
        Implement list-like slicing. No negative values allowed.

        Examples:
            >>> q[1]
            # 2nd element (0-index)
            >>> q[:10]
            # first 10 elements
            >>> q[10:20]
            # next 10 elements

        Returns:
            The updated `Query` instance
        """
        if isinstance(value, int):
            if value < 0:
                raise QueryError(f"Invalid slicing: `{value}`")
            return self._chain(slice=slice(value, value + 1))
        if isinstance(value, slice):
            if value.step is not None:
                raise QueryError(f"Invalid slicing: `{value}`")
            return self._chain(slice=value)
        raise NotImplementedError

    def __bool__(self) -> bool:
        """
        Detect if any filter, ordering or slicing is defined

        Examples:
            >>> bool(Query())
            False
            >>> bool(Query().where(M(dataset="my_dataset")))
            True
        """
        return bool(self.to_dict())

    def __hash__(self) -> int:
        """
        Generate a unique key of the current state, useful for caching.

        Like any Python object this is a within-process hash (not stable
        across processes); `hash_data` normalizes ordering so equal queries
        hash equal.
        """
        return hash(hash_data(self.to_dict()))

    def _chain(self, **kwargs: Any) -> Self:
        data: dict[str, Any] = dict(
            q=self.q,
            aggregations=self.aggregations,
            aggregator=self.aggregator,
            sort=self.sort,
            slice=self.slice,
        )
        data.update(kwargs)
        return self.__class__(**data)

    # --- filter accessors (tree-walking collectors) ------------------------

    @property
    def _leaves(self) -> list[Leaf]:
        return list(self.q.iter_leaves()) if self.q else []

    @property
    def limit(self) -> int | None:
        """
        The current limit (inferred from a slice)
        """
        if self.slice is None:
            return None
        start, stop = self.slice.start, self.slice.stop
        if start and stop:
            return int(stop) - int(start)
        return None if stop is None else int(stop)

    @property
    def offset(self) -> int | None:
        """
        The current offset (inferred from a slice)

        A start-less slice (`q[:10]`) reports offset `0`, so it serializes and
        round-trips identically to `q[0:10]`.
        """
        if self.slice is None:
            return None
        return int(self.slice.start or 0)

    @property
    def sql(self) -> "Sql":
        """
        An adapter of this query for sql interfaces, against the default
        nomenklatura statement table. For a custom / extended table pass a
        [`SqlSource`][ftmq.query.sql.SqlSource] to [`compile`][ftmq.Query.compile] or
        build `Sql(query, source)` directly.
        """
        return Sql(self)

    def compile(self, source: "SqlSource | None" = None) -> "Select[Any]":
        """
        Compile this query to a SQLAlchemy `Select` of statements against a
        [`SqlSource`][ftmq.query.sql.SqlSource] (a store's table descriptor).

        Args:
            source: The SQL source to compile against (default: the base
                nomenklatura statement table).

        Returns:
            The statements `Select`.
        """
        return Sql(self, source).statements

    @property
    def ids(self) -> set[IdLeaf]:
        """
        The current id filters
        """
        return {f for f in self._leaves if isinstance(f, IdLeaf)}

    @property
    def datasets(self) -> set[DatasetLeaf]:
        """
        The current dataset filters
        """
        return {f for f in self._leaves if isinstance(f, DatasetLeaf)}

    @property
    def dataset_names(self) -> set[str]:
        """
        The names of the current filtered datasets
        """
        names: set[str] = set()
        for f in self.datasets:
            names.update(ensure_list(f.value))
        return names

    @property
    def schemata(self) -> set[SchemaLeaf]:
        """
        The current schema filters
        """
        return {f for f in self._leaves if isinstance(f, SchemaLeaf)}

    @property
    def schemata_names(self) -> set[str]:
        """
        The names of the current filtered schemas

        Exact `schema` leaves contribute their name; `schemata` (is-a) leaves
        expand to the schema plus its non-abstract descendants.
        """
        names: set[str] = set()
        for f in self._leaves:
            if isinstance(f, SchemataLeaf):
                for schema in f.schemata:
                    names.add(schema.name)
                    names.update(d.name for d in schema.descendants if not d.abstract)
            elif isinstance(f, SchemaLeaf):
                names.update(ensure_list(f.value))
        return names

    @property
    def context(self) -> set[ContextLeaf]:
        """
        The current context filters (the `C` family, e.g. `origin`)
        """
        return {f for f in self._leaves if isinstance(f, ContextLeaf)}

    @property
    def countries(self) -> set[str]:
        """
        The current filtered countries
        """
        names: set[str] = set()
        for f in self._leaves:
            if isinstance(f, GroupLeaf) and f.key == "countries":
                names.update(ensure_list(f.value))
        return names

    @property
    def groups(self) -> set[GroupLeaf]:
        """
        The current property groups lookup filters
        """
        return {f for f in self._leaves if isinstance(f, GroupLeaf)}

    @property
    def properties(self) -> set[PropertyLeaf]:
        """
        The current property lookup filters
        """
        return {f for f in self._leaves if isinstance(f, PropertyLeaf)}

    # --- serialization -----------------------------------------------------

    def to_dict(self) -> dict[str, Any]:
        """
        Lossless nested-tree representation of the current object.

        Example:
            ```python
            q = Query().where(M(dataset__in=["d1", "d2"]))
            q = q.where(P(name="Jane") | P(name__ilike="j%"))
            data = q.to_dict()
            assert Query.from_dict(data).to_dict() == data
            ```
        """
        data: dict[str, Any] = {}
        if self.q:
            data["q"] = self.q.to_dict()
        if self.sort:
            data["order_by"] = self.sort.serialize()
        if self.slice:
            data["limit"] = self.limit
            data["offset"] = self.offset
        if self.aggregations:
            data["aggregations"] = aggregations_to_dict(self.aggregations)
        return data

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> Self:
        """Rebuild a `Query` from its [`to_dict`][ftmq.Query.to_dict] output."""
        q = Expr.from_dict(data["q"]) if data.get("q") else None
        sort = None
        if data.get("order_by"):
            values = list(data["order_by"])
            ascending = not (values and str(values[0]).startswith("-"))
            sort = Sort(
                values=[str(v).lstrip("-") for v in values], ascending=ascending
            )
        slice_ = _make_slice(data.get("limit"), data.get("offset"))
        aggregations = None
        if data.get("aggregations"):
            aggregations = aggregations_from_dict(data["aggregations"])
        return cls(q=q, sort=sort, slice=slice_, aggregations=aggregations)

    def to_params(self) -> dict[str, list[str]]:
        """
        Project to an Aleph-style filter param dict (`filter:` / `exclude:` /
        `empty:` keys, `metric:` / `facet` aggregation keys, plus `sort` /
        `limit` / `offset`).

        Raises `QueryError` for queries outside the flat Aleph-expressible
        subset (cross-field OR, negated groups).
        """
        params = {k: list(v) for k, v in expr_to_params(self.q).items()}
        if self.aggregations:
            params.update(aggregations_to_params(self.aggregations))
        if self.sort:
            params["sort"] = [
                f"{v[1:]}:desc" if v.startswith("-") else f"{v}:asc"
                for v in self.sort.serialize()
            ]
        if self.slice:
            if self.offset:
                params["offset"] = [str(self.offset)]
            if self.limit is not None:
                params["limit"] = [str(self.limit)]
        return params

    @classmethod
    def from_params(cls, args: Any) -> Self:
        """Build a `Query` from an Aleph-style param dict / MultiDict."""
        items = normalize_multidict(args)
        q = params_to_expr(items)
        aggregations = params_to_aggregations(items) or None
        sort = None
        if items.get("sort"):
            svalues: list[str] = []
            ascending = True
            for value in items["sort"]:
                field, _, direction = value.partition(":")
                svalues.append(field)
                ascending = direction != "desc"
            sort = Sort(values=svalues, ascending=ascending)
        slice_ = None
        if "limit" in items or "offset" in items:
            offset = int((items.get("offset") or ["0"])[0] or 0)
            _limit = items.get("limit")
            limit = int(_limit[0]) if _limit else None
            slice_ = _make_slice(limit, offset)
        return cls(q=q, sort=sort, slice=slice_, aggregations=aggregations)

    def to_string(self) -> str:
        """
        Project to an Aleph URL query string, e.g.
        `filter:properties.name=Jane&filter:schemata=LegalEntity`.
        """
        return params_to_string(self.to_params())

    @classmethod
    def from_string(cls, value: str) -> Self:
        """Build a `Query` from an Aleph URL query string."""
        return cls.from_params(string_to_params(value))

    @classmethod
    def from_rql(cls, value: str) -> Self:
        """Build a `Query` from an [RQL](https://github.com/pjwerneck/pyrql) string.

        Unlike the flat Aleph grammar, RQL expresses arbitrary `& | ~` nesting,
        e.g. `and(eq(schema,Person),or(eq(properties.name,jane),eq(countries,de)))`,
        and carries aggregations via its `sum` / `aggregate(...)` operators.
        """
        expr, aggregations = parse_rql(value)
        return cls(q=expr, aggregations=aggregations)

    def to_rql(self) -> str:
        """Serialize the filter tree and aggregations to an
        [RQL](https://github.com/pjwerneck/pyrql) string.

        RQL is the only string surface that preserves arbitrary `& | ~` nesting
        (unlike the flat Aleph params) and carries aggregations losslessly, so it
        is the way to hand a full query to another HTTP-like connector. Raises
        `QueryError` for a comparator with no RQL equivalent (`null`,
        `startswith`, `endswith`, ...).
        """
        return serialize_rql(self.q, self.aggregations)

    # --- building ----------------------------------------------------------

    def where(self, *nodes: Expr) -> Self:
        """
        AND another set of `M` / `P` / `G` nodes into the current `Query`.

        Example:
            ```python
            q = Query().where(M(schema="Payment"), P(date__gte="2024-10"))
            q = q.where(G(countries="de") | G(countries="at"))
            ```

        Args:
            *nodes: `M` / `P` / `G` nodes (optionally composed with `&`/`|`/`~`)

        Returns:
            The updated `Query` instance
        """
        new = combine(*nodes)
        if new is None:
            return self._chain()
        q = new if self.q is None else (self.q & new)
        return self._chain(q=q)

    def order_by(self, *values: str, ascending: bool | None = True) -> Self:
        """
        Add or update the current sorting.

        Args:
            *values: Fields to order by
            ascending: Ascending or descending

        Returns:
            The updated `Query` instance.
        """
        self.sort = Sort(values=values, ascending=ascending)
        return self._chain()

    def aggregate(self, *nodes: A) -> Self:
        """Add aggregation projections to the query.

        Example:
            ```python
            from ftmq import Query, M, A

            q = Query().where(M(schema="Payment")).aggregate(
                A(sum="amountEur", by="beneficiary"),
                A(avg="amountEur"),
            )
            ```

        Args:
            *nodes: `A` nodes, e.g. `A(sum="amountEur", by="beneficiary")`.

        Returns:
            The updated `Query` instance.
        """
        aggs = set(self.aggregations)
        for node in nodes:
            aggs.update(node.aggs)
        return self._chain(aggregations=aggs)

    def get_aggregator(self) -> Aggregator:
        """Build an in-memory `Aggregator` from the query's aggregation specs.

        Returns:
            A fresh accumulator over this query's aggregations.
        """
        return Aggregator(self.aggregations)

    # --- execution ---------------------------------------------------------

    def apply(self, entity: EntityProxy) -> bool:
        """
        Test if a entity matches the current `Query` instance.
        """
        if self.q is None:
            return True
        return self.q.apply(entity)

    def apply_iter(self, entities: EntityProxies) -> EntityProxies:
        """
        Apply the current `Query` instance to a generator of entities and return
        a generator of filtered entities

        Example:
            ```python
            entities = [...]
            q = Query().where(M(dataset="my_dataset"), M(schema="Company"))
            for entity in q.apply_iter(entities):
                assert entity.schema.name == "Company"
            ```

        Yields:
            A generator of `EntityProxy` or a sub-type
        """
        if not self:
            yield from entities
            return

        entities = (e for e in entities if self.apply(e))
        if self.sort:
            entities = self.sort.apply_iter(entities)
        if self.slice:
            entities = islice(
                entities, self.slice.start, self.slice.stop, self.slice.step
            )
        if self.aggregations:
            self.aggregator = self.get_aggregator()
            entities = self.aggregator.apply(cast(Any, entities))
        yield from entities

context property

The current context filters (the C family, e.g. origin)

countries property

The current filtered countries

dataset_names property

The names of the current filtered datasets

datasets property

The current dataset filters

groups property

The current property groups lookup filters

ids property

The current id filters

limit property

The current limit (inferred from a slice)

offset property

The current offset (inferred from a slice)

A start-less slice (q[:10]) reports offset 0, so it serializes and round-trips identically to q[0:10].

properties property

The current property lookup filters

schemata property

The current schema filters

schemata_names property

The names of the current filtered schemas

Exact schema leaves contribute their name; schemata (is-a) leaves expand to the schema plus its non-abstract descendants.

sql property

An adapter of this query for sql interfaces, against the default nomenklatura statement table. For a custom / extended table pass a SqlSource to compile or build Sql(query, source) directly.

__bool__()

Detect if any filter, ordering or slicing is defined

Examples:

>>> bool(Query())
False
>>> bool(Query().where(M(dataset="my_dataset")))
True
Source code in ftmq/query/main.py
def __bool__(self) -> bool:
    """
    Detect if any filter, ordering or slicing is defined

    Examples:
        >>> bool(Query())
        False
        >>> bool(Query().where(M(dataset="my_dataset")))
        True
    """
    return bool(self.to_dict())

__getitem__(value)

Implement list-like slicing. No negative values allowed.

Examples:

>>> q[1]
# 2nd element (0-index)
>>> q[:10]
# first 10 elements
>>> q[10:20]
# next 10 elements

Returns:

Type Description
Self

The updated Query instance

Source code in ftmq/query/main.py
def __getitem__(self, value: Any) -> Self:
    """
    Implement list-like slicing. No negative values allowed.

    Examples:
        >>> q[1]
        # 2nd element (0-index)
        >>> q[:10]
        # first 10 elements
        >>> q[10:20]
        # next 10 elements

    Returns:
        The updated `Query` instance
    """
    if isinstance(value, int):
        if value < 0:
            raise QueryError(f"Invalid slicing: `{value}`")
        return self._chain(slice=slice(value, value + 1))
    if isinstance(value, slice):
        if value.step is not None:
            raise QueryError(f"Invalid slicing: `{value}`")
        return self._chain(slice=value)
    raise NotImplementedError

__hash__()

Generate a unique key of the current state, useful for caching.

Like any Python object this is a within-process hash (not stable across processes); hash_data normalizes ordering so equal queries hash equal.

Source code in ftmq/query/main.py
def __hash__(self) -> int:
    """
    Generate a unique key of the current state, useful for caching.

    Like any Python object this is a within-process hash (not stable
    across processes); `hash_data` normalizes ordering so equal queries
    hash equal.
    """
    return hash(hash_data(self.to_dict()))

aggregate(*nodes)

Add aggregation projections to the query.

Example
from ftmq import Query, M, A

q = Query().where(M(schema="Payment")).aggregate(
    A(sum="amountEur", by="beneficiary"),
    A(avg="amountEur"),
)

Parameters:

Name Type Description Default
*nodes A

A nodes, e.g. A(sum="amountEur", by="beneficiary").

()

Returns:

Type Description
Self

The updated Query instance.

Source code in ftmq/query/main.py
def aggregate(self, *nodes: A) -> Self:
    """Add aggregation projections to the query.

    Example:
        ```python
        from ftmq import Query, M, A

        q = Query().where(M(schema="Payment")).aggregate(
            A(sum="amountEur", by="beneficiary"),
            A(avg="amountEur"),
        )
        ```

    Args:
        *nodes: `A` nodes, e.g. `A(sum="amountEur", by="beneficiary")`.

    Returns:
        The updated `Query` instance.
    """
    aggs = set(self.aggregations)
    for node in nodes:
        aggs.update(node.aggs)
    return self._chain(aggregations=aggs)

apply(entity)

Test if a entity matches the current Query instance.

Source code in ftmq/query/main.py
def apply(self, entity: EntityProxy) -> bool:
    """
    Test if a entity matches the current `Query` instance.
    """
    if self.q is None:
        return True
    return self.q.apply(entity)

apply_iter(entities)

Apply the current Query instance to a generator of entities and return a generator of filtered entities

Example
entities = [...]
q = Query().where(M(dataset="my_dataset"), M(schema="Company"))
for entity in q.apply_iter(entities):
    assert entity.schema.name == "Company"

Yields:

Type Description
EntityProxies

A generator of EntityProxy or a sub-type

Source code in ftmq/query/main.py
def apply_iter(self, entities: EntityProxies) -> EntityProxies:
    """
    Apply the current `Query` instance to a generator of entities and return
    a generator of filtered entities

    Example:
        ```python
        entities = [...]
        q = Query().where(M(dataset="my_dataset"), M(schema="Company"))
        for entity in q.apply_iter(entities):
            assert entity.schema.name == "Company"
        ```

    Yields:
        A generator of `EntityProxy` or a sub-type
    """
    if not self:
        yield from entities
        return

    entities = (e for e in entities if self.apply(e))
    if self.sort:
        entities = self.sort.apply_iter(entities)
    if self.slice:
        entities = islice(
            entities, self.slice.start, self.slice.stop, self.slice.step
        )
    if self.aggregations:
        self.aggregator = self.get_aggregator()
        entities = self.aggregator.apply(cast(Any, entities))
    yield from entities

compile(source=None)

Compile this query to a SQLAlchemy Select of statements against a SqlSource (a store's table descriptor).

Parameters:

Name Type Description Default
source 'SqlSource | None'

The SQL source to compile against (default: the base nomenklatura statement table).

None

Returns:

Type Description
'Select[Any]'

The statements Select.

Source code in ftmq/query/main.py
def compile(self, source: "SqlSource | None" = None) -> "Select[Any]":
    """
    Compile this query to a SQLAlchemy `Select` of statements against a
    [`SqlSource`][ftmq.query.sql.SqlSource] (a store's table descriptor).

    Args:
        source: The SQL source to compile against (default: the base
            nomenklatura statement table).

    Returns:
        The statements `Select`.
    """
    return Sql(self, source).statements

from_dict(data) classmethod

Rebuild a Query from its to_dict output.

Source code in ftmq/query/main.py
@classmethod
def from_dict(cls, data: dict[str, Any]) -> Self:
    """Rebuild a `Query` from its [`to_dict`][ftmq.Query.to_dict] output."""
    q = Expr.from_dict(data["q"]) if data.get("q") else None
    sort = None
    if data.get("order_by"):
        values = list(data["order_by"])
        ascending = not (values and str(values[0]).startswith("-"))
        sort = Sort(
            values=[str(v).lstrip("-") for v in values], ascending=ascending
        )
    slice_ = _make_slice(data.get("limit"), data.get("offset"))
    aggregations = None
    if data.get("aggregations"):
        aggregations = aggregations_from_dict(data["aggregations"])
    return cls(q=q, sort=sort, slice=slice_, aggregations=aggregations)

from_params(args) classmethod

Build a Query from an Aleph-style param dict / MultiDict.

Source code in ftmq/query/main.py
@classmethod
def from_params(cls, args: Any) -> Self:
    """Build a `Query` from an Aleph-style param dict / MultiDict."""
    items = normalize_multidict(args)
    q = params_to_expr(items)
    aggregations = params_to_aggregations(items) or None
    sort = None
    if items.get("sort"):
        svalues: list[str] = []
        ascending = True
        for value in items["sort"]:
            field, _, direction = value.partition(":")
            svalues.append(field)
            ascending = direction != "desc"
        sort = Sort(values=svalues, ascending=ascending)
    slice_ = None
    if "limit" in items or "offset" in items:
        offset = int((items.get("offset") or ["0"])[0] or 0)
        _limit = items.get("limit")
        limit = int(_limit[0]) if _limit else None
        slice_ = _make_slice(limit, offset)
    return cls(q=q, sort=sort, slice=slice_, aggregations=aggregations)

from_rql(value) classmethod

Build a Query from an RQL string.

Unlike the flat Aleph grammar, RQL expresses arbitrary & | ~ nesting, e.g. and(eq(schema,Person),or(eq(properties.name,jane),eq(countries,de))), and carries aggregations via its sum / aggregate(...) operators.

Source code in ftmq/query/main.py
@classmethod
def from_rql(cls, value: str) -> Self:
    """Build a `Query` from an [RQL](https://github.com/pjwerneck/pyrql) string.

    Unlike the flat Aleph grammar, RQL expresses arbitrary `& | ~` nesting,
    e.g. `and(eq(schema,Person),or(eq(properties.name,jane),eq(countries,de)))`,
    and carries aggregations via its `sum` / `aggregate(...)` operators.
    """
    expr, aggregations = parse_rql(value)
    return cls(q=expr, aggregations=aggregations)

from_string(value) classmethod

Build a Query from an Aleph URL query string.

Source code in ftmq/query/main.py
@classmethod
def from_string(cls, value: str) -> Self:
    """Build a `Query` from an Aleph URL query string."""
    return cls.from_params(string_to_params(value))

get_aggregator()

Build an in-memory Aggregator from the query's aggregation specs.

Returns:

Type Description
Aggregator

A fresh accumulator over this query's aggregations.

Source code in ftmq/query/main.py
def get_aggregator(self) -> Aggregator:
    """Build an in-memory `Aggregator` from the query's aggregation specs.

    Returns:
        A fresh accumulator over this query's aggregations.
    """
    return Aggregator(self.aggregations)

order_by(*values, ascending=True)

Add or update the current sorting.

Parameters:

Name Type Description Default
*values str

Fields to order by

()
ascending bool | None

Ascending or descending

True

Returns:

Type Description
Self

The updated Query instance.

Source code in ftmq/query/main.py
def order_by(self, *values: str, ascending: bool | None = True) -> Self:
    """
    Add or update the current sorting.

    Args:
        *values: Fields to order by
        ascending: Ascending or descending

    Returns:
        The updated `Query` instance.
    """
    self.sort = Sort(values=values, ascending=ascending)
    return self._chain()

to_dict()

Lossless nested-tree representation of the current object.

Example
q = Query().where(M(dataset__in=["d1", "d2"]))
q = q.where(P(name="Jane") | P(name__ilike="j%"))
data = q.to_dict()
assert Query.from_dict(data).to_dict() == data
Source code in ftmq/query/main.py
def to_dict(self) -> dict[str, Any]:
    """
    Lossless nested-tree representation of the current object.

    Example:
        ```python
        q = Query().where(M(dataset__in=["d1", "d2"]))
        q = q.where(P(name="Jane") | P(name__ilike="j%"))
        data = q.to_dict()
        assert Query.from_dict(data).to_dict() == data
        ```
    """
    data: dict[str, Any] = {}
    if self.q:
        data["q"] = self.q.to_dict()
    if self.sort:
        data["order_by"] = self.sort.serialize()
    if self.slice:
        data["limit"] = self.limit
        data["offset"] = self.offset
    if self.aggregations:
        data["aggregations"] = aggregations_to_dict(self.aggregations)
    return data

to_params()

Project to an Aleph-style filter param dict (filter: / exclude: / empty: keys, metric: / facet aggregation keys, plus sort / limit / offset).

Raises QueryError for queries outside the flat Aleph-expressible subset (cross-field OR, negated groups).

Source code in ftmq/query/main.py
def to_params(self) -> dict[str, list[str]]:
    """
    Project to an Aleph-style filter param dict (`filter:` / `exclude:` /
    `empty:` keys, `metric:` / `facet` aggregation keys, plus `sort` /
    `limit` / `offset`).

    Raises `QueryError` for queries outside the flat Aleph-expressible
    subset (cross-field OR, negated groups).
    """
    params = {k: list(v) for k, v in expr_to_params(self.q).items()}
    if self.aggregations:
        params.update(aggregations_to_params(self.aggregations))
    if self.sort:
        params["sort"] = [
            f"{v[1:]}:desc" if v.startswith("-") else f"{v}:asc"
            for v in self.sort.serialize()
        ]
    if self.slice:
        if self.offset:
            params["offset"] = [str(self.offset)]
        if self.limit is not None:
            params["limit"] = [str(self.limit)]
    return params

to_rql()

Serialize the filter tree and aggregations to an RQL string.

RQL is the only string surface that preserves arbitrary & | ~ nesting (unlike the flat Aleph params) and carries aggregations losslessly, so it is the way to hand a full query to another HTTP-like connector. Raises QueryError for a comparator with no RQL equivalent (null, startswith, endswith, ...).

Source code in ftmq/query/main.py
def to_rql(self) -> str:
    """Serialize the filter tree and aggregations to an
    [RQL](https://github.com/pjwerneck/pyrql) string.

    RQL is the only string surface that preserves arbitrary `& | ~` nesting
    (unlike the flat Aleph params) and carries aggregations losslessly, so it
    is the way to hand a full query to another HTTP-like connector. Raises
    `QueryError` for a comparator with no RQL equivalent (`null`,
    `startswith`, `endswith`, ...).
    """
    return serialize_rql(self.q, self.aggregations)

to_string()

Project to an Aleph URL query string, e.g. filter:properties.name=Jane&filter:schemata=LegalEntity.

Source code in ftmq/query/main.py
def to_string(self) -> str:
    """
    Project to an Aleph URL query string, e.g.
    `filter:properties.name=Jane&filter:schemata=LegalEntity`.
    """
    return params_to_string(self.to_params())

where(*nodes)

AND another set of M / P / G nodes into the current Query.

Example
q = Query().where(M(schema="Payment"), P(date__gte="2024-10"))
q = q.where(G(countries="de") | G(countries="at"))

Parameters:

Name Type Description Default
*nodes Expr

M / P / G nodes (optionally composed with &/|/~)

()

Returns:

Type Description
Self

The updated Query instance

Source code in ftmq/query/main.py
def where(self, *nodes: Expr) -> Self:
    """
    AND another set of `M` / `P` / `G` nodes into the current `Query`.

    Example:
        ```python
        q = Query().where(M(schema="Payment"), P(date__gte="2024-10"))
        q = q.where(G(countries="de") | G(countries="at"))
        ```

    Args:
        *nodes: `M` / `P` / `G` nodes (optionally composed with `&`/`|`/`~`)

    Returns:
        The updated `Query` instance
    """
    new = combine(*nodes)
    if new is None:
        return self._chain()
    q = new if self.q is None else (self.q & new)
    return self._chain(q=q)

Query nodes

The composable filter-node constructors: M (meta), P (property), G (property-type group) and C (context). A is the aggregation-projection node (see Aggregations below).

Bases: _FamilyExpr

Meta-field conditions: dataset, schema, schemata, origin, id, ...

Source code in ftmq/query/nodes.py
class M(_FamilyExpr):
    """Meta-field conditions: dataset, schema, schemata, origin, id, ..."""

    @staticmethod
    def _make(key: str, value: Any) -> Leaf:
        return make_meta_leaf(key, value)

Bases: _FamilyExpr

Specific-property conditions, e.g. P(name="Jane", amountEur__gte=1000).

Source code in ftmq/query/nodes.py
class P(_FamilyExpr):
    """Specific-property conditions, e.g. `P(name="Jane", amountEur__gte=1000)`."""

    @staticmethod
    def _make(key: str, value: Any) -> Leaf:
        return make_property_leaf(key, value)

Bases: _FamilyExpr

Property-type group conditions, e.g. G(countries="de"), G(entities=id).

Source code in ftmq/query/nodes.py
class G(_FamilyExpr):
    """Property-type group conditions, e.g. `G(countries="de")`, `G(entities=id)`."""

    @staticmethod
    def _make(key: str, value: Any) -> Leaf:
        return make_group_leaf(key, value)

Bases: _FamilyExpr

Context / storage-column conditions, e.g. C(origin="crawl"), C(first_seen__gte="2024-01").

Source code in ftmq/query/nodes.py
class C(_FamilyExpr):
    """Context / storage-column conditions, e.g. `C(origin="crawl")`,
    `C(first_seen__gte="2024-01")`."""

    @staticmethod
    def _make(key: str, value: Any) -> Leaf:
        return make_context_leaf(key, value)

An aggregation projection node: A(sum="amountEur", by="beneficiary").

Each keyword is an aggregation function (min, max, sum, avg, count) whose value is the property (or properties) to aggregate; by= groups by one or more properties / fields. Unlike the M/P/G/C filter nodes, A is not a boolean leaf - it does not compose with & | ~; pass it to Query.aggregate.

Examples:

A(sum="amountEur", by="beneficiary")
A(count="id")
A(sum=["amountEur", "amount"])
Source code in ftmq/query/aggregations.py
class A:
    """An aggregation projection node: `A(sum="amountEur", by="beneficiary")`.

    Each keyword is an aggregation function (`min`, `max`, `sum`, `avg`,
    `count`) whose value is the property (or properties) to aggregate; `by=`
    groups by one or more properties / fields. Unlike the `M`/`P`/`G`/`C`
    filter nodes, `A` is not a boolean leaf - it does not compose with
    `& | ~`; pass it to [`Query.aggregate`][ftmq.Query.aggregate].

    Examples:
        ```python
        A(sum="amountEur", by="beneficiary")
        A(count="id")
        A(sum=["amountEur", "amount"])
        ```
    """

    def __init__(
        self,
        *,
        by: str | Iterable[str] | None = None,
        **funcs: str | Iterable[str],
    ) -> None:
        groups: tuple[str, ...] = tuple(str(g) for g in ensure_list(by))
        aggs: list[Agg] = []
        for func, props in funcs.items():
            for prop in ensure_list(props):
                aggs.append(make_agg(func, prop, groups))
        if not aggs:
            raise QueryError("Empty aggregation: pass at least one `func=prop`")
        self.aggs: tuple[Agg, ...] = tuple(aggs)

Expression tree

A boolean node: a connector (AND/OR), an optional negation, and a list of children (nested Expr nodes and/or Leaf conditions).

Source code in ftmq/query/nodes.py
class Expr:
    """A boolean node: a connector (`AND`/`OR`), an optional negation, and a
    list of children (nested `Expr` nodes and/or `Leaf` conditions)."""

    def __init__(
        self,
        *children: "Expr | Leaf",
        connector: str = AND,
        negated: bool = False,
    ) -> None:
        self.connector = connector
        self.negated = negated
        self.children: list[Expr | Leaf] = list(children)

    def __bool__(self) -> bool:
        return bool(self.children) or self.negated

    def _copy(self) -> "Expr":
        clone = Expr(connector=self.connector, negated=self.negated)
        clone.children = list(self.children)
        return clone

    def _combine(self, other: "Expr", connector: str) -> "Expr":
        if not self:
            return other._copy()
        if not other:
            return self._copy()
        return Expr(self._copy(), other._copy(), connector=connector)

    def __and__(self, other: Any) -> "Expr":
        if not isinstance(other, Expr):
            return NotImplemented
        return self._combine(other, AND)

    def __or__(self, other: Any) -> "Expr":
        if not isinstance(other, Expr):
            return NotImplemented
        return self._combine(other, OR)

    def __invert__(self) -> "Expr":
        clone = self._copy()
        clone.negated = not self.negated
        return clone

    def apply(self, entity: EntityProxy) -> bool:
        """Evaluate the boolean expression against an entity.

        Args:
            entity: The entity to test.

        Returns:
            `True` if the entity matches this (possibly nested, possibly
            negated) tree of conditions.
        """
        if not self.children:
            result = True
        elif self.connector == OR:
            result = any(c.apply(entity) for c in self.children)
        else:
            result = all(c.apply(entity) for c in self.children)
        return (not result) if self.negated else result

    def iter_leaves(self, cls: type | None = None) -> Iterator[Leaf]:
        """Walk the tree and yield its leaf conditions.

        Args:
            cls: Optionally restrict to leaves of this class.

        Yields:
            Each matching leaf, depth-first.
        """
        for child in self.children:
            if isinstance(child, Expr):
                yield from child.iter_leaves(cls)
            elif cls is None or isinstance(child, cls):
                yield child

    def to_dict(self) -> dict[str, Any]:
        """Serialize the tree to a nested, canonically-ordered dict.

        Nested nodes that share the connector and are not negated are flattened
        (associativity) and children are sorted, so structurally-equivalent
        trees (e.g. built by different `where()` orderings) serialize
        identically and hash equal.

        Returns:
            A `{"and" | "or": [...], "not": bool}` mapping, round-trippable via
            [`from_dict`][ftmq.query.nodes.Expr.from_dict].
        """
        key = self.connector.lower()
        children: list[Any] = []
        for child in self.children:
            if isinstance(child, Expr):
                child_dict = child.to_dict()
                if not child.negated and child.connector == self.connector:
                    children.extend(child_dict[key])
                else:
                    children.append(child_dict)
            else:
                children.append({"leaf": child.field_dict()})
        children.sort(key=hash_data)
        data: dict[str, Any] = {key: children}
        if self.negated:
            data["not"] = True
        return data

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "Expr":
        """Rebuild a tree from its [`to_dict`][ftmq.query.nodes.Expr.to_dict] form.

        Args:
            data: The nested mapping to deserialize.

        Returns:
            The reconstructed expression.
        """
        connector = OR if "or" in data else AND
        children: list[Expr | Leaf] = []
        for child in data.get(connector.lower(), []):
            if "leaf" in child:
                children.append(leaf_from_dict(child["leaf"]))
            else:
                children.append(cls.from_dict(child))
        return cls(*children, connector=connector, negated=bool(data.get("not")))

    def __hash__(self) -> int:
        # a within-process hash over a normalized serialization; like any
        # Python object it is not stable across processes (banal's hash_data
        # normalizes key/element order so equal trees hash equal in-process)
        return hash(hash_data(self.to_dict()))

    def __eq__(self, other: Any) -> bool:
        return isinstance(other, Expr) and hash(self) == hash(other)

    def __repr__(self) -> str:
        return f"<Expr {self.to_dict()}>"

apply(entity)

Evaluate the boolean expression against an entity.

Parameters:

Name Type Description Default
entity EntityProxy

The entity to test.

required

Returns:

Type Description
bool

True if the entity matches this (possibly nested, possibly

bool

negated) tree of conditions.

Source code in ftmq/query/nodes.py
def apply(self, entity: EntityProxy) -> bool:
    """Evaluate the boolean expression against an entity.

    Args:
        entity: The entity to test.

    Returns:
        `True` if the entity matches this (possibly nested, possibly
        negated) tree of conditions.
    """
    if not self.children:
        result = True
    elif self.connector == OR:
        result = any(c.apply(entity) for c in self.children)
    else:
        result = all(c.apply(entity) for c in self.children)
    return (not result) if self.negated else result

from_dict(data) classmethod

Rebuild a tree from its to_dict form.

Parameters:

Name Type Description Default
data dict[str, Any]

The nested mapping to deserialize.

required

Returns:

Type Description
'Expr'

The reconstructed expression.

Source code in ftmq/query/nodes.py
@classmethod
def from_dict(cls, data: dict[str, Any]) -> "Expr":
    """Rebuild a tree from its [`to_dict`][ftmq.query.nodes.Expr.to_dict] form.

    Args:
        data: The nested mapping to deserialize.

    Returns:
        The reconstructed expression.
    """
    connector = OR if "or" in data else AND
    children: list[Expr | Leaf] = []
    for child in data.get(connector.lower(), []):
        if "leaf" in child:
            children.append(leaf_from_dict(child["leaf"]))
        else:
            children.append(cls.from_dict(child))
    return cls(*children, connector=connector, negated=bool(data.get("not")))

iter_leaves(cls=None)

Walk the tree and yield its leaf conditions.

Parameters:

Name Type Description Default
cls type | None

Optionally restrict to leaves of this class.

None

Yields:

Type Description
Leaf

Each matching leaf, depth-first.

Source code in ftmq/query/nodes.py
def iter_leaves(self, cls: type | None = None) -> Iterator[Leaf]:
    """Walk the tree and yield its leaf conditions.

    Args:
        cls: Optionally restrict to leaves of this class.

    Yields:
        Each matching leaf, depth-first.
    """
    for child in self.children:
        if isinstance(child, Expr):
            yield from child.iter_leaves(cls)
        elif cls is None or isinstance(child, cls):
            yield child

to_dict()

Serialize the tree to a nested, canonically-ordered dict.

Nested nodes that share the connector and are not negated are flattened (associativity) and children are sorted, so structurally-equivalent trees (e.g. built by different where() orderings) serialize identically and hash equal.

Returns:

Type Description
dict[str, Any]

A {"and" | "or": [...], "not": bool} mapping, round-trippable via

dict[str, Any]
Source code in ftmq/query/nodes.py
def to_dict(self) -> dict[str, Any]:
    """Serialize the tree to a nested, canonically-ordered dict.

    Nested nodes that share the connector and are not negated are flattened
    (associativity) and children are sorted, so structurally-equivalent
    trees (e.g. built by different `where()` orderings) serialize
    identically and hash equal.

    Returns:
        A `{"and" | "or": [...], "not": bool}` mapping, round-trippable via
        [`from_dict`][ftmq.query.nodes.Expr.from_dict].
    """
    key = self.connector.lower()
    children: list[Any] = []
    for child in self.children:
        if isinstance(child, Expr):
            child_dict = child.to_dict()
            if not child.negated and child.connector == self.connector:
                children.extend(child_dict[key])
            else:
                children.append(child_dict)
        else:
            children.append({"leaf": child.field_dict()})
    children.sort(key=hash_data)
    data: dict[str, Any] = {key: children}
    if self.negated:
        data["not"] = True
    return data

Combine a series of nodes with a single connector, skipping empties.

Parameters:

Name Type Description Default
*nodes Expr

The M / P / G / Expr nodes to combine.

()
connector str

AND (default) or OR.

AND

Returns:

Type Description
Expr | None

The combined expression, or None if no non-empty node was passed.

Source code in ftmq/query/nodes.py
def combine(*nodes: Expr, connector: str = AND) -> Expr | None:
    """Combine a series of nodes with a single connector, skipping empties.

    Args:
        *nodes: The `M` / `P` / `G` / `Expr` nodes to combine.
        connector: `AND` (default) or `OR`.

    Returns:
        The combined expression, or `None` if no non-empty node was passed.
    """
    result: Expr | None = None
    for node in nodes:
        if not node:
            continue
        if result is None:
            result = node
        elif connector == OR:
            result = result | node
        else:
            result = result & node
    return result

Leaves

Leaf conditions for the ftmq query language, split by the statement-table column they target:

  • meta leaves (M): dataset, schema (exact), schemata (is-a), id / entity_id / canonical_id.
  • the property leaf (P): a specific FtM property (the prop column).
  • the group leaf (G): a followthemoney property-type group (the prop_type column, keyed by registry.groups: names, dates, countries, entities, ...).
  • the context leaf (C): a provenance / storage column such as origin, fragment or first_seen (read from entity.context in-memory).

Lookup / BaseFilter handle comparator matching and value casting; the Leaf subclasses add the per-family entity access plus correct null (present/absent) semantics.

BaseFilter

Comparator + cast value; the shared base for all query Leaf classes.

The comparator is validated upstream by parse_lookup; here it is a plain string.

Source code in ftmq/query/leaves.py
class BaseFilter:
    """Comparator + cast value; the shared base for all query `Leaf` classes.

    The comparator is validated upstream by
    [`parse_lookup`][ftmq.query.leaves.parse_lookup]; here it is a plain string.
    """

    key: str = ""

    def __init__(self, value: Any, comparator: str | None = None) -> None:
        self.comparator: str = comparator or "eq"
        self.value: Any = self.get_casted_value(value)
        self.lookup: Lookup = Lookup(self.comparator, self.value)

    def __hash__(self) -> int:
        return hash((self.key, self.comparator, str(self.value)))

    def __eq__(self, other: Any) -> bool:
        return hash(self) == hash(other)

    def __lt__(self, other: Any) -> bool:
        # allow ordering (helpful for testing and stable sql output)
        return hash(self) < hash(other)

    def __le__(self, other: Any) -> bool:
        return hash(self) <= hash(other)

    def __gt__(self, other: Any) -> bool:
        return hash(self) > hash(other)

    def __ge__(self, other: Any) -> bool:
        return hash(self) >= hash(other)

    def to_dict(self) -> dict[str, Any]:
        key = self.key if self.comparator == "eq" else f"{self.key}__{self.comparator}"
        return {key: self.value}

    def get_casted_value(self, value: Any) -> Any:
        if self.comparator in ("in", "not_in"):
            return set(self.stringify(v) for v in ensure_list(value))
        if self.comparator == "null":
            return as_bool(value)
        if is_listish(value):
            raise QueryError(f"Invalid value for `{self.comparator}`: {value}")
        return self.stringify(value) if value is not None else None

    def stringify(self, value: Any) -> str:
        if hasattr(value, "name"):
            return str(value.name)
        return str(value)

CanonicalIdLeaf

Bases: IdLeaf

Matches the canonical_id column (the resolved id).

Source code in ftmq/query/leaves.py
class CanonicalIdLeaf(IdLeaf):
    """Matches the `canonical_id` column (the resolved id)."""

    key = "canonical_id"

ContextLeaf

Bases: Leaf

A context field (the C family).

In-memory it reads entity.context[key] (always treated as multi-valued); in SQL it maps to the same-named statement-table column. This is the general form of provenance / storage fields - origin, and extra columns such as fragment, first_seen, bucket - that are not followthemoney properties. An entity without the key (or without a context) simply does not match.

Source code in ftmq/query/leaves.py
class ContextLeaf(Leaf):
    """A context field (the `C` family).

    In-memory it reads `entity.context[key]` (always treated as multi-valued);
    in SQL it maps to the same-named statement-table column. This is the general
    form of provenance / storage fields - `origin`, and extra columns such as
    `fragment`, `first_seen`, `bucket` - that are not followthemoney properties.
    An entity without the key (or without a `context`) simply does not match.
    """

    family = "C"

    def __init__(self, key: str, value: Any, comparator: str | None = None):
        super().__init__(value, comparator)
        self.key = key

    def values(self, entity: EntityProxy) -> Iterator[str]:
        context = getattr(entity, "context", None) or {}
        value: Any = context.get(self.key)
        yield from ensure_list(value)

DatasetLeaf

Bases: Leaf

Matches an entity's datasets membership.

Source code in ftmq/query/leaves.py
class DatasetLeaf(Leaf):
    """Matches an entity's `datasets` membership."""

    family, key = "M", "dataset"

    def values(self, entity: EntityProxy) -> Iterator[str]:
        # `.datasets` is added by the StatementEntity / ValueEntity subclasses
        yield from getattr(entity, "datasets", [])

EntityIdLeaf

Bases: IdLeaf

Matches the entity_id column (the pre-resolution id).

Source code in ftmq/query/leaves.py
class EntityIdLeaf(IdLeaf):
    """Matches the `entity_id` column (the pre-resolution id)."""

    key = "entity_id"

GroupLeaf

Bases: Leaf

A property-type group (the prop_type column). entities is the reverse-lookup group.

Source code in ftmq/query/leaves.py
class GroupLeaf(Leaf):
    """A property-type group (the `prop_type` column). `entities` is the
    reverse-lookup group."""

    family = "G"

    def __init__(self, group: str, value: Any, comparator: str | None = None):
        if group not in registry.groups:
            raise QueryError(f"Invalid property group: `{group}`")
        super().__init__(value, comparator)
        self.key = group
        self.prop_type = registry.groups[group]

    def values(self, entity: EntityProxy) -> Iterator[str]:
        yield from entity.get_type_values(self.prop_type)

IdLeaf

Bases: Leaf

Matches an entity's id.

Source code in ftmq/query/leaves.py
class IdLeaf(Leaf):
    """Matches an entity's id."""

    family, key = "M", "id"

    def values(self, entity: EntityProxy) -> Iterator[str]:
        if entity.id is not None:
            yield entity.id

Leaf

Bases: BaseFilter

A single condition. Subclasses set family and implement values() (the entity values to test) or override apply().

Source code in ftmq/query/leaves.py
class Leaf(BaseFilter):
    """A single condition. Subclasses set `family` and implement `values()`
    (the entity values to test) or override `apply()`."""

    family: str = ""
    key: str = ""

    def values(self, entity: EntityProxy) -> Iterator[str]:
        """Yield the entity values this leaf tests against.

        Args:
            entity: The entity to read values from.

        Yields:
            The relevant string values (property values, schema name, ...).
        """
        raise NotImplementedError

    def apply(self, entity: EntityProxy) -> bool:
        """Test whether the entity matches this condition.

        Args:
            entity: The entity to test.

        Returns:
            `True` if any of the entity's values satisfy the comparator (or,
            for the `null` comparator, the presence / absence check).
        """
        if str(self.comparator) == "null":
            present = any(True for _ in self.values(entity))
            # value was cast to a bool by `BaseFilter.get_casted_value`
            return (not present) if self.value else present
        return any(self.lookup.apply(v) for v in self.values(entity))

    def field_dict(self) -> LeafDict:
        """Serialize this leaf to a family-tagged mapping.

        Returns:
            The `{t, f, op, v}` [`LeafDict`][ftmq.query.leaves.LeafDict] used by
            the query-tree serialization.
        """
        value = self.value
        if isinstance(value, (set, frozenset)):
            value = sorted(value)
        return LeafDict(t=self.family, f=self.key, op=str(self.comparator), v=value)

apply(entity)

Test whether the entity matches this condition.

Parameters:

Name Type Description Default
entity EntityProxy

The entity to test.

required

Returns:

Type Description
bool

True if any of the entity's values satisfy the comparator (or,

bool

for the null comparator, the presence / absence check).

Source code in ftmq/query/leaves.py
def apply(self, entity: EntityProxy) -> bool:
    """Test whether the entity matches this condition.

    Args:
        entity: The entity to test.

    Returns:
        `True` if any of the entity's values satisfy the comparator (or,
        for the `null` comparator, the presence / absence check).
    """
    if str(self.comparator) == "null":
        present = any(True for _ in self.values(entity))
        # value was cast to a bool by `BaseFilter.get_casted_value`
        return (not present) if self.value else present
    return any(self.lookup.apply(v) for v in self.values(entity))

field_dict()

Serialize this leaf to a family-tagged mapping.

Returns:

Type Description
LeafDict

The {t, f, op, v} LeafDict used by

LeafDict

the query-tree serialization.

Source code in ftmq/query/leaves.py
def field_dict(self) -> LeafDict:
    """Serialize this leaf to a family-tagged mapping.

    Returns:
        The `{t, f, op, v}` [`LeafDict`][ftmq.query.leaves.LeafDict] used by
        the query-tree serialization.
    """
    value = self.value
    if isinstance(value, (set, frozenset)):
        value = sorted(value)
    return LeafDict(t=self.family, f=self.key, op=str(self.comparator), v=value)

values(entity)

Yield the entity values this leaf tests against.

Parameters:

Name Type Description Default
entity EntityProxy

The entity to read values from.

required

Yields:

Type Description
str

The relevant string values (property values, schema name, ...).

Source code in ftmq/query/leaves.py
def values(self, entity: EntityProxy) -> Iterator[str]:
    """Yield the entity values this leaf tests against.

    Args:
        entity: The entity to read values from.

    Yields:
        The relevant string values (property values, schema name, ...).
    """
    raise NotImplementedError

LeafDict

Bases: TypedDict

Serialized form of a single Leaf condition.

Source code in ftmq/query/leaves.py
class LeafDict(TypedDict):
    """Serialized form of a single [`Leaf`][ftmq.query.leaves.Leaf] condition."""

    t: str  # family tag: "M" (meta) | "P" (property) | "G" (group)
    f: str  # field / property / group name
    op: str  # comparator, e.g. "eq", "in", "gte", "null"
    v: "str | bool | list[str]"  # cast value (list for `in` / `not_in`)

Lookup

Applies a single comparator to values (the in-memory match logic).

The comparator is a plain string, already validated by parse_lookup.

Source code in ftmq/query/leaves.py
class Lookup:
    """Applies a single comparator to values (the in-memory match logic).

    The comparator is a plain string, already validated by
    [`parse_lookup`][ftmq.query.leaves.parse_lookup].
    """

    def __init__(self, comparator: str, value: Any = None) -> None:
        self.comparator = comparator
        self.value = value

    def __str__(self) -> str:
        return self.comparator

    def apply(self, value: Any) -> bool:
        c = self.comparator
        if c == "eq":
            return bool(value == self.value)
        if c == "not":
            return bool(value != self.value)
        if c == "in":
            return value in self.value
        if c == "not_in":
            return value not in self.value
        if c == "startswith":
            return bool(value.startswith(self.value))
        if c == "endswith":
            return bool(value.endswith(self.value))
        if c == "null":
            return not value == self.value
        if c == "gt":
            return bool(value > self.value)
        if c == "gte":
            return bool(value >= self.value)
        if c == "lt":
            return bool(value < self.value)
        if c == "lte":
            return bool(value <= self.value)
        if c == "like":
            return self.value in value
        if c == "ilike":
            return self.value.lower() in value.lower()
        return False

PropertyLeaf

Bases: Leaf

Matches a specific FtM property value (the prop column).

Source code in ftmq/query/leaves.py
class PropertyLeaf(Leaf):
    """Matches a specific FtM property value (the `prop` column)."""

    family = "P"

    def __init__(self, prop: str | Property, value: Any, comparator: str | None = None):
        super().__init__(value, comparator)
        self.key = self._validate(prop)

    @staticmethod
    def _validate(prop: str | Property) -> str:
        if isinstance(prop, Property):
            return prop.name
        if isinstance(prop, str):
            for p in model.properties:
                if p.name == prop or p.qname == prop:
                    return prop
        raise QueryError(f"Invalid prop: `{prop}`")

    def values(self, entity: EntityProxy) -> Iterator[str]:
        yield from entity.get(self.key, quiet=True)

SchemaLeaf

Bases: Leaf

Exact schema match.

Source code in ftmq/query/leaves.py
class SchemaLeaf(Leaf):
    """Exact schema match."""

    family, key = "M", "schema"

    def __init__(self, value: Any, comparator: str | None = None) -> None:
        super().__init__(value, comparator)
        # validate real schema names for equality-style comparators (a
        # `startswith`/`ilike` prefix is not expected to be a full schema)
        if str(self.comparator) in ("eq", "in", "not", "not_in"):
            for name in ensure_list(value):
                if model.get(name) is None:
                    raise QueryError(f"Invalid schema: `{name}`")

    def values(self, entity: EntityProxy) -> Iterator[str]:
        yield entity.schema.name

SchemataLeaf

Bases: Leaf

is-a match: the entity's schema (or one of its ancestors) is the queried schema, i.e. model[X] in entity.schema.schemata.

Source code in ftmq/query/leaves.py
class SchemataLeaf(Leaf):
    """`is-a` match: the entity's schema (or one of its ancestors) is the
    queried schema, i.e. `model[X] in entity.schema.schemata`."""

    family, key = "M", "schemata"

    def __init__(self, value: Any, comparator: str | None = None) -> None:
        super().__init__(value, comparator)
        self.schemata: set[Schema] = set()
        for item in ensure_list(value):
            schema = item if isinstance(item, Schema) else model.get(item)
            if schema is None:
                raise QueryError(f"Invalid schema: `{item}`")
            self.schemata.add(schema)
        if not self.schemata:
            raise QueryError(f"Invalid schemata: `{value}`")
        if str(self.comparator) not in ("eq", "in", "not", "not_in"):
            raise QueryError(f"Invalid comparator for `schemata`: `{self.comparator}`")

    def apply(self, entity: EntityProxy) -> bool:
        hit = bool(self.schemata & entity.schema.schemata)
        if str(self.comparator) in ("not", "not_in"):
            return not hit
        return hit

leaf_from_dict(data)

Reconstruct a leaf from its serialized LeafDict.

Parameters:

Name Type Description Default
data LeafDict

The {t, f, op, v} mapping produced by Leaf.field_dict.

required

Returns:

Type Description
Leaf

The reconstructed leaf.

Source code in ftmq/query/leaves.py
def leaf_from_dict(data: LeafDict) -> Leaf:
    """Reconstruct a leaf from its serialized [`LeafDict`][ftmq.query.leaves.LeafDict].

    Args:
        data: The `{t, f, op, v}` mapping produced by
            [`Leaf.field_dict`][ftmq.query.leaves.Leaf.field_dict].

    Returns:
        The reconstructed leaf.
    """
    field, op, value = data["f"], data["op"], data["v"]
    key = field if op == "eq" else f"{field}__{op}"
    return LEAF_FACTORIES[data["t"]](key, value)

make_context_leaf(key, value)

Build a context leaf (the C family) from a lookup.

Parameters:

Name Type Description Default
key str

A context / column key, e.g. origin, fragment or first_seen__gte. Any identifier is accepted; validity of a SQL column is checked at compile time.

required
value Any

The lookup value.

required

Returns:

Type Description
Leaf

The resolved context leaf.

Source code in ftmq/query/leaves.py
def make_context_leaf(key: str, value: Any) -> Leaf:
    """Build a context leaf (the `C` family) from a lookup.

    Args:
        key: A context / column key, e.g. `origin`, `fragment` or
            `first_seen__gte`. Any identifier is accepted; validity of a SQL
            column is checked at compile time.
        value: The lookup value.

    Returns:
        The resolved context leaf.
    """
    field, comparator = parse_lookup(key)
    return ContextLeaf(field, value, comparator)

make_group_leaf(key, value)

Build a property-type group leaf (the G family) from a lookup.

Parameters:

Name Type Description Default
key str

A group lookup key, e.g. countries, dates__gte or entities.

required
value Any

The lookup value.

required

Returns:

Type Description
Leaf

The resolved group leaf.

Raises:

Type Description
QueryError

If the group is not a valid registry.groups name.

Source code in ftmq/query/leaves.py
def make_group_leaf(key: str, value: Any) -> Leaf:
    """Build a property-type group leaf (the `G` family) from a lookup.

    Args:
        key: A group lookup key, e.g. `countries`, `dates__gte` or `entities`.
        value: The lookup value.

    Returns:
        The resolved group leaf.

    Raises:
        QueryError: If the group is not a valid `registry.groups` name.
    """
    group, comparator = parse_lookup(key)
    return GroupLeaf(group, value, comparator)

make_meta_leaf(key, value)

Build a meta leaf (the M family) from a lookup.

Parameters:

Name Type Description Default
key str

A meta lookup key, e.g. dataset__in, schema or id__startswith.

required
value Any

The lookup value.

required

Returns:

Type Description
Leaf

The resolved meta leaf.

Raises:

Type Description
QueryError

If the field is not a known meta field.

Source code in ftmq/query/leaves.py
def make_meta_leaf(key: str, value: Any) -> Leaf:
    """Build a meta leaf (the `M` family) from a lookup.

    Args:
        key: A meta lookup key, e.g. `dataset__in`, `schema` or `id__startswith`.
        value: The lookup value.

    Returns:
        The resolved meta leaf.

    Raises:
        QueryError: If the field is not a known meta field.
    """
    field, comparator = parse_lookup(key)
    cls = _META_LEAVES.get(field)
    if cls is None:
        raise QueryError(f"Unknown meta field: `{field}`")
    return cls(value, comparator)

make_property_leaf(key, value)

Build a property leaf (the P family) from a lookup.

Parameters:

Name Type Description Default
key str

A property lookup key, e.g. name or amountEur__gte.

required
value Any

The lookup value.

required

Returns:

Type Description
Leaf

The resolved property leaf.

Raises:

Type Description
QueryError

If the property is not a valid FtM property.

Source code in ftmq/query/leaves.py
def make_property_leaf(key: str, value: Any) -> Leaf:
    """Build a property leaf (the `P` family) from a lookup.

    Args:
        key: A property lookup key, e.g. `name` or `amountEur__gte`.
        value: The lookup value.

    Returns:
        The resolved property leaf.

    Raises:
        QueryError: If the property is not a valid FtM property.
    """
    prop, comparator = parse_lookup(key)
    return PropertyLeaf(prop, value, comparator)

parse_lookup(key)

Split a field__comparator lookup key into its parts.

Parameters:

Name Type Description Default
key str

A lookup key such as name, date__gte or schema__in.

required

Returns:

Type Description
tuple[str, str]

A (field, comparator) tuple; the comparator defaults to eq.

Raises:

Type Description
QueryError

If the comparator suffix is not a valid comparator.

Source code in ftmq/query/leaves.py
def parse_lookup(key: str) -> tuple[str, str]:
    """Split a `field__comparator` lookup key into its parts.

    Args:
        key: A lookup key such as `name`, `date__gte` or `schema__in`.

    Returns:
        A `(field, comparator)` tuple; the comparator defaults to `eq`.

    Raises:
        QueryError: If the comparator suffix is not a valid comparator.
    """
    try:
        field, comparator = parse_comparator(key)
    except KeyError:
        raise QueryError(f"Invalid comparator in lookup: `{key}`")
    return field, str(comparator)

Aleph bridge

The filter half of the Aleph URL-param grammar. Query.to_params / from_params and to_string / from_string wrap these.

The Aleph / OpenAleph URL-param grammar: a bidirectional bridge between a Query filter tree and the filter: / exclude: / empty: param convention used by openaleph_search.SearchQueryParser.

The filter half (the mapping between the boolean tree and the param keys) lives here, plus the aggregation half (metric:<func>=<prop> and facet=<field>, matching openaleph's metric aggregations). sort / limit / offset are query-level concerns handled by Query.to_params / Query.from_params.

The param model is flat (AND across keys, OR within a key, exclude: and empty: for negation / absence), so:

  • params_to_expr is total and always yields a flat AND-of-leaves.
  • expr_to_params is defined on that flat subset and raises QueryError for a cross-field OR or a negated group.

aggregations_to_params(aggs)

Project aggregation specs to openaleph metric / facet params.

Each spec becomes a metric:<func>=<prop> entry (the convention openaleph_search.SearchQueryParser reads as metrics = {func: {props}}); grouped fields become facet=<field> values. openaleph nests every metric inside every facet bucket, so groups apply across all metrics - a per-metric grouping that differs between metrics collapses to their union here.

Parameters:

Name Type Description Default
aggs set[Agg]

The query's aggregation specs.

required

Returns:

Type Description
dict[str, list[str]]

The {"metric:<func>": [props], "facet": [groups]} param mapping.

Source code in ftmq/query/aleph.py
def aggregations_to_params(aggs: set[Agg]) -> dict[str, list[str]]:
    """Project aggregation specs to openaleph metric / facet params.

    Each spec becomes a `metric:<func>=<prop>` entry (the convention
    `openaleph_search.SearchQueryParser` reads as
    `metrics = {func: {props}}`); grouped fields become `facet=<field>`
    values. openaleph nests every metric inside every facet bucket, so groups
    apply across all metrics - a per-metric grouping that differs between
    metrics collapses to their union here.

    Args:
        aggs: The query's aggregation specs.

    Returns:
        The `{"metric:<func>": [props], "facet": [groups]}` param mapping.
    """
    params: dict[str, list[str]] = defaultdict(list)
    facets: set[str] = set()
    for agg in sorted(aggs, key=lambda a: (a.func, a.prop)):
        key = f"metric:{agg.func}"
        if agg.prop not in params[key]:
            params[key].append(agg.prop)
        facets.update(agg.groups)
    if facets:
        params["facet"] = sorted(facets)
    return dict(params)

expr_to_params(expr)

Project a filter tree to Aleph filter: / exclude: / empty: params.

Parameters:

Name Type Description Default
expr Expr | None

The filter tree (or None).

required

Returns:

Type Description
dict[str, list[str]]

The Aleph param mapping.

Raises:

Type Description
QueryError

If the tree is not Aleph-expressible (a cross-field OR, or a negated multi-leaf group).

Source code in ftmq/query/aleph.py
def expr_to_params(expr: Expr | None) -> dict[str, list[str]]:
    """Project a filter tree to Aleph `filter:` / `exclude:` / `empty:` params.

    Args:
        expr: The filter tree (or `None`).

    Returns:
        The Aleph param mapping.

    Raises:
        QueryError: If the tree is not Aleph-expressible (a cross-field `OR`, or
            a negated multi-leaf group).
    """
    params: dict[str, list[str]] = defaultdict(list)
    if expr:
        for leaf, inverted in _collect_terms(expr):
            prefix, key, values = _leaf_to_param(leaf, inverted)
            params[f"{prefix}{key}"].extend(values)
    return dict(params)

normalize_multidict(args)

Coerce params into a plain dict[str, list[str]].

Parameters:

Name Type Description Default
args Any

A werkzeug MultiDict, a plain dict, or a dict of lists.

required

Returns:

Type Description
dict[str, list[str]]

A mapping of each key to its list of string values.

Source code in ftmq/query/aleph.py
def normalize_multidict(args: Any) -> dict[str, list[str]]:
    """Coerce params into a plain `dict[str, list[str]]`.

    Args:
        args: A werkzeug `MultiDict`, a plain dict, or a dict of lists.

    Returns:
        A mapping of each key to its list of string values.
    """
    items: dict[str, list[str]] = defaultdict(list)
    if hasattr(args, "lists"):  # werkzeug MultiDict
        for key, values in args.lists():
            items[key].extend(str(v) for v in values)
    elif hasattr(args, "items"):
        for key, value in args.items():
            if isinstance(value, (list, tuple, set)):
                items[key].extend(str(v) for v in value)
            else:
                items[key].append(str(value))
    return dict(items)

params_to_aggregations(items)

Rebuild aggregation specs from openaleph metric: / facet params.

The inverse of aggregations_to_params: every facet field groups every metric:<func>=<prop> (matching how openaleph computes a metric within each facet bucket).

Parameters:

Name Type Description Default
items dict[str, list[str]]

A normalized param mapping.

required

Returns:

Type Description
set[Agg]

The reconstructed aggregation specs (empty if there are no metric:

set[Agg]

params).

Source code in ftmq/query/aleph.py
def params_to_aggregations(items: dict[str, list[str]]) -> set[Agg]:
    """Rebuild aggregation specs from openaleph `metric:` / `facet` params.

    The inverse of [`aggregations_to_params`][ftmq.query.aleph.aggregations_to_params]:
    every `facet` field groups every `metric:<func>=<prop>` (matching how
    openaleph computes a metric within each facet bucket).

    Args:
        items: A normalized param mapping.

    Returns:
        The reconstructed aggregation specs (empty if there are no `metric:`
        params).
    """
    groups = tuple(sorted(set(items.get("facet", []))))
    aggs: set[Agg] = set()
    for key, values in items.items():
        if key.startswith("metric:"):
            func = key[len("metric:") :]
            for prop in values:
                aggs.add(make_agg(func, prop, groups))
    return aggs

params_to_expr(items)

Build a filter tree from Aleph params (non-filter keys are ignored).

Parameters:

Name Type Description Default
items dict[str, list[str]]

A normalized param mapping (see normalize_multidict).

required

Returns:

Type Description
Expr | None

The flat AND-of-leaves filter tree, or None if there are no filters.

Source code in ftmq/query/aleph.py
def params_to_expr(items: dict[str, list[str]]) -> Expr | None:
    """Build a filter tree from Aleph params (non-filter keys are ignored).

    Args:
        items: A normalized param mapping (see
            [`normalize_multidict`][ftmq.query.aleph.normalize_multidict]).

    Returns:
        The flat AND-of-leaves filter tree, or `None` if there are no filters.
    """
    nodes: list[Expr] = []
    for key, values in items.items():
        for prefix in ("filter:", "exclude:", "empty:"):
            if key.startswith(prefix):
                nodes.append(_param_to_node(prefix, key[len(prefix) :], values))
                break
    return combine(*nodes) if nodes else None

params_to_string(params)

Render an Aleph param mapping as a URL query string.

Keys are sorted for deterministic output; value order within a key is preserved (multi-field sort priority must not be reordered).

Parameters:

Name Type Description Default
params dict[str, list[str]]

The param mapping.

required

Returns:

Type Description
str

A key=value&... string with url-encoded values, sorted by key.

Source code in ftmq/query/aleph.py
def params_to_string(params: dict[str, list[str]]) -> str:
    """Render an Aleph param mapping as a URL query string.

    Keys are sorted for deterministic output; value order within a key is
    preserved (multi-field `sort` priority must not be reordered).

    Args:
        params: The param mapping.

    Returns:
        A `key=value&...` string with url-encoded values, sorted by key.
    """
    parts = []
    for key in sorted(params):
        for value in params[key]:
            parts.append(f"{key}={quote(str(value))}")
    return "&".join(parts)

string_to_params(value)

Parse an Aleph URL query string into a param mapping.

Parameters:

Name Type Description Default
value str

A key=value&... query string.

required

Returns:

Type Description
dict[str, list[str]]

A mapping of each key to its list of url-decoded values.

Source code in ftmq/query/aleph.py
def string_to_params(value: str) -> dict[str, list[str]]:
    """Parse an Aleph URL query string into a param mapping.

    Args:
        value: A `key=value&...` query string.

    Returns:
        A mapping of each key to its list of url-decoded values.
    """
    items: dict[str, list[str]] = defaultdict(list)
    for part in value.split("&"):
        if not part:
            continue
        key, _, val = part.partition("=")
        items[key].append(unquote(val))
    return dict(items)

RQL bridge

RQL support for nested filter trees, used by Query.from_rql.

RQL (Resource Query Language) bridge.

RQL is a URL-friendly query language of nestable named operators - e.g. and(eq(schema,Person),or(eq(properties.name,jane),eq(countries,de))) - which maps directly onto the ftmq Expr tree (and/or/not + comparison leaves). Unlike the flat Aleph param grammar, RQL expresses arbitrary nesting, so this is the way to carry a full M & (P | G) tree through a single string. It also carries aggregations: RQL's native sum / min / max / mean / count and aggregate(...) operators map onto ftmq A nodes, side by side with the filter under a top-level and.

Field names follow the Aleph convention (schema, dataset, properties.<name>, a registry.groups name, or origin); a bare name that matches none of those is treated as an FtM property.

expr_to_rql(expr)

Convert an Expr tree to an RQL AST ({"name": ..., "args": [...]}).

Source code in ftmq/query/rql.py
def expr_to_rql(expr: Expr) -> dict[str, Any]:
    """Convert an `Expr` tree to an RQL AST (`{"name": ..., "args": [...]}`)."""
    group = "or" if expr.connector == OR else "and"
    parts: list[dict[str, Any]] = []
    for child in expr.children:
        if isinstance(child, Expr):
            child_ast = expr_to_rql(child)
            # flatten a non-negated same-connector subgroup into this one
            if not child.negated and child_ast.get("name") == group:
                parts.extend(child_ast["args"])
            else:
                parts.append(child_ast)
        else:
            parts.append(_leaf_to_rql(child))
    if not parts:
        raise QueryError("Cannot serialize an empty query to RQL")
    # a single-child group is just that child
    body = parts[0] if len(parts) == 1 else {"name": group, "args": parts}
    if expr.negated:
        return {"name": "not", "args": [body]}
    return body

parse_rql(value)

Parse an RQL query string into a filter Expr and aggregation specs.

Filter operators (and / or / not + comparisons) build the tree; the aggregate operators (sum / min / max / mean / count / aggregate) build the aggregations. At the top level they sit side by side under and.

Raises:

Type Description
QueryError

If the RQL uses an unsupported operator or field.

Source code in ftmq/query/rql.py
def parse_rql(value: str) -> tuple[Expr | None, set[Agg]]:
    """Parse an RQL query string into a filter `Expr` and aggregation specs.

    Filter operators (`and` / `or` / `not` + comparisons) build the tree; the
    aggregate operators (`sum` / `min` / `max` / `mean` / `count` / `aggregate`)
    build the aggregations. At the top level they sit side by side under `and`.

    Raises:
        QueryError: If the RQL uses an unsupported operator or field.
    """
    data = pyrql.parse(value)
    if not data:
        return None, set()
    aggs: set[Agg] = set()
    if data["name"] in AGG_OPERATORS:
        aggs.update(_node_aggs(data))
        return None, aggs
    if data["name"] == "and":
        filters: list[dict[str, Any]] = []
        for child in data["args"]:
            if isinstance(child, dict) and child.get("name") in AGG_OPERATORS:
                aggs.update(_node_aggs(child))
            else:
                filters.append(child)
        expr = combine(*(rql_to_expr(f) for f in filters), connector=AND)
        return expr, aggs
    return rql_to_expr(data), aggs

rql_to_expr(data)

Convert a parsed RQL AST ({"name": ..., "args": [...]}) to an Expr.

Source code in ftmq/query/rql.py
def rql_to_expr(data: dict[str, Any]) -> Expr:
    """Convert a parsed RQL AST (`{"name": ..., "args": [...]}`) to an `Expr`."""
    op, args = data["name"], data["args"]
    if op == "and":
        result = combine(*(rql_to_expr(a) for a in args), connector=AND)
    elif op == "or":
        result = combine(*(rql_to_expr(a) for a in args), connector=OR)
    elif op == "not":
        return ~rql_to_expr(args[0])
    else:
        return _rql_leaf(op, args)
    if result is None:
        raise QueryError(f"Empty RQL group: `{op}`")
    return result

to_rql(expr, aggs=())

Serialize a filter tree and aggregation specs to an RQL query string.

Filters and aggregations sit side by side under a top-level and.

Raises:

Type Description
QueryError

If a filter leaf uses a comparator with no RQL equivalent (null, startswith, endswith, ...).

Source code in ftmq/query/rql.py
def to_rql(expr: Expr | None, aggs: Iterable[Agg] = ()) -> str:
    """Serialize a filter tree and aggregation specs to an RQL query string.

    Filters and aggregations sit side by side under a top-level `and`.

    Raises:
        QueryError: If a filter leaf uses a comparator with no RQL equivalent
            (`null`, `startswith`, `endswith`, ...).
    """
    nodes: list[dict[str, Any]] = []
    if expr is not None and expr:
        filter_ast = expr_to_rql(expr)
        # flatten a top-level `and` filter so aggregations join as siblings
        if not expr.negated and filter_ast.get("name") == "and":
            nodes.extend(filter_ast["args"])
        else:
            nodes.append(filter_ast)
    nodes.extend(_aggs_to_rql(aggs))
    if not nodes:
        return ""
    if len(nodes) == 1:
        return str(pyrql.unparse(nodes[0]))
    return str(pyrql.unparse({"name": "and", "args": nodes}))

Aggregations

The A projection node, the immutable Agg spec and the in-memory Aggregator. See the aggregation guide.

Aggregations for the ftmq query language.

An aggregation is a projection over the matched entities (a SELECT-list / GROUP BY concern), not a filter predicate: the A node does not compose with the & | ~ boolean tree the M/P/G/C filter nodes build. It is declared with Query.aggregate, parallel to where() and order_by().

A(sum="amountEur", by="beneficiary") builds one immutable Agg spec per func=prop pair. Aggregator is the in-memory accumulator that runs those specs over a stream of entities; the SQL backend reads the same specs (see [ftmq.query.sql][]).

A

An aggregation projection node: A(sum="amountEur", by="beneficiary").

Each keyword is an aggregation function (min, max, sum, avg, count) whose value is the property (or properties) to aggregate; by= groups by one or more properties / fields. Unlike the M/P/G/C filter nodes, A is not a boolean leaf - it does not compose with & | ~; pass it to Query.aggregate.

Examples:

A(sum="amountEur", by="beneficiary")
A(count="id")
A(sum=["amountEur", "amount"])
Source code in ftmq/query/aggregations.py
class A:
    """An aggregation projection node: `A(sum="amountEur", by="beneficiary")`.

    Each keyword is an aggregation function (`min`, `max`, `sum`, `avg`,
    `count`) whose value is the property (or properties) to aggregate; `by=`
    groups by one or more properties / fields. Unlike the `M`/`P`/`G`/`C`
    filter nodes, `A` is not a boolean leaf - it does not compose with
    `& | ~`; pass it to [`Query.aggregate`][ftmq.Query.aggregate].

    Examples:
        ```python
        A(sum="amountEur", by="beneficiary")
        A(count="id")
        A(sum=["amountEur", "amount"])
        ```
    """

    def __init__(
        self,
        *,
        by: str | Iterable[str] | None = None,
        **funcs: str | Iterable[str],
    ) -> None:
        groups: tuple[str, ...] = tuple(str(g) for g in ensure_list(by))
        aggs: list[Agg] = []
        for func, props in funcs.items():
            for prop in ensure_list(props):
                aggs.append(make_agg(func, prop, groups))
        if not aggs:
            raise QueryError("Empty aggregation: pass at least one `func=prop`")
        self.aggs: tuple[Agg, ...] = tuple(aggs)

Agg dataclass

An immutable aggregation spec: a function over a property, optionally grouped. Built via the A node or Query.aggregate.

Source code in ftmq/query/aggregations.py
@dataclass(frozen=True)
class Agg:
    """An immutable aggregation spec: a function over a property, optionally
    grouped. Built via the [`A`][ftmq.query.aggregations.A] node or
    [`Query.aggregate`][ftmq.Query.aggregate]."""

    func: str
    prop: str
    groups: tuple[str, ...] = ()

    def values(self, proxy: Entity, prop: str | None = None) -> Iterator[str]:
        """Yield the entity values for this spec's property (or a group prop)."""
        prop = prop or self.prop
        if prop == "id":
            if proxy.id is not None:
                yield proxy.id
        elif prop == "dataset":
            yield from proxy.datasets
        elif prop == "schema":
            yield proxy.schema.name
        elif prop == "year":
            for value in proxy.get_type_values(registry.date):
                yield value[:4]
        else:
            yield from proxy.get(prop, quiet=True)

values(proxy, prop=None)

Yield the entity values for this spec's property (or a group prop).

Source code in ftmq/query/aggregations.py
def values(self, proxy: Entity, prop: str | None = None) -> Iterator[str]:
    """Yield the entity values for this spec's property (or a group prop)."""
    prop = prop or self.prop
    if prop == "id":
        if proxy.id is not None:
            yield proxy.id
    elif prop == "dataset":
        yield from proxy.datasets
    elif prop == "schema":
        yield proxy.schema.name
    elif prop == "year":
        for value in proxy.get_type_values(registry.date):
            yield value[:4]
    else:
        yield from proxy.get(prop, quiet=True)

Aggregator

In-memory accumulator: runs a set of Agg specs over an entity stream.

A fresh instance per run holds all mutable state, so applying the same query twice never double-counts (the specs themselves are immutable).

Source code in ftmq/query/aggregations.py
class Aggregator:
    """In-memory accumulator: runs a set of [`Agg`][ftmq.query.aggregations.Agg]
    specs over an entity stream.

    A fresh instance per run holds all mutable state, so applying the same
    query twice never double-counts (the specs themselves are immutable).
    """

    def __init__(self, aggs: Iterable[Agg]) -> None:
        self.aggs: list[Agg] = list(aggs)
        self._values: dict[Agg, Values] = defaultdict(list)
        self._grouped: dict[Agg, dict[str, dict[str, Values]]] = defaultdict(
            lambda: defaultdict(lambda: defaultdict(list))
        )

    def collect(self, proxy: Entity) -> None:
        """Accumulate one entity's values into every spec."""
        for agg in self.aggs:
            is_numeric = prop_is_numeric(proxy.schema, agg.prop)
            for raw in agg.values(proxy):
                value: Any = registry.number.to_number(raw) if is_numeric else raw
                if value is None:
                    continue
                self._values[agg].append(value)
                for group in agg.groups:
                    for g in agg.values(proxy, group):
                        self._grouped[agg][group][g].append(value)

    def apply(self, proxies: Iterable[Entity]) -> Iterator[Entity]:
        """Collect every entity while passing the stream through unchanged."""
        for proxy in proxies:
            self.collect(proxy)
            yield proxy

    @property
    def result(self) -> AggregatorResult:
        """The reduced result: `{func: {prop: value}, "groups": {group: {func:
        {prop: {group_value: value}}}}}` (empties removed)."""
        res: Any = defaultdict(dict)
        groups: Any = defaultdict(lambda: defaultdict(dict))
        for agg in self.aggs:
            res[agg.func][agg.prop] = reduce_values(agg.func, self._values[agg])
            for group in agg.groups:
                groups[group][agg.func][agg.prop] = {
                    g: reduce_values(agg.func, values)
                    for g, values in self._grouped[agg][group].items()
                }
        res["groups"] = groups
        return clean_dict(res)

result property

The reduced result: {func: {prop: value}, "groups": {group: {func: {prop: {group_value: value}}}}} (empties removed).

apply(proxies)

Collect every entity while passing the stream through unchanged.

Source code in ftmq/query/aggregations.py
def apply(self, proxies: Iterable[Entity]) -> Iterator[Entity]:
    """Collect every entity while passing the stream through unchanged."""
    for proxy in proxies:
        self.collect(proxy)
        yield proxy

collect(proxy)

Accumulate one entity's values into every spec.

Source code in ftmq/query/aggregations.py
def collect(self, proxy: Entity) -> None:
    """Accumulate one entity's values into every spec."""
    for agg in self.aggs:
        is_numeric = prop_is_numeric(proxy.schema, agg.prop)
        for raw in agg.values(proxy):
            value: Any = registry.number.to_number(raw) if is_numeric else raw
            if value is None:
                continue
            self._values[agg].append(value)
            for group in agg.groups:
                for g in agg.values(proxy, group):
                    self._grouped[agg][group][g].append(value)

aggregations_from_dict(data)

Rebuild aggregation specs from the output of aggregations_to_dict, restoring each spec's groups from the nested groups mapping.

Source code in ftmq/query/aggregations.py
def aggregations_from_dict(data: dict[str, Any]) -> set[Agg]:
    """Rebuild aggregation specs from the output of
    [`aggregations_to_dict`][ftmq.query.aggregations.aggregations_to_dict],
    restoring each spec's groups from the nested `groups` mapping."""
    data = dict(data)
    nested = data.pop("groups", None) or {}
    groups_by_agg: dict[tuple[str, str], set[str]] = defaultdict(set)
    for group, funcs in nested.items():
        for func, props in funcs.items():
            for prop in ensure_list(props):
                groups_by_agg[(func, prop)].add(group)
    aggs: set[Agg] = set()
    for func, props in data.items():
        for prop in ensure_list(props):
            groups = tuple(sorted(groups_by_agg.get((func, prop), ())))
            aggs.add(make_agg(func, prop, groups))
    return aggs

aggregations_to_dict(aggs)

Serialize aggregation specs to the query to_dict shape: {func: {props}, "groups": {group: {func: {props}}}}.

Source code in ftmq/query/aggregations.py
def aggregations_to_dict(aggs: Iterable[Agg]) -> dict[str, Any]:
    """Serialize aggregation specs to the query `to_dict` shape:
    `{func: {props}, "groups": {group: {func: {props}}}}`."""
    data: dict[str, Any] = defaultdict(set)
    data["groups"] = defaultdict(lambda: defaultdict(set))
    for agg in aggs:
        data[agg.func].add(agg.prop)
        for group in agg.groups:
            data["groups"][group][agg.func].add(agg.prop)
    return clean_dict(data)

make_agg(func, prop, groups=())

Validate and build a single Agg spec.

Source code in ftmq/query/aggregations.py
def make_agg(func: str, prop: str, groups: Iterable[str] = ()) -> Agg:
    """Validate and build a single [`Agg`][ftmq.query.aggregations.Agg] spec."""
    if func not in FUNCTIONS:
        raise QueryError(f"Invalid aggregation function: `{func}`")
    return Agg(
        func=func,
        prop=_validate_field(prop),
        groups=tuple(_validate_field(g) for g in groups),
    )

reduce_values(func, values)

Reduce collected values with an aggregation function (None if empty).

Source code in ftmq/query/aggregations.py
def reduce_values(func: str, values: Values) -> Value | None:
    """Reduce collected values with an aggregation function (`None` if empty)."""
    if not values:
        return None
    if func == "min":
        return min(values)
    if func == "max":
        return max(values)
    if func == "sum":
        return sum(cast("list[float]", values))
    if func == "avg":
        return statistics.mean(cast("list[float]", values))
    if func == "count":
        return len(set(values))
    return None

SQL

The SQL translation. A store passes its SqlSource to Query.compile (or builds Sql(query, source) directly).

Describes the SQL statement source a Query compiles against: the SQLAlchemy table (or view), the entity-identity column, and an optional partition-pruning rule.

Stores own one and pass it to Sql / Query.compile, replacing the old query.table mutation. A downstream store with extra columns (a lake / sharded table) supplies its own SqlSource so the same Query compiles against it unchanged.

Parameters:

Name Type Description Default
table Any

The SQLAlchemy Table / TableClause to query.

required
id_column str

The entity-identity column name (default canonical_id).

'canonical_id'
prune dict[str, PruneFn] | None

Optional {meta_field: fn} mapping folding a partition filter into every compiled query - e.g. {"schema": get_schema_bucket} maps a schema/schemata filter to a prune_column IN (...) predicate.

None
prune_column str | None

The partition column the prune values target (e.g. bucket).

None
Source code in ftmq/query/sql.py
class SqlSource:
    """Describes the SQL statement source a [`Query`][ftmq.Query] compiles
    against: the SQLAlchemy table (or view), the entity-identity column, and an
    optional partition-pruning rule.

    Stores own one and pass it to [`Sql`][ftmq.query.sql.Sql] /
    [`Query.compile`][ftmq.Query.compile], replacing the old
    `query.table` mutation. A downstream store with extra columns (a lake /
    sharded table) supplies its own `SqlSource` so the same `Query` compiles
    against it unchanged.

    Args:
        table: The SQLAlchemy `Table` / `TableClause` to query.
        id_column: The entity-identity column name (default `canonical_id`).
        prune: Optional `{meta_field: fn}` mapping folding a partition filter
            into every compiled query - e.g. `{"schema": get_schema_bucket}`
            maps a schema/schemata filter to a `prune_column IN (...)` predicate.
        prune_column: The partition column the `prune` values target
            (e.g. `bucket`).
    """

    def __init__(
        self,
        table: "Any",
        id_column: str = "canonical_id",
        prune: "dict[str, PruneFn] | None" = None,
        prune_column: str | None = None,
    ) -> None:
        self.table = table
        self.id_column = id_column
        self.prune = prune or {}
        self.prune_column = prune_column
Source code in ftmq/query/sql.py
class Sql:
    COMPARATORS = {
        Comparators["eq"]: "__eq__",
        Comparators["not"]: "__ne__",
        Comparators["in"]: "in_",
        Comparators.null: "is_",
        Comparators.gt: "__gt__",
        Comparators.gte: "__ge__",
        Comparators.lt: "__lt__",
        Comparators.lte: "__le__",
    }

    def __init__(self, q: "Query", source: "SqlSource | None" = None) -> None:
        self.q = q
        self.metadata = MetaData()
        if source is None:
            source = SqlSource(make_statement_table(self.metadata))
        self.source = source
        self.table = source.table
        self.id_col = self.table.c[source.id_column]
        self.META_COLUMNS = {
            "id": self.id_col,
            "dataset": self.table.c.dataset,
            "schema": self.table.c.schema,
        }

    def get_expression(self, column: Column, f: Leaf):
        value = f.value
        if f.comparator in (Comparators.ilike, Comparators.like):
            value = f"%{value}%"
        op = self.COMPARATORS.get(str(f.comparator), str(f.comparator))
        op = getattr(column, op)
        return op(value)

    @cached_property
    def clause(self) -> BooleanClauseList:
        clauses = []
        if self.q.ids:
            clauses.append(
                or_(
                    self.get_expression(self.table.c[f.key], f)
                    for f in sorted(self.q.ids)
                )
            )
        if self.q.datasets:
            clauses.append(
                or_(
                    self.get_expression(self.table.c.dataset, f)
                    for f in sorted(self.q.datasets)
                )
            )
        if self.q.schemata:
            clauses.append(
                or_(
                    self.get_expression(self.table.c.schema, f)
                    for f in sorted(self.q.schemata)
                )
            )
        # is-a schema filters (`M(schemata=...)`): expand to the schema plus its
        # non-abstract descendants and match on the `schema` column
        for f in sorted(s for s in self.q._leaves if isinstance(s, SchemataLeaf)):
            names: set[str] = set()
            for schema in f.schemata:
                names.add(schema.name)
                names.update(d.name for d in schema.descendants if not d.abstract)
            if str(f.comparator) in ("not", "not_in"):
                clauses.append(self.table.c.schema.not_in(names))
            else:
                clauses.append(self.table.c.schema.in_(names))
        # context / storage columns (`C(origin=...)`, `C(fragment=...)`, ...)
        context_by_key: dict[str, list[ContextLeaf]] = defaultdict(list)
        for f in self.q.context:
            context_by_key[f.key].append(f)
        for key, fs in context_by_key.items():
            if key not in self.table.c:
                raise QueryError(f"Unknown context column: `{key}`")
            clauses.append(
                or_(self.get_expression(self.table.c[key], f) for f in sorted(fs))
            )
        if self.q.properties:
            clauses.append(
                or_(
                    and_(
                        self.table.c.prop == f.key,
                        self.get_expression(self.table.c.value, f),
                    )
                    for f in sorted(self.q.properties)
                )
            )
        # prop-type groups: `G(countries=...)`, `G(entities=...)` (reverse), ...
        # Each is an entity-level membership ("the entity has a row of this
        # prop_type matching the value"), so it lifts to a `canonical_id`
        # subquery rather than a plain row predicate - the reverse `entities`
        # group is not special, it is just the `entity` prop-type. A row
        # predicate would force one row to satisfy both the group and any
        # property filter at once, which no single statement row can.
        if self.q.groups:
            gclause = or_(
                and_(
                    self.table.c.prop_type == str(f.prop_type),
                    self.get_expression(self.table.c.value, f),
                )
                for f in sorted(self.q.groups)
            )
            clauses.append(
                self.id_col.in_(select(self.id_col.distinct()).where(gclause))
            )
        # partition pruning: a schema/schemata filter narrows to the matching
        # partition values (e.g. the lake `bucket` column), folded into every
        # compiled query - so `count` prunes partitions too, not just statements
        prune_fn = self.source.prune.get("schema")
        if (
            prune_fn is not None
            and self.source.prune_column
            and self.q.schemata_names
            and self.source.prune_column in self.table.c
        ):
            values = {prune_fn(s) for s in self.q.schemata_names}
            clauses.append(self.table.c[self.source.prune_column].in_(values))
        return and_(*clauses)

    @cached_property
    def canonical_ids(self) -> Select:
        q = select(self.id_col.distinct()).where(self.clause)
        if self.q.sort is None:
            # offset 0 (a start-less slice) is redundant; omit it from the SQL
            q = q.limit(self.q.limit).offset(self.q.offset or None)
        return q

    @cached_property
    def all_canonical_ids(self) -> Select:
        return self.canonical_ids.limit(None).offset(None)

    @cached_property
    def _unsorted_statements(self) -> Select:
        where = self.clause
        if self.q.properties or self.q.groups or self.q.context or self.q.limit:
            where = self.id_col.in_(self.canonical_ids)
        return select(self.table).where(where).order_by(self.id_col)

    @cached_property
    def _sorted_statements(self) -> Select:
        if self.q.sort:
            if len(self.q.sort.values) > 1:
                raise ValueError(
                    f"Multi-valued sort not supported for `{self.__class__.__name__}`"
                )
            prop = self.q.sort.values[0]
            value = self.table.c.value
            if PropertyTypesMap[prop].value == registry.number:
                value = func.cast(self.table.c.value, NUMERIC)
            group_func = func.min if self.q.sort.ascending else func.max
            inner = (
                select(
                    self.id_col,
                    group_func(value).label("sortable_value"),
                )
                .where(
                    and_(
                        self.table.c.prop == prop,
                        self.id_col.in_(self.canonical_ids),
                    )
                )
                .group_by(self.id_col)
                .limit(self.q.limit)
                .offset(self.q.offset or None)
            )

            order_by = "sortable_value"
            if not self.q.sort.ascending:
                order_by = desc(order_by)
            order_by = [order_by, self.id_col]

            inner = inner.order_by(*order_by)

            return select(
                self.table.join(inner, self.id_col == inner.c.canonical_id)
            ).order_by(*order_by)

    @cached_property
    def statements(self) -> Select:
        if self.q.sort:
            return self._sorted_statements
        return self._unsorted_statements

    @cached_property
    def count(self) -> Select:
        return (
            select(func.count(self.id_col.distinct()))
            .select_from(self.table)
            .where(self.clause)
        )

    def _get_lookup_column(self, field: Field) -> Column:
        if field in self.META_COLUMNS:
            return self.META_COLUMNS[field]
        if isinstance(field, PropertyType):
            return self.table.c.prop_type
        if field in Properties:
            return self.table.c.prop
        if field in PropertyTypes or field == Fields.year:
            return self.table.c.prop_type
        raise NotImplementedError("Unknown field: `%s`" % field)

    def get_group_counts(
        self,
        group: Field,
        limit: int | None = None,
        extra_where: BooleanClauseList | None = None,
    ) -> Select:
        count = func.count(self.id_col.distinct()).label("count")
        column = self._get_lookup_column(group)
        group = str(group)
        if group in self.META_COLUMNS:
            grouper = column
            where = self.clause
        else:
            grouper = self.table.c.value
            where = and_(column == group, self.id_col.in_(self.all_canonical_ids))
        if extra_where is not None:
            where = and_(where, extra_where)
        return (
            select(grouper, count)
            .where(where)
            .group_by(grouper)
            .order_by(desc(count))
            .limit(limit)
        )

    @cached_property
    def datasets(self) -> Select:
        return self.get_group_counts("dataset")

    @cached_property
    def schemata(self) -> Select:
        return self.get_group_counts("schema")

    @cached_property
    def countries(self) -> Select:
        return self.get_group_counts(registry.country)

    @cached_property
    def countries_flat(self) -> Select:
        return select(self.table.c.value.distinct()).where(
            and_(
                self.table.c.prop_type == str(registry.country),
                self.id_col.in_(self.all_canonical_ids),
            )
        )

    @cached_property
    def things(self) -> Select:
        return self.get_group_counts(
            "schema", extra_where=self.table.c.schema.in_([str(x) for x in Things])
        )

    @cached_property
    def things_countries(self) -> Select:
        return self.get_group_counts(
            registry.country,
            extra_where=self.table.c.schema.in_([str(x) for x in Things]),
        )

    @cached_property
    def intervals(self) -> Select:
        return self.get_group_counts(
            "schema", extra_where=self.table.c.schema.in_([str(x) for x in Intervals])
        )

    @cached_property
    def intervals_countries(self) -> Select:
        return self.get_group_counts(
            registry.country,
            extra_where=self.table.c.schema.in_([str(x) for x in Intervals]),
        )

    @cached_property
    def dates(self) -> Select:
        return self.get_group_counts(registry.date)

    @cached_property
    def date_range(self) -> Select:
        return select(
            func.min(self.table.c.value),
            func.max(self.table.c.value),
        ).where(
            self.table.c.prop_type == "date",
            self.id_col.in_(self.all_canonical_ids),
        )

    @cached_property
    def aggregations(self) -> Select:
        qs = []
        for agg in sorted(self.q.aggregations, key=lambda a: (a.func, a.prop)):
            sql_agg = getattr(func, agg.func)
            sql_agg_value = self.table.c.value
            if agg.func == Aggregations.count:
                sql_agg_value = distinct(sql_agg_value)
            elif agg.func in (Aggregations.sum, Aggregations.avg):
                sql_agg_value = func.cast(sql_agg_value, NUMERIC)
            aggregator = sql_agg(sql_agg_value)
            qs.append(
                select(
                    text(f"'{agg.prop}'"),
                    text(f"'{agg.func}'"),
                    aggregator,
                ).where(
                    self.table.c.prop == str(agg.prop),
                    self.id_col.in_(self.all_canonical_ids),
                )
            )
        return union_all(*qs)

    def _get_grouping_where(self, grouper: Field, value: str) -> BooleanClauseList:
        column = self._get_lookup_column(grouper)
        clauses = [self.id_col.in_(self.all_canonical_ids)]
        if grouper in Properties:
            clauses.extend([column == str(grouper), self.table.c.value == value])
            return clauses
        if grouper == Fields.year:
            clauses.extend(
                [
                    column == str(registry.date),
                    func.substring(self.table.c.value, 1, 4) == str(value),
                ]
            )
            return clauses
        clauses.append(column == value)
        return clauses

    def get_group_aggregations(self, grouper: Field, group: str) -> Select:
        qs = []
        for agg in sorted(self.q.aggregations, key=lambda a: (a.func, a.prop)):
            if grouper in agg.groups:
                if agg.prop in self.META_COLUMNS:
                    sql_agg_value = self._get_lookup_column(agg.prop)
                else:
                    sql_agg_value = self.table.c.value
                sql_agg = getattr(func, agg.func)
                if agg.func == Aggregations.count:
                    sql_agg_value = distinct(sql_agg_value)
                elif agg.func in (Aggregations.sum, Aggregations.avg):
                    sql_agg_value = func.cast(sql_agg_value, NUMERIC)
                aggregator = sql_agg(sql_agg_value)

                inner = select(self.id_col.distinct()).where(
                    *self._get_grouping_where(grouper, group)
                )

                qs.append(
                    select(
                        text(f"'{agg.prop}'"),
                        text(f"'{agg.func}'"),
                        aggregator,
                    ).where(
                        self.table.c.prop == str(agg.prop),
                        self.id_col.in_(inner),
                    )
                )
        return union_all(*qs)

    @cached_property
    def group_props(self) -> set[Field]:
        props: set[Field] = set()
        for agg in self.q.aggregations:
            if agg.groups:
                props.update(agg.groups)
        return props

Errors

Bases: ValueError

Raised for an invalid query: an unknown field, an invalid comparator, or a query that cannot be projected to the requested serialization.

Subclasses ValueError so existing except ValueError handlers keep working.

Source code in ftmq/query/exceptions.py
1
2
3
4
5
6
7
class QueryError(ValueError):
    """Raised for an invalid query: an unknown field, an invalid comparator,
    or a query that cannot be projected to the requested serialization.

    Subclasses `ValueError` so existing `except ValueError` handlers keep
    working.
    """