type_bridge.query

query

Query builder for TypeQL.

Query

Query()

Builder for TypeQL queries.

Initialize query builder.

Source code in type_bridge/query/__init__.py

def __init__(self):
    """Initialize query builder."""
    self.match_clause = MatchClause(patterns=[])
    self.delete_clause = DeleteClause(statements=[])
    self.insert_clause = InsertClause(statements=[])
    self.fetch_clause = FetchClause(items=[])

    # Modifiers
    self.sort_clauses: list[tuple[str, str]] = []
    self.offset_val: int | None = None
    self.limit_val: int | None = None

    self.compiler = QueryCompiler()

match

match(pattern)

Add a match clause.

Parameters:

Name	Type	Description	Default
pattern	str	TypeQL match pattern	required

Returns:

Type	Description
Query	Self for chaining

Source code in type_bridge/query/__init__.py

def match(self, pattern: str) -> Query:
    """Add a match clause.

    Args:
        pattern: TypeQL match pattern

    Returns:
        Self for chaining
    """
    if pattern:
        # Clean string pattern
        clean_pattern = pattern.strip().rstrip(";")
        # RawPattern doesn't strictly require 'variable' if handled by compiler as raw string
        # But we made it subclass Pattern again with 'variable' moved to subclasses.
        # RawPattern definition: content: str
        self.match_clause.patterns.append(RawPattern(content=clean_pattern))
    return self

fetch

fetch(variable, *attributes)

Add variables and attributes to fetch.

In TypeQL 3.x, fetch uses the syntax: fetch { $e.* } (fetch all attributes)

Parameters:

Name	Type	Description	Default
variable	str	Variable name to fetch (e.g., "$e")	required
attributes	str	Not used in TypeQL 3.x (kept for API compatibility)	()

Returns:

Type	Description
Query	Self for chaining

Example

query.fetch("$e") # Fetches all attributes

Source code in type_bridge/query/__init__.py

def fetch(self, variable: str, *attributes: str) -> Query:
    """Add variables and attributes to fetch.

    In TypeQL 3.x, fetch uses the syntax:
    fetch { $e.* }  (fetch all attributes)

    Args:
        variable: Variable name to fetch (e.g., "$e")
        attributes: Not used in TypeQL 3.x (kept for API compatibility)

    Returns:
        Self for chaining

    Example:
        query.fetch("$e")  # Fetches all attributes
    """
    # For TypeQL 3.x, default to wildcard fetch
    # Use variable name (without $) as the key
    key = variable.lstrip("$")
    self.fetch_clause.items.append(FetchWildcard(key=key, var=variable))
    return self

delete

delete(pattern)

Add a delete clause.

Parameters:

Name	Type	Description	Default
pattern	str	TypeQL delete pattern	required

Returns:

Type	Description
Query	Self for chaining

Source code in type_bridge/query/__init__.py

def delete(self, pattern: str) -> Query:
    """Add a delete clause.

    Args:
        pattern: TypeQL delete pattern

    Returns:
        Self for chaining
    """
    if pattern:
        clean_pattern = pattern.strip().rstrip(";")
        self.delete_clause.statements.append(RawStatement(content=clean_pattern))
    return self

insert

insert(pattern)

Add an insert clause.

Parameters:

Name	Type	Description	Default
pattern	str	TypeQL insert pattern	required

Returns:

Type	Description
Query	Self for chaining

Source code in type_bridge/query/__init__.py

def insert(self, pattern: str) -> Query:
    """Add an insert clause.

    Args:
        pattern: TypeQL insert pattern

    Returns:
        Self for chaining
    """
    if pattern:
        clean_pattern = pattern.strip().rstrip(";")
        self.insert_clause.statements.append(RawStatement(content=clean_pattern))
    return self

limit

limit(limit)

Set query limit.

Parameters:

Name	Type	Description	Default
limit	int	Maximum number of results	required

Returns:

Type	Description
Query	Self for chaining

Source code in type_bridge/query/__init__.py

def limit(self, limit: int) -> Query:
    """Set query limit.

    Args:
        limit: Maximum number of results

    Returns:
        Self for chaining
    """
    self.limit_val = limit
    return self

offset

offset(offset)

Set query offset.

Parameters:

Name	Type	Description	Default
offset	int	Number of results to skip	required

Returns:

Type	Description
Query	Self for chaining

Source code in type_bridge/query/__init__.py

def offset(self, offset: int) -> Query:
    """Set query offset.

    Args:
        offset: Number of results to skip

    Returns:
        Self for chaining
    """
    self.offset_val = offset
    return self

sort

sort(variable, direction='asc')

Add sorting to the query.

Parameters:

Name	Type	Description	Default
variable	str	Variable to sort by	required
direction	str	Sort direction ("asc" or "desc")	'asc'

Returns:

Type	Description
Query	Self for chaining

Example

Query().match("$p isa person").fetch("$p").sort("$p", "asc")

Source code in type_bridge/query/__init__.py

def sort(self, variable: str, direction: str = "asc") -> Query:
    """Add sorting to the query.

    Args:
        variable: Variable to sort by
        direction: Sort direction ("asc" or "desc")

    Returns:
        Self for chaining

    Example:
        Query().match("$p isa person").fetch("$p").sort("$p", "asc")
    """
    if direction not in ("asc", "desc"):
        raise ValueError(f"Invalid sort direction: {direction}")
    self.sort_clauses.append((variable, direction))
    return self

build

build()

Build the final TypeQL query string.

Returns:

Type	Description
str	Complete TypeQL query

Source code in type_bridge/query/__init__.py

def build(self) -> str:
    """Build the final TypeQL query string.

    Returns:
        Complete TypeQL query
    """
    logger.debug("Building TypeQL query")
    parts = []

    # Match clause
    if self.match_clause.patterns:
        parts.append(self.compiler.compile(self.match_clause))

    # Delete clause
    if self.delete_clause.statements:
        parts.append(self.compiler.compile(self.delete_clause))

    # Insert clause
    if self.insert_clause.statements:
        parts.append(self.compiler.compile(self.insert_clause))

    # Sort, offset, and limit modifiers (must come BEFORE fetch in TypeQL 3.x)
    modifier_parts = []
    if self.sort_clauses:
        sort_items = [f"{var} {dir}" for var, dir in self.sort_clauses]
        modifier_parts.append(f"sort {', '.join(sort_items)};")

    # Order matters: form modifiers, put offset then limit
    if self.offset_val is not None:
        modifier_parts.append(f"offset {self.offset_val};")
    if self.limit_val is not None:
        modifier_parts.append(f"limit {self.limit_val};")

    if modifier_parts:
        parts.append("\n".join(modifier_parts))

    # Fetch clause
    if self.fetch_clause.items:
        parts.append(self.compiler.compile(self.fetch_clause))

    query = "\n".join(parts)
    logger.debug(f"Built query: {query}")
    return query

__str__

__str__()

String representation of query.

Source code in type_bridge/query/__init__.py

def __str__(self) -> str:
    """String representation of query."""
    return self.build()

QueryBuilder

Helper class for building queries with model classes.

match_entity staticmethod

match_entity(model_class, var='$e', **filters)

Create a match query for an entity.

Parameters:

Name	Type	Description	Default
model_class	type[Entity]	The entity model class	required
var	str	Variable name to use	'$e'
filters	Any	Attribute filters (field_name: value)	{}

Returns:

Type	Description
Query	Query object

Source code in type_bridge/query/__init__.py

@staticmethod
def match_entity(model_class: type[Entity], var: str = "$e", **filters: Any) -> Query:
    """Create a match query for an entity.

    Args:
        model_class: The entity model class
        var: Variable name to use
        filters: Attribute filters (field_name: value)

    Returns:
        Query object
    """
    from type_bridge.crud.patterns import build_entity_match_pattern

    logger.debug(
        f"QueryBuilder.match_entity: {model_class.__name__}, var={var}, filters={filters}"
    )
    query = Query()
    pattern = build_entity_match_pattern(model_class, var, filters or None)
    query.match(pattern)
    return query

insert_entity staticmethod

insert_entity(instance, var='$e')

Create an insert query for an entity instance.

Parameters:

Name	Type	Description	Default
instance	Entity	Entity instance	required
var	str	Variable name to use	'$e'

Returns:

Type	Description
Query	Query object

Source code in type_bridge/query/__init__.py

@staticmethod
def insert_entity(instance: Entity, var: str = "$e") -> Query:
    """Create an insert query for an entity instance.

    Args:
        instance: Entity instance
        var: Variable name to use

    Returns:
        Query object
    """
    logger.debug(f"QueryBuilder.insert_entity: {instance.__class__.__name__}, var={var}")
    query = Query()
    insert_pattern = instance.to_insert_query(var)
    query.insert(insert_pattern)
    return query

match_relation staticmethod

match_relation(model_class, var='$r', role_players=None)

Create a match query for a relation.

Parameters:

Name	Type	Description	Default
model_class	type[Relation]	The relation model class	required
var	str	Variable name to use	'$r'
role_players	dict[str, str] | None	Dict mapping role names to player variables	None

Returns:

Type	Description
Query	Query object

Raises:

Type	Description
ValueError	If a role name is not defined in the model

Source code in type_bridge/query/__init__.py

@staticmethod
def match_relation(
    model_class: type[Relation], var: str = "$r", role_players: dict[str, str] | None = None
) -> Query:
    """Create a match query for a relation.

    Args:
        model_class: The relation model class
        var: Variable name to use
        role_players: Dict mapping role names to player variables

    Returns:
        Query object

    Raises:
        ValueError: If a role name is not defined in the model
    """
    from type_bridge.crud.patterns import build_relation_match_pattern

    logger.debug(
        f"QueryBuilder.match_relation: {model_class.__name__}, var={var}, "
        f"role_players={role_players}"
    )
    query = Query()
    pattern = build_relation_match_pattern(model_class, var, role_players)
    query.match(pattern)
    return query