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)
The entity class to build queries for
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"
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"
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"
Dictionary of field names and new values to update
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"
A SQLModel condition expression
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"
A SQLModel condition expression for aggregated results
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"
One or more join expressions
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"
One or more ordering expressions
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"
Maximum number of results to return
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"
Number of results to skip
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
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]
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]
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]
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"
Named parameters to bind to the query
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()
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