reflex — Full-Stack Web Apps in Pure Python#
What it is#
Reflex is a Python framework for building full-stack web applications without writing JavaScript. You define UI components and application state in Python; Reflex compiles the frontend to React, runs a FastAPI backend, and synchronises state between them over WebSockets. Every interaction — button click, form submit, URL navigation — is handled by Python event handlers on the server. The result is a single-language codebase for apps that would otherwise require React + Python API.
Install#
pip install reflex
reflex init # creates a new project in the current directory
reflex run # starts dev server at http://localhost:3000
Output: (none — exits 0 on success)
Quick example#
import reflex as rx
class CounterState(rx.State):
count: int = 0
def increment(self):
self.count += 1
def decrement(self):
self.count -= 1
def counter_page() -> rx.Component:
return rx.center(
rx.vstack(
rx.heading(f"Count: {CounterState.count}", size="5"),
rx.hstack(
rx.button("−", on_click=CounterState.decrement),
rx.button("+", on_click=CounterState.increment),
),
),
)
app = rx.App()
app.add_page(counter_page, route="/")
When / why to use it#
- Internal tools and dashboards where shipping a React + Python split is too much overhead.
- Data apps (like Streamlit) but with full routing, forms, and auth — where Streamlit’s re-run model breaks down.
- Rapid prototyping of full-stack features when your team knows Python but not React/TypeScript.
- AI demos and chatbot UIs that need streaming and real-time state updates.
- Applications that would otherwise use Dash or Gradio but need more UI flexibility.
Common pitfalls#
[!WARNING] State mutations must happen inside event handlers — you cannot mutate
self.fieldoutside an event handler method. Direct assignment in__init__or other methods is silently ignored or raises an error.
[!WARNING]
rx.Statefields must be typed — untyped fields are not tracked by Reflex’s reactivity system. Always annotate:count: int = 0, notcount = 0.
[!WARNING]
reflex runruns both frontend and backend — the first run compiles the React frontend (~30s). Subsequent runs are faster. Do not kill the process during the first compile.
[!TIP] Use
rx.varfor computed/derived properties that depend on other state fields. They update automatically when their dependencies change, just like React’suseMemo.
[!TIP]
yieldinside an event handler streams intermediate state updates to the frontend. Use it to show progress during long-running operations.
State — the reactive core#
rx.State is the single source of truth. Fields declared on a state class are synchronised to the frontend automatically. Event handlers are methods that mutate fields.
import reflex as rx
from typing import Optional
class AppState(rx.State):
# Reactive fields — changes trigger frontend re-render
message: str = ""
items: list[str] = []
loading: bool = False
selected: Optional[str] = None
# Computed property — recalculated when items changes
@rx.var
def item_count(self) -> int:
return len(self.items)
@rx.var
def has_items(self) -> bool:
return len(self.items) > 0
# Event handlers — called by UI events
def add_item(self, item: str):
if item.strip():
self.items.append(item.strip())
self.message = f"Added: {item}"
def remove_item(self, item: str):
self.items = [i for i in self.items if i != item]
def clear_all(self):
self.items = []
self.message = "Cleared"
def select_item(self, item: str):
self.selected = item
Components — building UI#
Reflex wraps every HTML element and many higher-level components. All accept Python keyword arguments for props and event handlers.
import reflex as rx
def item_card(item: str) -> rx.Component:
return rx.box(
rx.hstack(
rx.text(item),
rx.button(
"×",
on_click=lambda: AppState.remove_item(item),
color_scheme="red",
size="1",
),
),
border="1px solid #ccc",
border_radius="8px",
padding="8px",
)
def item_list() -> rx.Component:
return rx.vstack(
rx.foreach(AppState.items, item_card),
width="100%",
)
rx.foreach — iterate over state lists#
rx.foreach renders a component for each item in a reactive list. Unlike a Python for loop, it re-renders only changed items when the list updates.
import reflex as rx
class ListState(rx.State):
fruits: list[str] = ["Apple", "Banana", "Cherry"]
def remove(self, fruit: str):
self.fruits = [f for f in self.fruits if f != fruit]
def fruit_item(fruit: str) -> rx.Component:
return rx.hstack(
rx.text(fruit),
rx.icon_button(
rx.icon("trash"),
on_click=ListState.remove(fruit),
variant="ghost",
size="1",
),
)
def fruit_list() -> rx.Component:
return rx.vstack(
rx.heading("Fruits"),
rx.foreach(ListState.fruits, fruit_item),
)
Conditional rendering — rx.cond#
rx.cond renders one of two components based on a reactive boolean expression. It is the Reflex equivalent of {condition ? A : B} in JSX.
import reflex as rx
class AuthState(rx.State):
logged_in: bool = False
username: str = ""
def login(self, username: str):
self.logged_in = True
self.username = username
def logout(self):
self.logged_in = False
self.username = ""
def nav_bar() -> rx.Component:
return rx.hstack(
rx.text("My App"),
rx.cond(
AuthState.logged_in,
rx.hstack(
rx.text(f"Hello, {AuthState.username}"),
rx.button("Log out", on_click=AuthState.logout),
),
rx.button("Log in", on_click=lambda: AuthState.login("Alice Dev")),
),
)
Forms and input binding#
import reflex as rx
class FormState(rx.State):
name: str = ""
email: str = ""
submitted: bool = False
result: dict = {}
def handle_submit(self, form_data: dict):
self.result = form_data
self.submitted = True
def contact_form() -> rx.Component:
return rx.form(
rx.vstack(
rx.input(placeholder="Your name", name="name"),
rx.input(placeholder="Your email", name="email", type="email"),
rx.text_area(placeholder="Message", name="message"),
rx.button("Submit", type="submit"),
),
on_submit=FormState.handle_submit,
reset_on_submit=True,
)
def contact_page() -> rx.Component:
return rx.cond(
FormState.submitted,
rx.callout(f"Received: {FormState.result}", icon="check"),
contact_form(),
)
Async event handlers and streaming#
Async event handlers can yield to stream intermediate state updates — ideal for showing progress or streaming LLM output.
import reflex as rx
import asyncio
class StreamState(rx.State):
words: list[str] = []
generating: bool = False
async def generate_words(self):
self.generating = True
self.words = []
yield # stream initial state to frontend
sentences = ["Hello", "World", "from", "Reflex", "streaming"]
for word in sentences:
await asyncio.sleep(0.4)
self.words.append(word)
yield # stream each word as it arrives
self.generating = False
yield
def stream_page() -> rx.Component:
return rx.vstack(
rx.button(
"Generate",
on_click=StreamState.generate_words,
loading=StreamState.generating,
),
rx.hstack(rx.foreach(StreamState.words, rx.text)),
)
Multiple pages and routing#
import reflex as rx
def home() -> rx.Component:
return rx.vstack(
rx.heading("Home"),
rx.link("Go to About", href="/about"),
)
def about() -> rx.Component:
return rx.vstack(
rx.heading("About"),
rx.link("Go home", href="/"),
)
app = rx.App()
app.add_page(home, route="/")
app.add_page(about, route="/about")
Database integration#
Reflex includes SQLModel integration via rx.Model.
import reflex as rx
class Todo(rx.Model, table=True):
id: int | None = None
text: str
done: bool = False
class TodoState(rx.State):
todos: list[Todo] = []
new_text: str = ""
def load_todos(self):
with rx.session() as session:
self.todos = session.exec(Todo.select()).all()
def add_todo(self):
with rx.session() as session:
todo = Todo(text=self.new_text)
session.add(todo)
session.commit()
self.new_text = ""
self.load_todos()
def toggle_done(self, todo_id: int):
with rx.session() as session:
todo = session.get(Todo, todo_id)
todo.done = not todo.done
session.commit()
self.load_todos()
def set_new_text(self, value: str):
self.new_text = value
Deployment#
# Export for self-hosting
reflex export --frontend-only # static files in frontend/
# Or run with production settings
reflex run --env prod
# Deploy to Reflex Cloud (one command)
reflex deploy
Output: (none — exits 0 on success)
Real-world recipes#
End-to-end snippets that show how state, components, async handlers, and routing compose in a real Reflex app.
1. Form with server-side validation#
import reflex as rx
class SignupState(rx.State):
name: str = ""
email: str = ""
errors: dict[str, str] = {}
success: bool = False
def submit(self, form: dict):
self.errors = {}
if not form.get("name", "").strip():
self.errors["name"] = "Name is required"
if "@" not in form.get("email", ""):
self.errors["email"] = "Invalid email"
if self.errors:
return
self.name = form["name"].strip()
self.email = form["email"].strip()
self.success = True
def signup_form() -> rx.Component:
return rx.form(
rx.vstack(
rx.input(placeholder="Name", name="name"),
rx.cond(SignupState.errors["name"],
rx.text(SignupState.errors["name"], color="red", size="1")),
rx.input(placeholder="Email", name="email", type="email"),
rx.cond(SignupState.errors["email"],
rx.text(SignupState.errors["email"], color="red", size="1")),
rx.button("Sign up", type="submit"),
),
on_submit=SignupState.submit,
reset_on_submit=False,
)
def signup_page() -> rx.Component:
return rx.cond(
SignupState.success,
rx.callout(f"Welcome, {SignupState.name}!", icon="check"),
signup_form(),
)
2. Live chat with streaming response#
import reflex as rx
import asyncio
class ChatState(rx.State):
messages: list[dict] = []
draft: str = ""
streaming: bool = False
async def send(self):
if not self.draft.strip():
return
self.messages.append({"role": "user", "text": self.draft})
prompt = self.draft
self.draft = ""
self.streaming = True
self.messages.append({"role": "bot", "text": ""})
yield
# Simulate streaming tokens
for chunk in ("Hello ", "there! ", "You said: ", f"'{prompt}'"):
await asyncio.sleep(0.15)
self.messages[-1]["text"] += chunk
yield
self.streaming = False
yield
def msg_bubble(m) -> rx.Component:
return rx.box(
rx.text(m["text"]),
background_color=rx.cond(m["role"] == "user", "#e0e7ff", "#f3f4f6"),
padding="8px 12px",
border_radius="8px",
margin_y="4px",
)
def chat_page() -> rx.Component:
return rx.vstack(
rx.foreach(ChatState.messages, msg_bubble),
rx.hstack(
rx.input(
value=ChatState.draft,
on_change=ChatState.set_draft,
placeholder="Type a message...",
),
rx.button("Send",
on_click=ChatState.send,
loading=ChatState.streaming),
),
width="100%",
)
3. CRUD dashboard with SQLModel#
import reflex as rx
class Note(rx.Model, table=True):
id: int | None = None
title: str
body: str = ""
class NotesState(rx.State):
notes: list[Note] = []
title: str = ""
body: str = ""
editing: int | None = None
def load(self):
with rx.session() as s:
self.notes = list(s.exec(Note.select()))
def save(self):
with rx.session() as s:
if self.editing:
n = s.get(Note, self.editing)
n.title = self.title
n.body = self.body
else:
s.add(Note(title=self.title, body=self.body))
s.commit()
self.title = self.body = ""
self.editing = None
self.load()
def edit(self, note_id: int):
with rx.session() as s:
n = s.get(Note, note_id)
self.title = n.title
self.body = n.body
self.editing = note_id
def delete(self, note_id: int):
with rx.session() as s:
n = s.get(Note, note_id)
s.delete(n)
s.commit()
self.load()
4. Multi-page app with shared state and layouts#
import reflex as rx
class GlobalState(rx.State):
theme: str = "light"
user: str = "guest"
def toggle_theme(self):
self.theme = "dark" if self.theme == "light" else "light"
def layout(content: rx.Component) -> rx.Component:
return rx.vstack(
rx.hstack(
rx.heading("MyApp"),
rx.spacer(),
rx.text(f"Theme: {GlobalState.theme}"),
rx.button("Toggle", on_click=GlobalState.toggle_theme),
width="100%",
),
content,
align="stretch",
padding="2em",
)
def home() -> rx.Component:
return layout(rx.vstack(rx.heading("Home"), rx.link("Dashboard", href="/dashboard")))
def dashboard() -> rx.Component:
return layout(rx.vstack(rx.heading("Dashboard"), rx.link("Home", href="/")))
app = rx.App()
app.add_page(home, route="/")
app.add_page(dashboard, route="/dashboard")
5. Wrapping a custom React component#
import reflex as rx
class Chart(rx.Component):
"""Wrap a react chart library installed via npm."""
library = "recharts@2.x"
tag = "BarChart"
data: rx.Var[list[dict]]
width: rx.Var[int] = 400
height: rx.Var[int] = 300
# Use it in a page
def stats_page() -> rx.Component:
return Chart(
data=[{"name": "Jan", "value": 10}, {"name": "Feb", "value": 20}],
width=600,
height=400,
)
Production deployment#
Reflex apps have two halves: the Next.js frontend (compiled bundle) and the FastAPI backend (Python websocket server). Deploy them together with reflex run --env prod or split them across CDN + backend.
# Build the frontend bundle and Python wheel
reflex export
# Or build only the frontend (for static hosting + remote backend)
reflex export --frontend-only --no-zip
# Run production stack on one box
REFLEX_DB_URL=postgresql://user:pw@db:5432/app reflex run --env prod \
--backend-host 0.0.0.0 --backend-port 8000
# Or deploy to Reflex Cloud
reflex deploy --project my-app
Output: (none — exits 0 on success)
rxconfig.py — production config:
import reflex as rx
config = rx.Config(
app_name="my_app",
db_url="postgresql://user:pw@db:5432/app",
api_url="https://api.example.com",
deploy_url="https://app.example.com",
cors_allowed_origins=["https://app.example.com"],
telemetry_enabled=False,
timeout=120,
)
Topology options:
| Topology | Frontend | Backend | Best for |
|---|---|---|---|
| Reflex Cloud | Managed | Managed | Fastest path to production; no infra |
| All-in-one container | Bundled with backend | FastAPI | Single VM, single Docker container |
| Split: CDN + backend | S3/CloudFront/Vercel | EC2/Fly.io/Render | Caching the static bundle; multi-region |
| Self-hosted K8s | Service + Ingress | StatefulSet | Enterprise / on-prem |
Production checklist:
- Pin the Node.js version that ships with Reflex (
reflex --versionshows it). The frontend build is sensitive to Node minor versions. - Use Postgres in production, not the default SQLite. Each backend instance needs the same DB.
- Set
cors_allowed_originsexplicitly; leaving it open is a security footgun. - Bundle size grows with components — Tailwind purging on the production build cuts ~40% off the JS bundle.
- WebSockets need sticky sessions behind a reverse proxy. Configure your load balancer to hash on the user’s connection ID.
Performance tuning#
Reflex’s performance hinges on three things: the frontend bundle size, server-side state-update frequency, and websocket round-trips.
# 1. Memoise expensive computed vars — they re-run on every state change
class State(rx.State):
items: list[dict] = []
@rx.var(cache=True) # only re-run when items changes
def sorted_items(self) -> list[dict]:
return sorted(self.items, key=lambda x: x["score"], reverse=True)
# 2. Batch state updates — one yield per logical step, not per field
class BadState(rx.State):
a: int = 0
b: int = 0
async def update_both(self):
self.a = 1
yield # round-trip 1
self.b = 2
yield # round-trip 2
class GoodState(rx.State):
a: int = 0
b: int = 0
async def update_both(self):
self.a = 1
self.b = 2
yield # single round-trip
# 3. Avoid huge lists in state — page server-side instead
class PagedState(rx.State):
page: int = 0
page_size: int = 50
@rx.var(cache=True)
def visible_items(self) -> list[dict]:
with rx.session() as s:
return list(s.exec(
Item.select().offset(self.page * self.page_size).limit(self.page_size)
))
Output: (none — exits 0 on success)
Tuning checklist:
- Memoise computed vars with
@rx.var(cache=True)for any expensive derivation. - Minimise the number of
yields in async handlers — each is a websocket message. - Don’t put 10k-item lists in state. Paginate server-side; only ship the visible window.
- Use
rx.foreachover Python comprehensions when rendering lists —foreachdiffs efficiently. - Lazy-load routes with
app.add_page(..., on_load=State.fetch)rather than fetching at module import. - Disable telemetry in production (
telemetry_enabled=False) — saves a network call on every page load.
Testing patterns#
Reflex’s testing story is still maturing. The most reliable approach is to unit-test state classes as plain Python objects and end-to-end-test the rendered app with Playwright.
import pytest
from my_app.state import SignupState
# 1. Unit-test state methods — instantiate without the framework
def test_submit_invalid_email_sets_error():
s = SignupState()
s.submit({"name": "Alice", "email": "bad"})
assert "email" in s.errors
assert not s.success
def test_submit_valid_succeeds():
s = SignupState()
s.submit({"name": "Alice", "email": "alice@example.com"})
assert s.success
assert s.name == "Alice"
# 2. Computed vars — call the underlying method
def test_item_count():
s = SignupState()
s.name = "Alice"
# Computed vars decorated with @rx.var still work as descriptors
Output: (none — exits 0 on success)
# 3. End-to-end with Playwright
# tests/test_e2e.py
import pytest
from playwright.sync_api import Page
def test_signup_flow(page: Page, reflex_app_url: str):
page.goto(reflex_app_url)
page.fill('input[name="name"]', "Alice Dev")
page.fill('input[name="email"]', "alice@example.com")
page.click('button[type="submit"]')
page.wait_for_selector("text=Welcome, Alice Dev!")
[!WARNING] Reflex’s built-in pytest helpers are limited. As of the current stable release, there is no first-class equivalent to FastAPI’s
TestClientfor state handlers. Test business logic on plain state objects, and exercise the UI through Playwright or a manual harness.
Migration from older Reflex versions#
Reflex (originally Pynecone) went 1.0 in mid-2024 and renamed in the process. Older code uses the pc import alias and several renamed APIs.
| Concept | Pynecone / pre-1.0 | Reflex 1.x |
|---|---|---|
| Module | import pynecone as pc | import reflex as rx |
| Config | pcconfig.py | rxconfig.py |
| State base | pc.State | rx.State |
| Component | pc.Component | rx.Component |
| Run command | pc run | reflex run |
| Init command | pc init | reflex init |
| Conditional | pc.cond | rx.cond |
| Loop | pc.foreach | rx.foreach |
| Computed var | @pc.var | @rx.var |
| DB session | pc.session() | rx.session() |
Migration steps:
- Rename
pcconfig.py→rxconfig.pyand updateapp_name. - Run a project-wide search-replace for
pc.→rx.andimport pynecone as pc→import reflex as rx. - Refresh the Node bundle: delete
.web/andassets/external/directories, thenreflex runto regenerate. - Re-pin extra component packages — Pynecone’s
pc.NextLinketc. are nowrx.next.link. - State subclass deprecations. Pre-1.0 allowed untyped fields; 1.x requires type annotations.
Ecosystem integrations#
Reflex’s component library wraps the most-used React libraries. Custom integrations are straightforward via rx.Component subclassing.
| Library | Wrapped as |
|---|---|
| Radix UI | Built-in — most rx.* primitives are Radix wrappers |
| Tailwind CSS | Built-in; pass class_name="..." to any component |
| Recharts | rx.recharts.line_chart(...), rx.recharts.bar_chart(...) |
| SQLModel | rx.Model(..., table=True) + rx.session() |
| FastAPI | Reflex backend IS FastAPI; mount endpoints on app.api |
| Chakra UI | Available as a community plugin |
| Any npm React component | class MyComp(rx.Component): library = "pkg@1.x"; tag = "Comp" |
Patterns & idioms#
- State splitting. Subclass
rx.Stateper page or feature:UserState,OrdersState. Avoid one mega-state. @rx.var(cache=True)for derived data that’s expensive to compute.rx.foreachover list comprehensions when rendering lists in JSX —foreachdiffs efficiently.yieldfor streaming state updates inside async handlers.on_load=State.fetchto load data when a route is entered, not at app boot.- Layouts as functions, not classes.
def layout(content): return rx.vstack(navbar(), content). rx.fragmentfor grouping without a wrapper DOM element when CSS layout requires direct parenting.
Troubleshooting common errors#
| Error | Cause | Fix |
|---|---|---|
State field 'count' must be typed | Missing annotation | Use count: int = 0, not count = 0 |
Cannot mutate state outside of an event handler | Direct assignment in __init__ or render | Move to a method decorated as event handler |
Component prop X does not accept type Y | Mismatched prop type | Cast in Python or use rx.Var.create(...) |
First reflex run takes 60+ seconds | Initial Node bundle compile | Expected; subsequent runs are cached in .web/ |
EADDRINUSE :3000 | Old dev server still running | pkill -f reflex or change port in rxconfig.py |
| WebSocket disconnects in production | Reverse proxy not configured for upgrade | Set Upgrade and Connection headers in nginx/Caddy |
rx.foreach only renders first item | Argument is not an rx.Var list | Make sure the list is a state attribute, not a Python local |
| Computed var doesn’t update | Missing cache=True or dependency not tracked | Read dependencies explicitly inside the var body |
| Static assets 404 in prod | assets/ not copied into image | Include assets/ in your Dockerfile |
When NOT to use this#
Reflex is excellent for full-stack Python apps but is not the right choice for every UI.
- Heavy client-side interactivity. Drag-drop, canvas-heavy editors, or 60fps animations need a real React app. Reflex’s server-round-trip model adds latency.
- Public SEO-critical marketing sites. Reflex SSRs the first page only; for deep static content use Astro or Next.js.
- Mobile-first PWAs with offline support. Reflex requires a live websocket; offline-first apps need a service worker and local state.
- Embedding into an existing React app. Reflex is a complete framework, not a component library; integration is non-trivial.
- When you don’t know Python well. Reflex’s abstraction hides a lot — debugging needs comfort with both React internals and the Reflex runtime.
- Lightweight dashboards or notebooks. Streamlit is simpler if you just need to display data; Reflex’s reactivity model is overkill.
Quick reference#
| Task | Code |
|---|---|
| Init project | reflex init |
| Dev server | reflex run |
| State field | class S(rx.State): count: int = 0 |
| Event handler | def increment(self): self.count += 1 |
| Computed var | @rx.var def doubled(self) -> int: return self.count * 2 |
| Bind to event | rx.button("Click", on_click=State.handler) |
| Foreach | rx.foreach(State.items, component_fn) |
| Conditional | rx.cond(State.flag, true_comp, false_comp) |
| Input bind | rx.input(on_change=State.set_field) |
| Form submit | rx.form(..., on_submit=State.handle_submit) |
| Stream updates | async def handler(self): yield between mutations |
| Add page | app.add_page(fn, route="/path") |
| Navigate | rx.link("text", href="/page") or rx.redirect("/page") |
| DB session | with rx.session() as s: s.exec(...) |
| Deploy | reflex deploy |