Query Builders

mosaicolabs.query.builders ¶

This module provides the high-level "Fluent" API for constructing complex searches across the Mosaico Data Platform.

It implements a Domain-Specific Language that allows users to filter Sequences, Topics, and Ontology data using a type-safe, method-chaining interface.

Key Components:

Query: The root container that aggregates multiple specialized sub-queries.
QueryOntologyCatalog: For fine-grained filtering based on sensor-specific field values (e.g., IMU.Q.acceleration.x > 9.8).
QueryTopic: Specifically for filtering topic-level metadata.
QuerySequence: Specifically for filtering sequence-level metadata.

QueryOntologyCatalog ¶

QueryOntologyCatalog(*expressions)

A top-level query object for the Data Catalog that combines multiple sensor-field expressions.

This builder allows for fine-grained filtering based on the actual values contained within sensor payloads (e.g., IMU acceleration, GPS coordinates, or custom telemetry). It produces a "flat" dictionary output where field paths utilize dot-notation (e.g., "IMU.acceleration.x").

This class is designed to work with the .Q query proxy injected into every Serializable data ontology model. You can use this proxy on any registered sensor class (like IMU, Vector3d, Point3d), etc. to create type-safe expressions.

Example

from mosaicolabs import MosaicoClient, IMU, QueryOntologyCatalog

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for a specific data value (using constructor)
    qresponse = client.query(
        QueryOntologyCatalog(IMU.Q.acceleration.x.lt(-4.0)) # Using constructor
        .with_expression(IMU.Q.acceleration.y.gt(5.0)) # Using with_expression
    )

    # Inspect the response
    if qresponse is not None:
        # Results are automatically grouped by Sequence for easier data management
        for item in qresponse:
            print(f"Sequence: {item.sequence.name}")
            print(f"Topics: {[topic.name for topic in item.topics]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

The constructor initializes the query with an optional list of _QueryCatalogExpression objects, generated via <Model>.Q. proxy, where model is any of the available data ontology (e.g. IMU.Q, GPS.Q, String.Q, etc.)

Parameters:

Name	Type	Description	Default
`*expressions`	`_QueryExpression`	A variable number of expressions, generated via the `.Q` proxy on an ontology model.	`()`

Raises:

Type	Description
`TypeError`	If an expression is not of the supported type.
`ValueError`	If an operator does not start with the required '$' prefix.
`NotImplementedError`	If a duplicate key (field path) is detected within the same query.

with_expression ¶

with_expression(expr)

Adds a new _QueryCatalogExpression expression to the query using a fluent interface.

Example

from mosaicolabs import MosaicoClient, Topic, QuerySequence

with MosaicoClient.connect("localhost", 6726) as client:
    # Chain multiple sensor filters together
    qresponse = client.query(
        QueryOntologyCatalog()
        .with_expression(GPS.Q.status.satellites.geq(8))
        .with_expression(GPS.Q.position.x.between([44.0, 45.0]))
    )

    # Inspect the response
    if qresponse is not None:
        # Results are automatically grouped by Sequence for easier data management
        for item in qresponse:
            print(f"Sequence: {item.sequence.name}")
            print(f"Topics: {[topic.name for topic in item.topics]}")

            # Clusterize all topics within the sequence to extract the time intervals
            clusters_dict = item.clusterize_all()

            # Since clusterize_all() used default clustering_dt_ns, each topic will have
            # just one cluster representing the first and last moment the query was satisfied
            for t_name, clusters in clusters_dict.items():
                print(f"{t_name}:\n", "\n".join(f"{cluster}" for cluster in clusters))

Parameters:

Name	Type	Description	Default
`expr`	`_QueryExpression`	A valid expression generated via the `.Q` proxy on an ontology model, e.g., `GPS.Q.status.satellites.leq(10)`.	required

Returns:

Name	Type	Description
`QueryOntologyCatalog`	`QueryOntologyCatalog`	The `QueryOntologyCatalog` instance for method chaining.

name ¶

name()

Returns the top-level key ('ontology') used for nesting inside a root Query.

to_dict ¶

to_dict()

Serializes the ontology expressions into a flat dictionary for the platform API.

Example Output

{"IMU.timestamp_ns": {"$between": [...]}, "IMU.acceleration.x": {"$leq": 10}}

Returns:

Type	Description
`Dict[str, Any]`	Dict[str, Any]: A dictionary containing all merged sensor-field expressions.

expressions ¶

expressions()

Return the list of query expressions.

QueryTopic ¶

QueryTopic()

A top-level query object for Topic data that combines multiple expressions with a logical AND.

This builder handles the complex partitioning required to query both flat system fields (like name or ontology_tag) and nested dictionary fields (like user_metadata). The resulting dictionary output preserves this hierarchical structure for server-side processing.

Example

from mosaicolabs import MosaicoClient, Image, QueryTopic

with MosaicoClient.connect("localhost", 6726) as client:
    # Query for all 'image' topics created in a specific timeframe, matching some metadata (key, value) pair
    qresponse = client.query(
        QueryTopic()
        .with_ontology_tag(Image.ontology_tag())
        .with_created_timestamp(time_start=1700000000_000000000)
        .with_user_metadata("camera_id.serial_number", eq="ABC123_XYZ")
    )

    # Inspect the response
    if qresponse is not None:
        # Results are automatically grouped by Sequence for easier data management
        for item in qresponse:
            print(f"Sequence: {item.sequence.name}")
            print(f"Topics: {[topic.name for topic in item.topics]}")

The constructor initializes an empty query builder

with_user_metadata ¶

with_user_metadata(key, **operator_kwargs)

Appends a metadata filter to the query using a fluent, operator-based interface.

This method simplifies metadata discovery by allowing direct filtering on the user_metadata dictionary of the Topic. Each call adds a logical AND condition to the query.

Parameters:

Name	Type	Description	Default
`key`	`str`	The metadata key to filter on (e.g., "sensor_id"). Supports dot-notation for nested dictionary access (e.g., "calibration.focal_length").	required
`**operator_kwargs`	`Any`	A single keyword argument where the key is the operator and the value is the comparison target, e.g. `eq="value"`, `lt=100`, `ex=False`, etc.	`{}`

Raises:

Type	Description
`ValueError`	If no operator is provided, if multiple operators are provided in a single call, or if an unsupported operator is used.

Supported Operators

eq: Equal to
neq: Not equal to
gt: Greater than
geq: Greater than or equal to
lt: Less than
leq: Less than or equal to
match: glob-style pattern matching, see QueryableString.match
between: Range filter (expects a list of [min, max])
outside: Range filter (expects a list of [min, max])
in_: One-of (expects a list of values)
ex: The key exist (ex=True) or does not exist (ex=False)

Example

# Find sequences for 'Apollo' project with visibility under 100m
query = (
    QueryTopic()
    .with_user_metadata("sensor_id", eq="ABC123_XYZ")
    .with_user_metadata("calibration.focal_length", between=(14, 24))
)
results = client.query(query)

Returns:

Name	Type	Description
`QueryTopic`	`QueryTopic`	The current instance to support method chaining.

with_name ¶

with_name(name)

Adds an exact match filter for the topic 'name' field.

Example

from mosaicolabs import MosaicoClient, Topic, QueryTopic

with MosaicoClient.connect("localhost", 6726) as client:
    # Target a specific known topic path
    qresponse = client.query(
        QueryTopic().with_name("vehicle/front/camera")
    )

    # Inspect the response
    if qresponse is not None:
        # Results are automatically grouped by Sequence for easier data management
        for item in qresponse:
            print(f"Sequence: {item.sequence.name}")
            print(f"Topics: {[topic.name for topic in item.topics]}")

Parameters:

Name	Type	Description	Default
`name`	`str`	The exact name of the topic to match.	required

Returns:

Name	Type	Description
`QueryTopic`	`QueryTopic`	The `QueryTopic` instance for method chaining.

with_name_match ¶

with_name_match(name)

Adds a RegEx-like filter for the topic 'name' field. Supported operators are:

*: matches a multiple (zero or more) characters, including space.
?: matches a single (exactly one) characters, including space.
[]: matches a character set. Examples: [aeiou] to match any vocals, or [a-z] to match a range
#: matches any single digit (0 — 9). Shortcut for [0-9]

Supposing the server contains a topic with name car1/imu/front we can get it using:

*imu*
car#/[a-z]*/[a-z]**
car?/[umi]*/?????

Note

If name contains none of the wildcards above, the operator is equivalent to .with_name(), i.e. an exact match will be performed.

Example

from mosaicolabs import MosaicoClient, Topic, QuerySequence

with MosaicoClient.connect("localhost", 6726) as client:
    # Search for all topics containing the word 'camera'
    qresponse = client.query(
        QueryTopic().with_name_match("/[a-z]/camera")
    )

    # Inspect the response
    if qresponse is not None:
        # Results are automatically grouped by Sequence for easier data management
        for item in qresponse:
            print(f"Sequence: {item.sequence.name}")
            print(f"Topics: {[topic.name for topic in item.topics]}")

Parameters:

Name	Type	Description	Default
`name`	`str`	The string pattern to search for within the topic name.	required

Returns:

Name	Type	Description
`QueryTopic`	`QueryTopic`	The `QueryTopic` instance for method chaining.

with_ontology_tag ¶

with_ontology_tag(ontology_tag)

Adds an exact match filter for the 'ontology_tag' field.

This filter restricts the search to topics belonging to a specific data type identifier (e.g., 'imu', 'gnss').

Example

from mosaicolabs import MosaicoClient, Topic, QuerySequence

with MosaicoClient.connect("localhost", 6726) as client:
    # Filter for IMU-only data streams
    qresponse = client.query(
        QueryTopic().with_ontology_tag(IMU.ontology_tag())
    )

    # Inspect the response
    if qresponse is not None:
        # Results are automatically grouped by Sequence for easier data management
        for item in qresponse:
            print(f"Sequence: {item.sequence.name}")
            print(f"Topics: {[topic.name for topic in item.topics]}")

Note: To ensure compatibility and avoid hardcoding strings, it is highly recommended to retrieve the tag dynamically using the ontology_tag() method of the desired ontology class.

Parameters:

Name	Type	Description	Default
`ontology_tag`	`str`	The string tag (e.g., 'imu', 'gps') to filter by.	required

Returns:

Name	Type	Description
`QueryTopic`	`QueryTopic`	The `QueryTopic` instance for method chaining.

with_created_timestamp ¶

with_created_timestamp(time_start=None, time_end=None)

Adds a filter for the 'created_at_ns' field using high-precision Time.

Example

from mosaicolabs import MosaicoClient, QueryTopic

with MosaicoClient.connect("localhost", 6726) as client:
    # Find sequences created during a specific day
    qresponse = client.query(
        QueryTopic().with_created_timestamp(
            time_start=1704067200_000000000, # 2024-01-01
            time_end=1704153600_000000000    # 2024-01-02
        )
    )

    # Inspect the response
    if qresponse is not None:
        # Results are automatically grouped by Sequence for easier data management
        for item in qresponse:
            print(f"Sequence: {item.sequence.name}")
            print(f"Topics: {[topic.name for topic in item.topics]}")

Parameters:

Name	Type	Description	Default
`time_start`	`Optional[int]`	Optional lower bound (inclusive).	`None`
`time_end`	`Optional[int]`	Optional upper bound (inclusive).	`None`

Returns:

Name	Type	Description
`QueryTopic`	`QueryTopic`	The `QueryTopic` instance for method chaining.

Raises:

Type	Description
`ValueError`	If both bounds are None or if `time_start > time_end`.

name ¶

name()

Returns the top-level key ('topic') used when nesting this query inside a root Query.

Returns:

Name	Type	Description
`str`	`str`	The string "topic" indicating the query type.

to_dict ¶

to_dict()

Serializes the query into a nested dictionary for the platform API.

This method partitions expressions into two groups:

System Fields: Standard fields like name are kept in the root dictionary.
Metadata Fields: Fields starting with a dictionary-type model key (e.g., user_metadata) are stripped of their prefix and nested under that key.

Returns:

Type	Description
`Dict[str, Any]`	Dict[str, Any]: A dictionary representation of the query, e.g., `{"name": {"$eq": "..."}, "user_metadata": {"key": {"$eq": "..."}}}`.

expressions ¶

expressions()

The list of query expressions.

Returns:

Type	Description
`List[_QueryExpression]`	List[_QueryExpression]: The list of expressions currently stored in the query.

QuerySequence ¶

QuerySequence()

A top-level query object for Sequence data that combines multiple expressions with a logical AND.

This builder handles the complex partitioning required to query both flat system fields (like name) and nested dictionary fields (like user_metadata). The resulting dictionary output preserves this hierarchical structure for server-side processing.

Example

from mosaicolabs import MosaicoClient, QuerySequence

with MosaicoClient.connect("localhost", 6726) as client:
    # Search for sequences by project name and creation date
    qresponse = client.query(
        QuerySequence()
        .with_user_metadata("project", eq="Apollo")
        .with_created_timestamp(time_start=1690000000_000000000)
    )

    # Inspect the response
    for item in qresponse:
        print(f"Sequence: {item.sequence.name}")
        print(f"Topics: {[topic.name for topic in item.topics]}")

The constructor initializes an empty query builder

with_user_metadata ¶

with_user_metadata(key, **operator_kwargs)

Appends a metadata filter to the query using a fluent, operator-based interface.

This method simplifies metadata discovery by allowing direct filtering on the user_metadata dictionary of the Sequence. Each call adds a logical AND condition to the query.

Parameters:

Name	Type	Description	Default
`key`	`str`	The metadata key to filter on (e.g., "project"). Supports dot-notation for nested dictionary access (e.g., "vehicle.id").	required
`**operator_kwargs`	`Any`	A single keyword argument where the key is the operator and the value is the comparison target, e.g. `eq="value"`, `lt=100`, etc.	`{}`

Raises:

Type	Description
`ValueError`	If no operator is provided, if multiple operators are provided in a single call, or if an unsupported operator is used.

Supported Operators

eq: Equal to
neq: Not equal to
gt: Greater than
geq: Greater than or equal to
lt: Less than
leq: Less than or equal to
match: glob-style pattern matching, see QueryableString.match
between: Range filter (expects a list of [min, max])
outside: Range filter (expects a list of [min, max])
in_: One-of (expects a list of values)
ex: The key exist (ex=True) or does not exist (ex=False)

Example

# Find sequences for 'Apollo' project with visibility under 100m
query = (
    QuerySequence()
    .with_user_metadata("project", eq="Apollo")
    .with_user_metadata("environment.visibility", lt=100)
)
results = client.query(query)

Returns:

Name	Type	Description
`QuerySequence`	`QuerySequence`	The current instance to support method chaining.

with_name ¶

with_name(name)

Adds an exact match filter for the sequence 'name' field.

Example

from mosaicolabs import MosaicoClient, Topic, QuerySequence

with MosaicoClient.connect("localhost", 6726) as client:
    # Find all sequences with name equal to 'test_winter_01'
    qresponse = client.query(
        QuerySequence().with_name("test_winter_01")
    )

    # Inspect the response
    for item in qresponse:
        print(f"Sequence: {item.sequence.name}")
        print(f"Topics: {[topic.name for topic in item.topics]}")

Parameters:

Name	Type	Description	Default
`name`	`str`	The exact name of the sequence to match.	required

Returns:

Name	Type	Description
`QuerySequence`	`QuerySequence`	The `QuerySequence` instance for method chaining.

with_name_match ¶

with_name_match(name)

Adds a RegEx filter for the sequence 'name' field. Supported RegEx operations are:

*: matches a multiple (zero or more) characters, including space.
?: matches a single (exactly one) characters, including space.
[]: matches a character set. Examples: [aeiou] to match any vocals, or [a-z] to match a range
#: matches any single digit (0 — 9). Shortcut for [0-9]

Supposing the server contains a sequence with name experiment1-car we can get it using:

experiment*
[a-z]*1-car
experiment?/[a-z]*
experiment1-car

Note

If name contains none of the wildcards above, the operator is equivalent to .with_name(), i.e. an exact match will be performed.

Example

from mosaicolabs import MosaicoClient, Topic, QuerySequence

with MosaicoClient.connect("localhost", 6726) as client:
    # Find all sequences with name containing 'calibration_run_'
    qresponse = client.query(
        QuerySequence().with_name_match("calibration_run_*")
    )

    # Inspect the response
    for item in qresponse:
        print(f"Sequence: {item.sequence.name}")
        print(f"Topics: {[topic.name for topic in item.topics]}")

Parameters:

Name	Type	Description	Default
`name`	`str`	The string pattern to search for within the sequence name.	required

Returns:

Name	Type	Description
`QuerySequence`	`QuerySequence`	The `QuerySequence` instance for method chaining.

with_created_timestamp ¶

with_created_timestamp(time_start=None, time_end=None)

Adds a filter for the 'created_at_ns' field using high-precision Time.

Example

from mosaicolabs import MosaicoClient, QuerySequence

with MosaicoClient.connect("localhost", 6726) as client:
    # Find sequences created during a specific time range
    qresponse = client.query(
        QuerySequence().with_created_timestamp(
            time_start=1704067200_000000000, # 2024-01-01
            time_end=1704153600_000000000    # 2024-01-02
        )
    )

    # Inspect the response
    for item in qresponse:
        print(f"Sequence: {item.sequence.name}")
        print(f"Topics: {[topic.name for topic in item.topics]}")

Parameters:

Name	Type	Description	Default
`time_start`	`Optional[int]`	Optional lower bound (inclusive).	`None`
`time_end`	`Optional[int]`	Optional upper bound (inclusive).	`None`

Returns:

Name	Type	Description
`QuerySequence`	`QuerySequence`	The `QuerySequence` instance for method chaining.

Raises:

Type	Description
`ValueError`	If both bounds are `None` or if `time_start > time_end`.

name ¶

name()

The top-level key ('sequence') used for nesting inside a root Query.

Returns:

Name	Type	Description
`str`	`str`	The string "sequence" indicating the query type.

to_dict ¶

to_dict()

Serializes the query into a nested dictionary for the platform API.

This method partitions expressions into:

Normal Fields: Fields like name are kept in a flat dictionary.
Metadata Fields: Fields targeting user_metadata are collected and nested.

Returns:

Type	Description
`Dict[str, Any]`	Dict[str, Any]: A dictionary representation preserving the hierarchical structure.

expressions ¶

expressions()

The list of query expressions.

Returns:

Type	Description
`List[_QueryExpression]`	List[_QueryExpression]: The list of expressions currently stored in the query.

Query ¶

Query(*queries)

A top-level "root" query object that aggregates multiple specialized sub-queries into a single request body.

This class serves as the final envelope for multi-domain queries, ensuring that different query types (Topic, Sequence, Ontology) do not overwrite each other.

Example

from mosaicolabs import QueryOntologyCatalog, QuerySequence, Query, IMU, MosaicoClient

# Establish a connection to the Mosaico Data Platform
with MosaicoClient.connect("localhost", 6726) as client:
    # Build a filter with name pattern and metadata-related expression
    query = Query(
        # Append a filter for sequence metadata
        QuerySequence()
        .with_user_metadata("environment.visibility", lt=50)
        .with_name_match("test_drive*"),
        # Append a filter with deep time-series data discovery and measurement time windowing
        QueryOntologyCatalog()
        .with_expression(IMU.Q.acceleration.x.gt(5.0))
        .with_expression(IMU.Q.timestamp_ns.gt(1700134567))
    )
    # Perform the server side query
    qresponse = client.query(query=query)
    # Inspect the response
    if qresponse is not None:
        # Results are automatically grouped by Sequence for easier data management
        for item in qresponse:
            print(f"Sequence: {item.sequence.name}")
            print(
                f"Topics: {
                    {
                        topic.name: [
                            (cluster.timerange.start, cluster.timerange.end)
                            for cluster in topic.clusterize()
                        ]
                        for topic in item.topics
                    }
                }"
            )

Initializes the root query with a set of sub-queries.

Parameters:

Name	Type	Description	Default
`*queries`	`QueryableProtocol`	A variable number of sub-query objects (e.g., `QueryTopic()`, `QuerySequence()`).	`()`

Raises:

Type	Description
`ValueError`	If duplicate query types are detected in the initial arguments.

append ¶

append(*queries)

Adds additional sub-queries to the existing root query.

Parameters:

Name	Type	Description	Default
`*queries`	`QueryableProtocol`	Additional sub-query instances.	`()`

Raises:

Type	Description
`ValueError`	If an appended query type is already present in the request.

Example

from mosaicolabs import QueryOntologyCatalog, QuerySequence, Query, IMU, MosaicoClient

# Build a filter with name pattern and metadata-related expression
query = Query(
    # Append a filter for sequence metadata
    QuerySequence()
    .with_user_metadata("environment.visibility", lt=50)
    .with_name_match("test_drive???")
)

# Append a filter with deep time-series data discovery and measurement time windowing
query.append(
    QueryOntologyCatalog()
    .with_expression(IMU.Q.acceleration.x.gt(5.0))
    .with_expression(IMU.Q.timestamp_ns.gt(1700134567))
)

to_dict ¶

to_dict()

Serializes the entire multi-domain query into the final JSON dictionary.

It orchestrates the conversion by calling the .name() and .to_dict() methods of each contained sub-query.

Example Output

{
    "topic": { ... topic filters ... },
    "sequence": { ... sequence filters ... },
    "ontology": { ... ontology filters ... }
}

Returns:

Type	Description
`Dict[str, Any]`	Dict[str, Any]: The final aggregated query dictionary.