queries.md
For AI agents: the complete documentation index is at llms.txt. Markdown versions are available by appending .md or sending Accept: text/markdown.
Queries
Queries are used to retrieve data from a database.
A query is a request for information from a database table or combination of tables. A query can be used to retrieve data from a single table or multiple tables. A query can also be used to insert, update, or delete data from a table.
Session
To execute a query you must first create a rx.session. You can use the session to query the database using SQLModel or SQLAlchemy syntax.
The rx.session statement will automatically close the session when the code block is finished. If session.commit() is not called, the changes will be rolled back and not persisted to the database. The code can also explicitly rollback without closing the session via session.rollback().
The following example shows how to create a session and query the database. First we create a table called User.
class User(rx.Model, table=True):
username: str
email: str
Select
Then we create a session and query the User table.
class QueryUser(rx.State):
name: str
users: list[User]
@rx.event
def get_users(self):
with rx.session() as session:
self.users = session.exec(
select(User).where(User.username.contains(self.name))
).all()
The get_users method will query the database for all users that contain the value of the state var name.
Insert
Similarly, the session.add() method to add a new record to the database or persist an existing object.
class AddUser(rx.State):
username: str
email: str
@rx.event
def add_user(self):
with rx.session() as session:
session.add(User(username=self.username, email=self.email))
session.commit()
Update
To update the user, first query the database for the object, make the desired modifications, .add the object to the session and finally call .commit().
class ChangeEmail(rx.State):
username: str
email: str
@rx.event
def modify_user(self):
with rx.session() as session:
user = session.exec(
select(User).where((User.username == self.username))
).first()
user.email = self.email
session.add(user)
session.commit()
Delete
To delete a user, first query the database for the object, then call .delete() on the session and finally call .commit().
class RemoveUser(rx.State):
username: str
@rx.event
def delete_user(self):
with rx.session() as session:
user = session.exec(
select(User).where(User.username == self.username)
).first()
session.delete(user)
session.commit()
ORM Object Lifecycle
The objects returned by queries are bound to the session that created them, and cannot generally be used outside that session. After adding or updating an object, not all fields are automatically updated, so accessing certain attributes may trigger additional queries to refresh the object.
To avoid this, the session.refresh() method can be used to update the object explicitly and ensure all fields are up to date before exiting the session.
class AddUserForm(rx.State):
user: User | None = None
@rx.event
def add_user(self, form_data: dict[str, Any]):
with rx.session() as session:
self.user = User(**form_data)
session.add(self.user)
session.commit()
session.refresh(self.user)
Now the self.user object will have a correct reference to the autogenerated primary key, id, even though this was not provided when the object was created from the form data.
If self.user needs to be modified or used in another query in a new session, it must be added to the session. Adding an object to a session does not necessarily create the object, but rather associates it with a session where it may either be created or updated accordingly.
class AddUserForm(rx.State):
...
@rx.event
def update_user(self, form_data: dict[str, Any]):
if self.user is None:
return
with rx.session() as session:
self.user.set(**form_data)
session.add(self.user)
session.commit()
session.refresh(self.user)
If an ORM object will be referenced and accessed outside of a session, you should call .refresh() on it to avoid stale object exceptions.
Using SQL Directly
Avoiding SQL is one of the main benefits of using an ORM, but sometimes it is necessary for particularly complex queries, or when using database-specific features.
SQLModel exposes the session.execute() method that can be used to execute raw SQL strings. If parameter binding is needed, the query may be wrapped in sqlalchemy.text, which allows colon-prefix names to be used as placeholders.
# Never use string formatting to construct SQL queries, as this may lead to SQL injection vulnerabilities in the app.
import sqlalchemy
import reflex as rx
class State(rx.State):
@rx.event
def insert_user_raw(self, username, email):
with rx.session() as session:
session.execute(
sqlalchemy.text(
"INSERT INTO user (username, email) VALUES (:username, :email)"
),
{"username": username, "email": email},
)
session.commit()
@rx.var
def raw_user_tuples(self) -> list[list]:
with rx.session() as session:
return [list(row) for row in session.execute("SELECT * FROM user").all()]
Writing Efficient Queries
Databases are generally much better at filtering, joining, and aggregating than Python. Every row returned by a query is sent over the network and held in memory as part of the state — for every user with an open session. Do the work in the query itself and return only display-ready results.
The examples below use a simple Order model.
class Order(rx.Model, table=True):
user_id: int
status: str
amount: float | None = None
Aggregate in SQL, not Python
Avoid fetching a whole table just to compute a summary in Python. Use the database's aggregate functions (COUNT, SUM, AVG, MIN, MAX) with GROUP BY so only the summary rows come back.
from sqlmodel import select, func
class OrderStats(rx.State):
order_count: int = 0
orders_by_status: list[tuple[str, int]] = []
@rx.event
def load_stats(self):
with rx.session() as session:
# Count rows in the database instead of len() on a full fetch.
self.order_count = session.exec(select(func.count(Order.id))).one()
# One row per group, ready for a chart or summary table.
self.orders_by_status = [
tuple(row)
for row in session.exec(
select(Order.status, func.count(Order.id)).group_by(Order.status)
).all()
]
Related metrics can be computed in a single query with conditional aggregation, instead of issuing one query per metric:
import sqlalchemy
from sqlmodel import select, func
class OrderKPIs(rx.State):
total_orders: int = 0
completed_orders: int = 0
total_amount: float = 0.0
@rx.event
def load_kpis(self):
with rx.session() as session:
total, completed, amount = session.exec(
select(
func.count(Order.id),
func.sum(
sqlalchemy.case((Order.status == "completed", 1), else_=0)
),
func.coalesce(func.sum(Order.amount), 0.0),
)
).one()
self.total_orders = total
self.completed_orders = completed or 0
self.total_amount = float(amount)
Avoid Queries Inside Loops
Issuing a query per item — the "N+1" pattern — multiplies network round trips. Fetch everything in one statement with a JOIN, an IN (...) list, or GROUP BY.
# Efficient: one query for all users at once.
with rx.session() as session:
orders = session.exec(
select(Order).where(Order.user_id.in_([user.id for user in users]))
).all()
# Inefficient: one query per user (N+1).
with rx.session() as session:
for user in users:
orders = session.exec(select(Order).where(Order.user_id == user.id)).all()
The .in_() method binds each value as a separate query parameter, so it is safe to use with user-provided values.
# Raw SQL: pass lists with an expanding bind parameter
In raw SQL, use an *expanding* bind parameter for an `IN` list — never join the values into the SQL string, which exposes the query to SQL injection.
The example below binds the list with an expanding parameter:
import sqlalchemy
with rx.session() as session:
rows = session.execute(
sqlalchemy.text("SELECT * FROM users WHERE id IN :user_ids").bindparams(
sqlalchemy.bindparam("user_ids", expanding=True)
),
{"user_ids": [1, 2, 3]},
).all()
If the related objects are linked with foreign keys, the relationship loading techniques can also fetch linked objects without extra queries.
Return Only What the UI Needs
- Select only the columns the UI displays when tables are wide, instead of whole rows.
- Give every query that returns detail rows a
.limit(), and use offset-based pagination when the user needs more rows. - Load static data (dropdown options, date bounds) once in an
on_loadevent handler. When a filter changes, re-run only the filtered queries — not the option lookups. - Avoid caching large raw query results in state vars to work around a slow query: per-user state duplicates that data in server memory for every session. Fix the query instead — aggregate in SQL, add a limit, combine round trips. Keeping small, display-ready results in state (chart rows, KPI values, one page of a table) is the intended pattern.
Handling NULL Values
Database columns may contain NULL, which arrives in Python as None. Aggregates like SUM and AVG also return NULL when they run over zero rows. Casting such results directly will crash:
float(row[0]) # TypeError: float() argument must be ... not 'NoneType'
There are two ways to handle this.
Coalesce in the Query (Preferred)
COALESCE returns its first non-NULL argument, so the query itself guarantees a usable value. This is especially important for nullable numeric columns and for aggregates that may run over empty groups.
from sqlmodel import select, func
with rx.session() as session:
total = session.exec(select(func.coalesce(func.sum(Order.amount), 0.0))).one()
The same works in raw SQL: SELECT COALESCE(SUM(amount), 0) FROM ....
Guard in Python
When consuming rows that may contain NULLs, fall back explicitly while building the values:
orders = [
{
"status": str(row[0] or ""),
"amount": float(row[1]) if row[1] is not None else 0.0,
}
for row in rows
]
Async Database Operations
Reflex provides an async version of the session function called rx.asession for asynchronous database operations. This is useful when you need to perform database operations in an async context, such as within async event handlers.
Configuring the Async Database URL
rx.asession needs its own async_db_url in rxconfig.py. It must point at the same database as db_url but use an async driver. Calling rx.asession() when it is unset raises an error.
config = rx.Config(
app_name="my_app",
db_url="sqlite:///reflex.db",
async_db_url="sqlite+aiosqlite:///reflex.db",
)
The matching async DBAPI driver is not part of the reflex[db] extra, so install it separately. For the SQLite URL above, install aiosqlite:
pip install aiosqlite
For PostgreSQL, install psycopg (psycopg3). It works as both the sync and async driver, so db_url and async_db_url can share one
postgresql+psycopg://... scheme:
pip install "psycopg[binary]"
The rx.asession function returns an async SQLAlchemy session that must be used with an async context manager. Most operations against the asession must be awaited.
import sqlalchemy.ext.asyncio
import sqlalchemy
import reflex as rx
class AsyncUserState(rx.State):
users: list[User] = []
@rx.event(background=True)
async def get_users_async(self):
async with rx.asession() as asession:
result = await asession.execute(select(User))
async with self:
self.users = result.all()
Async Select
The following example shows how to query the database asynchronously:
class AsyncQueryUser(rx.State):
name: str
users: list[User] = []
@rx.event(background=True)
async def get_users(self):
async with rx.asession() as asession:
stmt = select(User).where(User.username.contains(self.name))
result = await asession.execute(stmt)
async with self:
self.users = result.all()
Async Insert
To add a new record to the database asynchronously:
class AsyncAddUser(rx.State):
username: str
email: str
@rx.event(background=True)
async def add_user(self):
async with rx.asession() as asession:
asession.add(User(username=self.username, email=self.email))
await asession.commit()
Async Update
To update a user asynchronously:
class AsyncChangeEmail(rx.State):
username: str
email: str
@rx.event(background=True)
async def modify_user(self):
async with rx.asession() as asession:
stmt = select(User).where(User.username == self.username)
result = await asession.execute(stmt)
user = result.first()
if user:
user.email = self.email
asession.add(user)
await asession.commit()
Async Delete
To delete a user asynchronously:
class AsyncRemoveUser(rx.State):
username: str
@rx.event(background=True)
async def delete_user(self):
async with rx.asession() as asession:
stmt = select(User).where(User.username == self.username)
result = await asession.execute(stmt)
user = result.first()
if user:
await asession.delete(user)
await asession.commit()
Async Refresh
Similar to the regular session, you can refresh an object to ensure all fields are up to date:
class AsyncAddUserForm(rx.State):
user: User | None = None
@rx.event(background=True)
async def add_user(self, form_data: dict[str, str]):
async with rx.asession() as asession:
async with self:
self.user = User(**form_data)
asession.add(self.user)
await asession.commit()
await asession.refresh(self.user)
Async SQL Execution
You can also execute raw SQL asynchronously:
class AsyncRawSQL(rx.State):
users: list[list] = []
@rx.event(background=True)
async def insert_user_raw(self, username, email):
async with rx.asession() as asession:
await asession.execute(
sqlalchemy.text(
"INSERT INTO user (username, email) VALUES (:username, :email)"
),
dict(username=username, email=email),
)
await asession.commit()
@rx.event(background=True)
async def get_raw_users(self):
async with rx.asession() as asession:
result = await asession.execute("SELECT * FROM user")
async with self:
self.users = [list(row) for row in result.all()]
# Important Notes for Async Database Operations
- Always use the `@rx.event(background=True)` decorator for async event handlers
- Most operations against the `asession` must be awaited, including `commit()`, `execute()`, `refresh()`, and `delete()`
- The `add()` method does not need to be awaited
- Result objects from queries have methods like `all()` and `first()` that are synchronous and return data directly
- Use `async with self:` when updating state variables in background tasks