pip install reflex
reflex init          # creates a new project in the current directory
reflex run           # starts dev server at http://localhost:3000

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="/")

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

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%",
    )

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),
    )

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")),
        ),
    )

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(),
    )

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)),
    )

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")

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

# 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

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(),
    )

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%",
    )

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()

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")

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,
    )

# 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

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,
)

# 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)
            ))

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

# 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!")