Skip to main content

Overview

QueryBuilder provides a fluent, chainable API for constructing complex SQL queries. It supports SELECT, UPDATE, and DELETE operations with filtering, ordering, pagination, and joins.

Class Definition

from framefox.core.orm import QueryBuilder, EntityManager
from typing import Type, Any

class QueryBuilder:
    def __init__(self, model: Type[Any], entity_manager: EntityManager):
        ...

Constructor

init

Initializes a QueryBuilder for a specific entity model.
def __init__(self, model: Type[Any], entity_manager: EntityManager)
model
Type[Any]
required
The entity class to build queries for
entity_manager
EntityManager
required
The entity manager to execute queries with

Example

from framefox.core.orm import EntityManager, QueryBuilder

em = EntityManager()
qb = QueryBuilder(model=User, entity_manager=em)

Query Type Methods

select

Initiates a SELECT query.
def select(self) -> "QueryBuilder"
return
QueryBuilder
The QueryBuilder instance for method chaining

Example

qb = QueryBuilder(model=User, entity_manager=em)
results = qb.select().execute()

delete

Initiates a DELETE query.
def delete(self) -> "QueryBuilder"
return
QueryBuilder
The QueryBuilder instance for method chaining

Example

qb = QueryBuilder(model=User, entity_manager=em)
qb.delete().where(User.is_active == False).execute()

update

Initiates an UPDATE query with values to set.
def update(self, values: Dict[str, Any]) -> "QueryBuilder"
values
Dict[str, Any]
required
Dictionary of field names and new values to update
return
QueryBuilder
The QueryBuilder instance for method chaining

Example

qb = QueryBuilder(model=User, entity_manager=em)
qb.update({"is_active": True}) \
    .where(User.last_login > datetime.now() - timedelta(days=30)) \
    .execute()

Filtering Methods

where

Adds a WHERE condition to the query.
def where(self, condition: Any) -> "QueryBuilder"
condition
Any
required
A SQLModel condition expression
return
QueryBuilder
The QueryBuilder instance for method chaining

Example

qb = QueryBuilder(model=User, entity_manager=em)

# Single condition
results = qb.select().where(User.age >= 18).execute()

# Multiple conditions (chained)
results = qb.select() \
    .where(User.age >= 18) \
    .where(User.is_active == True) \
    .execute()

# Complex conditions
results = qb.select() \
    .where((User.age >= 18) & (User.country == "US")) \
    .execute()

having

Adds a HAVING clause to the query (only for SELECT queries with aggregations).
def having(self, condition: Any) -> "QueryBuilder"
condition
Any
required
A SQLModel condition expression for aggregated results
return
QueryBuilder
The QueryBuilder instance for method chaining

Example

from sqlmodel import func

qb = QueryBuilder(model=Order, entity_manager=em)
results = qb.select() \
    .group_by(Order.user_id) \
    .having(func.count(Order.id) > 5) \
    .execute()

Join Methods

join

Adds JOIN clauses to the query.
def join(self, *joins: Any) -> "QueryBuilder"
joins
*Any
required
One or more join expressions
return
QueryBuilder
The QueryBuilder instance for method chaining

Example

qb = QueryBuilder(model=Post, entity_manager=em)

# Join with related table
results = qb.select() \
    .join(User) \
    .where(User.is_active == True) \
    .execute()

Ordering and Pagination

order_by

Adds ORDER BY clauses to the query.
def order_by(self, *conditions: Any) -> "QueryBuilder"
conditions
*Any
required
One or more ordering expressions
return
QueryBuilder
The QueryBuilder instance for method chaining

Example

from sqlmodel import asc, desc

qb = QueryBuilder(model=User, entity_manager=em)

# Single field ordering
results = qb.select().order_by(User.created_at.desc()).execute()

# Multiple fields
results = qb.select() \
    .order_by(User.last_name.asc(), User.first_name.asc()) \
    .execute()

limit

Limits the number of results returned.
def limit(self, limit: int) -> "QueryBuilder"
limit
int
required
Maximum number of results to return
return
QueryBuilder
The QueryBuilder instance for method chaining

Example

qb = QueryBuilder(model=User, entity_manager=em)
results = qb.select().limit(10).execute()

offset

Skips a specified number of results (for pagination).
def offset(self, offset: int) -> "QueryBuilder"
offset
int
required
Number of results to skip
return
QueryBuilder
The QueryBuilder instance for method chaining

Example

qb = QueryBuilder(model=User, entity_manager=em)

# Page 2 with 10 items per page
results = qb.select().limit(10).offset(10).execute()

# Page 3
results = qb.select().limit(10).offset(20).execute()

Query Execution

get_query

Builds and returns the final SQL query object without executing it.
def get_query(self) -> Any
return
Any
The constructed SQLModel query object (Select, Delete, or Update)

Example

qb = QueryBuilder(model=User, entity_manager=em)
query = qb.select().where(User.age >= 18).get_query()

# Inspect or modify the query
print(query)

execute

Executes the query and returns the results.
def execute(self) -> List[Any]
return
List[Any]
For SELECT queries: list of entity instances. For UPDATE/DELETE: execution result

Example

qb = QueryBuilder(model=User, entity_manager=em)

# Execute SELECT query
users = qb.select().where(User.is_active == True).execute()

# Execute DELETE query
qb.delete().where(User.last_login < datetime.now() - timedelta(days=365)).execute()

# Execute UPDATE query
qb.update({"status": "inactive"}).where(User.last_login < datetime.now() - timedelta(days=90)).execute()

first

Executes the query and returns only the first result.
def first(self) -> Optional[Any]
return
Optional[Any]
The first entity that matches the query, or None if no results

Example

qb = QueryBuilder(model=User, entity_manager=em)

# Get first matching user
user = qb.select().where(User.email == "[email protected]").first()

if user:
    print(f"Found: {user.username}")

last

Executes the query and returns the last result by reversing the order.
def last(self) -> Optional[Any]
return
Optional[Any]
The last entity that matches the query, or None if no results
Requires an order_by clause to be defined, as it reverses the sort order to get the last result.

Example

qb = QueryBuilder(model=User, entity_manager=em)

# Get most recently created user
latest_user = qb.select() \
    .order_by(User.created_at.asc()) \
    .last()

if latest_user:
    print(f"Latest user: {latest_user.username}")

params

Adds named parameters to the query.
def params(self, **kwargs: Any) -> "QueryBuilder"
kwargs
**Any
required
Named parameters to bind to the query
return
QueryBuilder
The QueryBuilder instance for method chaining

Example

qb = QueryBuilder(model=User, entity_manager=em)
results = qb.select() \
    .where(User.age >= :min_age) \
    .params(min_age=18) \
    .execute()

Complete Examples

Basic Queries

from framefox.core.orm import EntityManager, QueryBuilder, AbstractEntity
from sqlmodel import Field
from datetime import datetime, timedelta

class User(AbstractEntity, table=True):
    id: int | None = Field(default=None, primary_key=True)
    username: str
    email: str
    age: int
    is_active: bool = True
    last_login: datetime | None = None
    created_at: datetime = Field(default_factory=datetime.now)

em = EntityManager()
qb = QueryBuilder(model=User, entity_manager=em)

# Simple SELECT
all_users = qb.select().execute()

# SELECT with WHERE
active_users = qb.select() \
    .where(User.is_active == True) \
    .execute()

# SELECT with multiple conditions
adult_active_users = qb.select() \
    .where(User.age >= 18) \
    .where(User.is_active == True) \
    .execute()

# SELECT with ordering
recent_users = qb.select() \
    .order_by(User.created_at.desc()) \
    .limit(10) \
    .execute()

# SELECT with pagination
page_2_users = qb.select() \
    .order_by(User.username.asc()) \
    .limit(20) \
    .offset(20) \
    .execute()

Advanced Queries

# Complex filtering
premium_users = qb.select() \
    .where((User.age >= 18) & (User.age <= 65)) \
    .where(User.is_active == True) \
    .where(User.last_login > datetime.now() - timedelta(days=30)) \
    .order_by(User.last_login.desc()) \
    .execute()

# Get first result
admin_user = qb.select() \
    .where(User.email == "[email protected]") \
    .first()

# Get last result (requires order_by)
newest_user = qb.select() \
    .order_by(User.created_at.asc()) \
    .last()

# UPDATE query
qb.update({"is_active": False}) \
    .where(User.last_login < datetime.now() - timedelta(days=365)) \
    .execute()

# DELETE query
qb.delete() \
    .where(User.is_active == False) \
    .where(User.created_at < datetime.now() - timedelta(days=730)) \
    .execute()

Pagination Helper

def paginate_users(page: int, per_page: int = 20):
    """Paginate users with consistent ordering"""
    qb = QueryBuilder(model=User, entity_manager=em)
    
    offset = (page - 1) * per_page
    
    users = qb.select() \
        .order_by(User.id.asc()) \
        .limit(per_page) \
        .offset(offset) \
        .execute()
    
    return users

# Get page 1
page_1 = paginate_users(page=1, per_page=20)

# Get page 3
page_3 = paginate_users(page=3, per_page=20)

Search and Filter

def search_users(query: str, filters: dict):
    """Search users with dynamic filters"""
    qb = QueryBuilder(model=User, entity_manager=em)
    
    builder = qb.select()
    
    # Add search condition
    if query:
        builder = builder.where(
            (User.username.contains(query)) | 
            (User.email.contains(query))
        )
    
    # Add dynamic filters
    if filters.get("is_active") is not None:
        builder = builder.where(User.is_active == filters["is_active"])
    
    if filters.get("min_age"):
        builder = builder.where(User.age >= filters["min_age"])
    
    if filters.get("max_age"):
        builder = builder.where(User.age <= filters["max_age"])
    
    # Add sorting
    builder = builder.order_by(User.created_at.desc())
    
    return builder.execute()

# Usage
results = search_users(
    query="john",
    filters={"is_active": True, "min_age": 18}
)

Notes

  • QueryBuilder provides a fluent interface for method chaining
  • All methods (except execute(), first(), and last()) return the QueryBuilder instance
  • Use where() multiple times to add AND conditions
  • For OR conditions, use the | operator within a single where() call
  • first() and last() are only valid for SELECT queries
  • last() requires an order_by() clause
  • UPDATE and DELETE queries are committed automatically when executed
  • The query is not executed until execute(), first(), or last() is called

Build docs developers (and LLMs) love