Core

The FastHTML subclass of Starlette.

Create an app with app = FastHTML(). The @rt decorator adds routes (rt = app.route), inferring what it can from the function:

@rt
def index(): ...      # GET,POST /
@rt
def foo(): ...        # GET,POST /foo
@rt('/hi')
def get(): ...        # GET /hi (function named after a verb handles just that verb)
@rt('/cards/{id}')
class Cards: ...      # class route group: each verb-named method (get, patch, ...) becomes a handler

The decorated function doubles as a URL builder, so links never need hardcoded paths:

A('More', href=foo.to(a=1))   # <a href="/foo?a=1"> `.to()` fills path params; the rest become query params.
Div('...', hx_get=foo)        # route functions work directly as htmx attr values: hx-get="/foo"

Handlers return FT trees (from fastcore.xml), rendered as HTML automatically: a full page for regular requests, a bare fragment for HTMX requests. In a full page, top-level Title, Meta, Link, and Style items go in <head>; the rest goes in <body>. Strings are sent as HTML, dicts as JSON, and a Starlette Response is sent untouched. A return tuple may mix content with special items: any HttpHeader (e.g. from cookie() or HtmxResponseHeaders()) becomes a response header, any BackgroundTask runs after the response is sent, and the rest renders as content. Redirect picks the right redirect mechanism for HTMX vs regular requests, and FtResponse wraps FT content when you need the status code or headers.

Handler parameters are filled from the request: each is looked up in path, cookie, header (snake_case names match Hyphen-Case headers), query, then form data, and cast by calling its annotation on the value. Hyphens in form/query names match underscores in handler parameters and typed body fields. Exact names take precedence within each source. Dictionary and request access retain the original keys. bool, int, and date get smart string parsing; UploadFile passes through. Repeated params take the last value unless annotated list[T], which collects them all. A param with no annotation and no special name is ignored, with a warning. A dataclass, TypedDict, namedtuple, or any annotated class collects the whole form body; a __from_request__ classmethod customizes construction. A missing required param is a 400, a failed cast a 404. These special names need no annotation:

req, ws   the Request / WebSocket connection
sess      session dict
app       the FastHTML app
state     app.state
scope     ASGI scope
auth      scope['auth']
htmx      HtmxHeaders (parsed HX-* request headers)
body      raw body text
data      parsed form data as a dict
api       ApiReturn (API vs browser dual responses)
send      websocket send function
hdrs,ftrs the per-request copies of the app's `hdrs`/`ftrs` (head and footer content)
htmlkw    the per-request copy of the app's `<html>` attrs
bodykw    the per-request copy of the app's `<body>` attrs
resp      the handler's response (injected into `after` functions)

before/after functions (app-level, with Beforeware skip patterns, or per-route before=) get the same param injection as handlers. A before returning a value responds with it and skips the handler (the usual auth-guard pattern).

See the Handling Handlers tutorial for worked examples of the above.

This is the source code to fasthtml. You won’t need to read this unless you want to understand how things are built behind the scenes, or need full details of a particular API. The notebook is converted to the Python module fasthtml/core.py using nbdev.

Imports and utils

import time, tempfile

from IPython import display
from enum import Enum
from pprint import pprint
from collections import namedtuple
from typing import Optional, TypedDict

from starlette.testclient import TestClient
from fastcore.test import *
from fastcore.net import run_sync
from starlette.datastructures import UploadFile

We write source code first, and then tests come after. The tests serve as both a means to confirm that the code works and also serves as working examples. The first exported function, parsed_date, is an example of this pattern.

---

parsed_date

def parsed_date(
    s:str
):

Convert s to a datetime

parsed_date('2pm')

datetime.datetime(2026, 3, 30, 14, 0)

isinstance(date.fromtimestamp(0), date)

True

---

snake2hyphens

def snake2hyphens(
    s:str
):

Convert s from snake case to hyphenated and capitalised

snake2hyphens("snake_case")

'Snake-Case'

---

HtmxHeaders

def HtmxHeaders(
    boosted:str | None=None, current_url:str | None=None, history_restore_request:str | None=None,
    prompt:str | None=None, request:str | None=None, request_type:str | None=None, source:str | None=None,
    target:str | None=None, trigger_name:str | None=None, trigger:str | None=None
)->None:

Parsed HX- request headers*

def test_request(url: str='/', headers: dict=None, method: str='get')->Request:
    if headers is None: headers={}
    scope = dict(type='http', method=method, path=url, headers=Headers(headers).raw, query_string=b'', scheme='http',
        client=('127.0.0.1', 8000), server=('127.0.0.1', 8000))
    async def receive(): return dict(type="http.request", body=b"", more_body=False)
    return Request(scope, receive)

h = test_request(headers=Headers({'HX-Request':'1'}))
_get_htmx(h.headers)

HtmxHeaders(boosted=None, current_url=None, history_restore_request=None, prompt=None, request='1', request_type=None, source=None, target=None, trigger_name=None, trigger=None)

Request and response

test_eq(_fix_anno(Union[str,None], 'a'), 'a')
test_eq(_fix_anno(float, 0.9), 0.9)
test_eq(_fix_anno(int, '1'), 1)
test_eq(_fix_anno(int, ['1','2']), 2)
test_eq(_fix_anno(list[int], ['1','2']), [1,2])
test_eq(_fix_anno(list[int], '1'), [1])
test_eq(_fix_anno(Optional[list[str]], ['1','2']), ['1','2'])
test_eq(_fix_anno(list[int] | None, ['1','2']), [1,2])
test_eq(_fix_anno(list, ['1','2']), None)
test_eq(_fix_anno(tuple, ['1','2']), None)
test_eq(_fix_anno(list[str], []), [])
test_eq(_fix_anno(Optional[list[str]], '1'), ['1'])

d = dict(k=int, l=List[int])
test_eq(_form_arg('k', "1", d), 1)
test_eq(_form_arg('l', "1", d), [1])
test_eq(_form_arg('l', ["1","2"], d), [1,2])

---

HttpHeader

def HttpHeader(
    k:str, v:str
)->None:

A response header as a k,v pair

_to_htmx_header('trigger_after_settle')

'HX-Trigger-After-Settle'

---

HtmxResponseHeaders

def HtmxResponseHeaders(
    *, location:NoneType=None, push_url:NoneType=None, redirect:NoneType=None, refresh:NoneType=None,
    replace_url:NoneType=None, reswap:NoneType=None, retarget:NoneType=None, reselect:NoneType=None,
    trigger:NoneType=None, trigger_after_settle:NoneType=None, trigger_after_swap:NoneType=None
):

HTMX response headers

HtmxResponseHeaders(trigger_after_settle='hi')

HttpHeader(k='HX-Trigger-After-Settle', v='hi')

@dataclass
class Foo: a:int; b:str
_annotations(Foo)

{'a': int, 'b': str}

Point = namedtuple('Point', ['x', 'y'])
_annotations(Point)

{'x': str, 'y': str}

---

form2dict

def form2dict(
    form:starlette.datastructures.FormData
)->dict:

Convert starlette form data to a dict

d = [('a',1),('a',2),('b',0)]
fd = FormData(d)
res = form2dict(fd)
test_eq(res['a'], [1,2])
test_eq(res['b'], 0)

---

parse_form

async def parse_form(
    req:starlette.requests.Request
)->starlette.datastructures.FormData:

Starlette errors on empty multipart/json forms and 500s on malformed multipart, so this handles those situations

Use the special __from_request__ class method for custom creation:

class Foo(BasicRepr):
    def __init__(self, d): self.d = d

    @classmethod
    def __from_request__(cls, data, req): return cls(data)

p = Parameter('a', Parameter.POSITIONAL_OR_KEYWORD, annotation=Foo)

req = test_request(method='POST')
async def mock_form(**kwargs): return FormData({'a':1})
req.form = mock_form

---

ApiReturn

def ApiReturn(
    isapi:bool=False
):

Request param: call with browser and API responses; returns kwargs as JSON if the client accepts application/json, else norm

Routes often need to return different responses depending on whether the request comes from a browser or an API client. The api:ApiReturn parameter handles this automatically — add it to any route signature and call it with two arguments: the API response and the browser response. If the request’s Accept header is application/json then a dict of the kwargs is returned, otherwise, the first value is returned. For example, return api(Redirect('/records'), id=record.id) returns {‘id’:id} for API callers and a redirect for the browser. This keeps your route logic in one place instead of splitting it across separate endpoints.

---

JSONResponse

def JSONResponse(
    content:Any, status_code:int=200, headers:Mapping[str, str] | None=None, media_type:str | None=None,
    background:BackgroundTask | None=None
)->None:

Same as starlette’s version, but auto-stringifies non serializable types

async def f(req):
    def _f(p:HttpHeader): ...
    p = first(_params(_f).values())
    data = form2dict(await parse_form(req))
    result = await _from_body(req, p, data)
    return JSONResponse(result.__dict__)

client = TestClient(Starlette(routes=[Route('/', f, methods=['POST'])]))

d = dict(k='value1',v=['value2','value3'])
response = client.post('/', data=d)
print(response.json())

{'k': 'value1', 'v': 'value3'}

# An empty `application/json` body yields `{}` (like the empty-multipart guard), not a 500
async def _empty_json(req): return JSONResponse(form2dict(await parse_form(req)))
client = TestClient(Starlette(routes=[Route('/', _empty_json, methods=['POST'])]))
res = client.post('/', headers={'content-type':'application/json'})
test_eq(res.status_code, 200)
test_eq(res.json(), {})
res = client.post('/', json={'a':1})          # non-empty json still parses (behavior preserved)
test_eq(res.json(), {'a':1})

A malformed multipart body (here the opening boundary isn’t followed by CRLF) is the client’s error, so parse_form answers 400 rather than letting the parser’s exception surface as a 500:

bad = b'--xyzXX\r\nContent-Disposition: form-data; name="a"\r\n\r\n1\r\n--xyz--\r\n'
res = client.post('/', content=bad, headers={'content-type':'multipart/form-data; boundary=xyz'})
test_eq(res.status_code, 400)
res.text

A JSON body that does not decode, or decodes to something other than an object (an array, say), cannot supply route parameters, so it is refused with 400 rather than failing inside the form conversion:

res = client.post('/', json=[1,2])
test_eq(res.status_code, 400)
res = client.post('/', content=b'not json', headers={'content-type':'application/json'})
test_eq(res.status_code, 400)
res.text

async def f(req): return Response(str(req.query_params.getlist('x')))
client = TestClient(Starlette(routes=[Route('/', f, methods=['GET'])]))
client.get('/?x=1&x=2').text

"['1', '2']"

data = form2dict(await parse_form(req))
await _from_body(req, p, data)

Foo(d={'a': 1})

def g(test_val:int=0, item:HttpHeader=None, htmx:HtmxHeaders=None, api:ApiReturn=None): ...

async def f(req):
    await _wrap_req(req, _params(g))
    body = await req.body()
    return Response(f"ok body={len(body)}")

client = TestClient(Starlette(routes=[Route('/', f, methods=['POST'])]))
r = client.post('/', data=dict(test_val='21', k='a', v='b'), headers={'HX-Request': '1'})
test_eq(r.status_code, 200)
print(r.text)

ok body=19

def g(req, this:Starlette, a:str, b:HttpHeader, state): ...

async def f(req):
    a = await _wrap_req(req, _params(g))
    return Response(str(a))

client = TestClient(Starlette(routes=[Route('/', f, methods=['POST'])]))
response = client.post('/?a=1', data=d)
print(response.text)

{'req': <starlette.requests.Request object>, 'this': <starlette.applications.Starlette object>, 'a': '1', 'b': HttpHeader(k='value1', v='value3'), 'state': <starlette.datastructures.State object>}

def g(req, this:Starlette, a:str, b:HttpHeader): ...

async def f(req):
    a = await _wrap_req(req, _params(g))
    return Response(str(a))

client = TestClient(Starlette(routes=[Route('/', f, methods=['POST'])]))
response = client.post('/?a=1', data=d)
print(response.text)

{'req': <starlette.requests.Request object>, 'this': <starlette.applications.Starlette object>, 'a': '1', 'b': HttpHeader(k='value1', v='value3')}

class Bar(BasicRepr):
    @classmethod
    def __from_request__(cls, accept:str=None): return cls(accept)
    def __init__(self, accept): self.accept = accept

def g(b:Bar): ...

async def f(req):
    result = await _wrap_req(req, _params(g))
    return Response(repr(result['b']))

client = TestClient(Starlette(routes=[Route('/', f, methods=['POST'])]))
client.post('/', headers={'Accept': 'application/json'}).text

"Bar(accept='application/json')"

Missing Request Params

If a request param has a default value (e.g. a:str=''), the request is valid even if the user doesn’t include the param in their request.

def g(req, this:Starlette, a:str=''): ...

async def f(req):
    a = await _wrap_req(req, _params(g))
    return Response(str(a))

client = TestClient(Starlette(routes=[Route('/', f, methods=['POST'])]))
response = client.post('/', json={}) # no param in request
print(response.text)

{'req': <starlette.requests.Request object>, 'this': <starlette.applications.Starlette object>, 'a': ''}

If we remove the default value and re-run the request, we should get the following error Missing required field: a.

def g(req, this:Starlette, a:str): ...

async def f(req):
    a = await _wrap_req(req, _params(g))
    return Response(str(a))

client = TestClient(Starlette(routes=[Route('/', f, methods=['POST'])]))
response = client.post('/', json={}) # no param in request
print(response.text)

Missing required field: a

---

flat_xt

def flat_xt(
    lst
):

Flatten lists

x = ft('a',1)
test_eq(flat_xt([x, x, [x,x]]), (x,)*4)
test_eq(flat_xt(x), (x,))

---

Beforeware

def Beforeware(
    f, skip:NoneType=None
):

Wrap a before function with skip path patterns that bypass it

async def _f(f): return f"got {f}"
test_eq(await _handle(_f, f='hello'), 'got hello')

Websockets / SSE

def on_receive(msg:str): return f"Message text was: {msg}"
c = _ws_endp(on_receive)
cli = TestClient(Starlette(routes=[WebSocketRoute('/', _ws_endp(on_receive))]))
with cli.websocket_connect('/') as ws:
    ws.send_text('{"msg":"Hi!"}')
    data = ws.receive_text()
    assert data == 'Message text was: Hi!', data

def on_receive(hx_trigger:str): return f"trigger: {hx_trigger}"
cli = TestClient(Starlette(routes=[WebSocketRoute('/', _ws_endp(on_receive))]))
with cli.websocket_connect('/') as ws:
    ws.send_text(json.dumps({"HEADERS": {"Hx-Trigger": "my-btn"}}))
    data = ws.receive_text()
    assert data == 'trigger: my-btn', data

---

EventStream

def EventStream(
    s
):

Create a text/event-stream response from s

---

signal_shutdown

def signal_shutdown():

Return an asyncio.Event set when uvicorn receives a shutdown signal

Routing and application

---

uri

def uri(
    _arg, **kwargs
):

Create a URI by URL-encoding _arg and appending query parameters from kwargs

---

decode_uri

def decode_uri(
    s
):

Decode a URI created by uri() back into argument and keyword dict

---

StringConvertor.to_string

def to_string(
    value:str
)->str:

---

HTTPConnection.url_path_for

def url_path_for(
    name:str, **path_params
):

---

flat_tuple

def flat_tuple(
    o
):

Flatten nested iterables into a single tuple

---

noop_body

def noop_body(
    c, req
):

Default Body wrap function which just returns the content

---

respond

def respond(
    req, heads, bdy
):

Default FT response creation function

Render fragment if HX-Request header is present and HX-History-Restore-Request header is absent.

---

is_full_page

def is_full_page(
    req, resp
):

Check if response should be rendered as full page or fragment

---

Redirect

def Redirect(
    loc
):

Use HTMX or Starlette RedirectResponse as required to redirect to loc

The FastHTML exts param supports the following:

print(' '.join(htmx_exts))

morph head-support preload class-tools loading-states multi-swap path-deps remove-me debug ws chunked-transfer

---

get_key

def get_key(
    key:NoneType=None, fname:str='.sesskey'
):

Get session key from key param or read/create from file fname

---

qp

def qp(
    p:str, **kw
)->str:

Add parameters kw to path p

qp adds query parameters to route path strings

vals = dict(a=5, b=False, c=[1,2], d='bar', e=None, ab=42)

res = qp('/foo', **vals)
test_eq(res, '/foo?a=5&b=&c=1&c=2&d=bar&e=&ab=42')

qp checks to see if each param should be sent as a query parameter or as part of the route, and encodes that properly.

path = '/foo/{a}/{d}/{ab:int}'
res = qp(path, **vals)
test_eq(res, '/foo/5/bar/42?b=&c=1&c=2&e=')

---

def_hdrs

def def_hdrs(
    htmx:bool=True, htmx4:bool=False, surreal:bool=True
):

Default headers for a FastHTML app

---

Lifespan

def Lifespan(
    startup:NoneType=None, shutdown:NoneType=None, ls:NoneType=None
):

Combine on_startup/on_shutdown callbacks and an optional lifespan generator into one context manager

---

FastHTML

def FastHTML(
    debug:bool=False, routes:NoneType=None, middleware:NoneType=None, title:str='FastHTML page',
    exception_handlers:NoneType=None, on_startup:NoneType=None, on_shutdown:NoneType=None, lifespan:NoneType=None,
    hdrs:NoneType=None, ftrs:NoneType=None, exts:NoneType=None, before:NoneType=None, after:NoneType=None,
    surreal:bool=True, htmx:bool=True, htmx4:bool=False, default_hdrs:bool=True, sess_cls:type=SessionMiddleware,
    secret_key:NoneType=None, session_cookie:str='session_', max_age:int=31536000, sess_path:str='/',
    same_site:str='lax', sess_https_only:bool=False, sess_domain:NoneType=None, key_fname:str='.sesskey',
    body_wrap:function=noop_body, htmlkw:NoneType=None, nb_hdrs:bool=False, canonical:bool=True,
    max_part_size:int=104857600, **bodykw
):

An HTML-first Starlette app: handler params filled from the request, FT returns rendered as pages or HTMX fragments

Routes

---

HostRoute

def HostRoute(
    path, endpoint, *, host:NoneType=None, **kwargs
):

Route with optional host-header constraint using Starlette’s {param} pattern syntax

app = FastHTML()
app.router.routes = []

app.router.routes.append(HostRoute('/', lambda _: Response('auth page'), host='auth.{parentdom}', methods=['GET'], name='auth_home'))
app.router.routes.append(HostRoute('/', lambda _: Response('main page'), methods=['GET'], name='main_home'))

cli = TestClient(app)
test_eq(cli.get('/', headers={'host': 'auth.localhost'}).text, 'auth page')
test_eq(cli.get('/', headers={'host': 'localhost'}).text, 'main page')
test_eq(cli.get('/', headers={'host': 'auth.example.com'}).text, 'auth page')
print('all passed')

all passed

app.router.routes.append(HostRoute('/hello', lambda _: Response('api hello'), host='api.{parentdom}', methods=['GET'], name='api_hello'))
app.router.routes.append(HostRoute('/hello', lambda _: Response('normal hello'), methods=['GET'], name='hello'))

test_eq(cli.get('/hello', headers={'host': 'api.mysite.com'}).text, 'api hello')
test_eq(cli.get('/hello', headers={'host': 'mysite.com'}).text, 'normal hello')
test_eq(cli.get('/hello', headers={'host': 'other.mysite.com'}).text, 'normal hello')
print('all passed')

all passed

---

FastHTML.add_route

def add_route(
    route
):

Add or replace a route in the FastHTML app

---

FastHTML.ws

def ws(
    path:str, conn:NoneType=None, disconn:NoneType=None, name:NoneType=None, middleware:NoneType=None
):

Add a websocket route at path

---

FastHTML.add_websocket_route

def add_websocket_route(
    path, func, conn:NoneType=None, disconn:NoneType=None, name:NoneType=None, middleware:NoneType=None
):

Add a websocket route at path (Starlette-compatible API)

---

nested_name

def nested_name(
    f
):

*Get name of function f using ’_’ to join nested function names*

def f():
    def g(): ...
    return g

func = f()
nested_name(func)

'f_g'

---

FastHTML.route

def route(
    path:str=None, methods:NoneType=None, name:NoneType=None, include_in_schema:bool=True, body_wrap:NoneType=None,
    host:NoneType=None, before:Union[Callable, tuple, NoneType]=None
):

Add a route at path

app = FastHTML()
@app.get
def foo(a:str, b:list[int]): ...

foo.to(a='bar', b=[1,2])

'/foo?a=bar&b=1&b=2'

@app.get('/foo/{a}')
def foo(a:str, b:list[int]): ...

foo.to(a='bar', b=[1,2])

'/foo/bar?b=1&b=2'

Utils

---

FastHTML.set_lifespan

def set_lifespan(
    value
):

Set the lifespan context manager for the FastHTML app

---

serve

def serve(
    appname:NoneType=None, # Name of the module
    app:str='app', # App instance to be served
    host:str='0.0.0.0', # If host is 0.0.0.0 will convert to localhost
    port:NoneType=None, # If port is None it will default to 5001 or the PORT environment variable
    reload:bool=True, # Default is to reload the app upon code changes
    procs:NoneType=None, # Processes to run on consecutive ports from `port`; default 1 or the PROCS environment variable
    **kwargs
):

Run the app in an async server, with live reload set as the default.

procs runs the app as that many processes on consecutive ports from port, each with its own reload supervisor, and defaults to the PROCS environment variable. serve re-executes the script once per process with FH_PROCNUM set to its number, which is also the guard against a child spawning again. An app reads FH_PROCNUM for work that only one process should do, such as a background job: it is '0' in the first process, and unset in an ordinary single-process run. A reverse proxy in front of the ports decides how requests reach them.

Starlette does not cancel a running handler when a client disconnects: the disconnect is only visible as an http.disconnect message on the ASGI receive channel, which ordinary handlers never read. until_disconnect races a coroutine against that message, cancelling the work if the client goes away first.

---

until_disconnect

async def until_disconnect(
    req, coro
):

Await coro, cancelling it (and returning an empty response) if the client disconnects first

cancel_on_disconnect applies this to a route handler. splice_sig adds a private Request-annotated parameter to the exposed signature, which FastHTML fills with the connection, so the handler’s own parameters are unchanged. It requires an async handler, since a sync handler runs in a threadpool and cannot be cancelled.

---

cancel_on_disconnect

def cancel_on_disconnect(
    f:NoneType=None, # Async route handler to wrap
    *, timeout:NoneType=None, # Also cancel the handler after this many seconds
):

Route decorator: cancel the handler when the client disconnects

---

Client

def Client(
    app, url:str='http://testserver'
):

A simple httpx ASGI client that doesn’t require async

app = FastHTML(routes=[Route('/', lambda _: Response('test'))])
cli = Client(app)

cli.get('/').text

'test'

Note that you can also use Starlette’s TestClient instead of FastHTML’s Client. They should be largely interchangable.

FastHTML Tests

def get_cli(app): return app,TestClient(app),app.route

app,cli,rt = get_cli(FastHTML(secret_key='soopersecret'))

app,cli,rt = get_cli(FastHTML(title="My Custom Title"))
@app.get
def foo(): return Div("Hello World")

print(app.routes)

response = cli.get('/foo')
assert '<title>My Custom Title</title>' in response.text

foo.to(param='value')

[HostRoute(path='/foo', name='foo', methods=['GET', 'HEAD'], host=None)]

'/foo?param=value'

@rt(host='test.foo')
def foo(): return Div("Goodbye World")

print(app.routes)

assert "Goodbye World" in cli.get('/foo', headers={'host': 'test.foo'}).text
assert "Hello World" in cli.get('/foo').text

[HostRoute(path='/foo', name='foo', methods=['GET', 'HEAD', 'POST'], host='test.foo'), HostRoute(path='/foo', name='foo', methods=['GET', 'HEAD'], host=None)]

response = cli.get('/foo')
assert '<title>My Custom Title</title>' in response.text

foo.to(param='value')

'/foo?param=value'

app,cli,rt = get_cli(FastHTML())

@rt('/xt2')
def get(): return H1('bar')

txt = cli.get('/xt2').text
assert '<title>FastHTML page</title>' in txt and '<h1>bar</h1>' in txt and '<html>' in txt

def test_r(cli, path, exp, meth='get', hx=False, **kwargs):
    if hx: kwargs['headers'] = {'hx-request':"1"}
    test_eq(getattr(cli, meth)(path, **kwargs).text, exp)

@dataclass
class Card: id:str; title:str

@rt('/class-cards/{id}')
class CardMethods:
    async def get(id:str): return f'get {id}'
    async def patch(card:Card): return asdict(card)
    async def delete(id:str): return f'delete {id}'

test_eq(cli.get('/class-cards/abc').text, 'get abc')
test_r(cli, '/class-cards/abc', '{"id":"abc","title":"Foo"}', 'patch', data={'title':'Foo'})
test_eq(cli.delete('/class-cards/abc').text, 'delete abc')
test_eq(CardMethods.to(id='xyz'), '/class-cards/xyz')

@rt
class Things:
    def get(id:str): return f'thing {id}'

test_eq(cli.get('/things?id=7').text, 'thing 7')
test_eq(Things.to(id=7), '/things?id=7')

@rt("/hi")
def get(): return 'Hi there'

r = cli.get('/hi')
r.text

'Hi there'

@rt("/hi")
def post(): return 'Postal'

cli.post('/hi').text

'Postal'

@app.get("/hostie")
def show_host(req): return req.headers['host']

cli.get('/hostie').text

'testserver'

@app.get("/setsess")
def set_sess(session):
   session['foo'] = 'bar'
   return 'ok'

@app.ws("/ws")
def ws(self, msg:str, ws:WebSocket, session): return f"Message text was: {msg} with session {session.get('foo')}, from client: {ws.client}"

cli.get('/setsess')
with cli.websocket_connect('/ws') as ws:
    ws.send_text('{"msg":"Hi!"}')
    data = ws.receive_text()
assert 'Message text was: Hi! with session bar' in data
print(data)

Message text was: Hi! with session bar, from client: Address(host='testclient', port=50000)

@rt
def yoyo(): return 'a yoyo'

cli.post('/yoyo').text

'a yoyo'

@app.get
def autopost(): return Html(Div('Text.', hx_post=yoyo()))
print(cli.get('/autopost').text)

<!doctype html>
<html>
  <div hx-post="a yoyo">Text.</div>
</html>

@app.get
def autopost2(): return Html(Body(Div('Text.', cls='px-2', hx_post=show_host.to(a='b'))))
print(cli.get('/autopost2').text)

<!doctype html>
<html>
  <body>
    <div class="px-2" hx-post="/hostie?a=b">Text.</div>
  </body>
</html>

@app.get
def autoget2(): return Html(Div('Text.', hx_get=show_host))
print(cli.get('/autoget2').text)

<!doctype html>
<html>
  <div hx-get="/hostie">Text.</div>
</html>

@rt('/user/{nm}', name='gday')
def get(nm:str=''): return f"Good day to you, {nm}!"
cli.get('/user/Alexis').text

'Good day to you, Alexis!'

@app.get
def autolink(): return Html(Div('Text.', link=uri('gday', nm='Alexis')))
print(cli.get('/autolink').text)

<!doctype html>
<html>
  <div href="/user/Alexis">Text.</div>
</html>

@rt('/link')
def get(req): return f"{req.url_for('gday', nm='Alexis')}; {req.url_for('show_host')}"

cli.get('/link').text

'http://testserver/user/Alexis; http://testserver/hostie'

@app.get("/background")
async def background_task(request):
    async def long_running_task():
        await asyncio.sleep(0.1)
        print("Background task completed!")
    return P("Task started"), BackgroundTask(long_running_task)

response = cli.get("/background")

Background task completed!

test_eq(app.router.url_path_for('gday', nm='Jeremy'), '/user/Jeremy')

hxhdr = {'headers':{'hx-request':"1"}}

@rt('/ft')
def get(): return Title('Foo'),H1('bar')

txt = cli.get('/ft').text
assert '<title>Foo</title>' in txt and '<h1>bar</h1>' in txt and '<html>' in txt

@rt('/xt2')
def get(): return H1('bar')

txt = cli.get('/xt2').text
assert '<title>FastHTML page</title>' in txt and '<h1>bar</h1>' in txt and '<html>' in txt

assert cli.get('/xt2', **hxhdr).text.strip() == '<h1>bar</h1>'

@rt('/xt3')
def get(): return Html(Head(Title('hi')), Body(P('there')))

txt = cli.get('/xt3').text
assert '<title>FastHTML page</title>' not in txt and '<title>hi</title>' in txt and '<p>there</p>' in txt

# declaration-time warnings
def _noanno(nope): return nope
def _bare(names:list): return str(names)
test_warns(lambda: rt('/oops')(_noanno))
test_warns(lambda: rt('/barelist')(_bare))

# runtime warnings
test_warns(lambda: cli.get('/oops?nope=1'))
test_warns(lambda: cli.get('/barelist?names=1'))

ModelName = str_enum('ModelName', "alexnet", "resnet", "lenet")
fake_db = [{"name": "Foo"}, {"name": "Bar"}]

@rt('/html/{idx}')
async def get(idx:int): return Body(H4(f'Next is {idx+1}.'))

@rt("/models/{nm}")
def get(nm:ModelName): return nm

@rt("/files/{path}")
async def get(path: Path): return path.with_suffix('.txt')

@rt("/items/")
def get(idx:int|None = 0): return fake_db[idx]

@rt("/idxl/")
def get(idx:list[int]): return str(idx)

@rt("/optidxl/")
def get(idx:Optional[list[int]]): return str(idx)  # chkstyle: ignore

@rt("/unionidxl/")
def get(idx:list[int] | None): return str(idx)

r = cli.get('/html/1', headers={'hx-request':"1"})
assert '<h4>Next is 2.</h4>' in r.text
test_r(cli, '/models/alexnet', 'alexnet')
test_r(cli, '/files/foo', 'foo.txt')
test_r(cli, '/items/?idx=1', '{"name":"Bar"}')
test_r(cli, '/items/', '{"name":"Foo"}')
assert cli.get('/items/?idx=g').text=='404 Not Found'
assert cli.get('/items/?idx=g').status_code == 404
test_r(cli, '/idxl/?idx=1&idx=2', '[1, 2]')
assert cli.get('/idxl/?idx=1&idx=g').status_code == 404
test_r(cli, '/optidxl/?idx=1&idx=2', '[1, 2]')
test_r(cli, '/unionidxl/?idx=1&idx=2', '[1, 2]')

app = FastHTML()
rt = app.route
cli = TestClient(app)
@app.route(r'/static/{path:path}.jpg')
def index(path:str): return f'got {path}'
@app.route(r'/static/{path:path}')
def foo(path:str, a:int): return f'also got {path},{a}'
cli.get('/static/sub/a.b.jpg').text

'got sub/a.b'

cli.get('/static/sub/a.b?a=1').text

'also got sub/a.b,1'

app.chk = 'foo'

@app.get("/booly/")
def _(coming:bool=True): return 'Coming' if coming else 'Not coming'

@app.get("/datie/")
def _(d:parsed_date): return d

@app.get("/ua")
async def _(user_agent:str): return user_agent

@app.get("/hxtest")
def _(htmx): return htmx.request

@app.get("/hxtest2")
def _(foo:HtmxHeaders, req): return foo.request

@app.get("/app")
def _(app): return app.chk

@app.get("/app2")
def _(foo:FastHTML): return foo.chk,HttpHeader("mykey", "myval")

@app.get("/app3")
def _(foo:FastHTML): return HtmxResponseHeaders(location="http://example.org")

@app.get("/app4")
def _(foo:FastHTML): return Redirect("http://example.org")

test_r(cli, '/booly/?coming=true', 'Coming')
test_r(cli, '/booly/?coming=no', 'Not coming')
date_str = "17th of May, 2024, 2p"
test_r(cli, f'/datie/?d={date_str}', '2024-05-17 14:00:00')
test_r(cli, '/ua', 'FastHTML', headers={'User-Agent':'FastHTML'})
test_r(cli, '/hxtest' , '1', headers={'HX-Request':'1'})
test_r(cli, '/hxtest2', '1', headers={'HX-Request':'1'})
test_r(cli, '/app' , 'foo')

r = cli.get('/app2', **hxhdr)
test_eq(r.text, 'foo')
test_eq(r.headers['mykey'], 'myval')

r = cli.get('/app3')
test_eq(r.headers['HX-Location'], 'http://example.org')

r = cli.get('/app4', follow_redirects=False)
test_eq(r.status_code, 303)

r = cli.get('/app4', headers={'HX-Request':'1'})
test_eq(r.headers['HX-Redirect'], 'http://example.org')

@rt
def meta(): return ((Title('hi'), H1('hi')), (Meta(property='image'), Meta(property='site_name')))

t = cli.post('/meta').text
assert re.search(r'<body>\s*<h1>hi</h1>\s*</body>', t)
assert '<meta' in t

@app.post('/profile/me')
def profile_update(user_name: str): return user_name

test_r(cli, '/profile/me', 'Alexis', 'post', data={'user-name':'Alexis'})
test_r(cli, '/profile/me?user-name=Jeremy&user-name=Alexis', 'Alexis', 'post')
test_r(cli, '/profile/me', 'Jeremy', 'post', data={'user-name':'Alexis', 'user_name':'Jeremy'})
test_r(cli, '/profile/me', 'Missing required field: user_name', 'post', data={})

# Example post request with parameter that has a default value
@app.post('/pet/dog')
def pet_dog(dogname: str = None): return dogname

# Working post request with optional parameter
test_r(cli, '/pet/dog', '', 'post', data={})

@dataclass
class Bodie: a:int;b:str

@rt("/bodie/{nm}")
def post(nm:str, data:Bodie):
    res = asdict(data)
    res['nm'] = nm
    return res

@rt("/cards/{id}")
def put(card:Card): return asdict(card)

@app.post("/bodied/")
def bodied(data:dict): return data

nt = namedtuple('Bodient', ['a','b'])

@app.post("/bodient/")
def bodient(data:nt): return asdict(data)

class BodieTD(TypedDict): a:int;b:str='foo'

@app.post("/bodietd/")
def bodient(data:BodieTD): return data

class Bodie2:
    a_val:int|None; b:str
    def __init__(self, a_val, b='foo'): store_attr()

@rt("/bodie2/", methods=['get','post'])
def bodie(d:Bodie2): return f"a: {d.a_val}; b: {d.b}"

from fasthtml.xtend import Titled

d = dict(a=1, b='foo')

test_r(cli, '/bodie/me', '{"a":1,"b":"foo","nm":"me"}', 'post', data=dict(a=1, b='foo', nm='me'))
test_r(cli, '/cards/abc', '{"id":"abc","title":"Foo"}', 'put', data={'title':'Foo'})
test_r(cli, '/bodied/', '{"user-name":"Alexis"}', 'post', data={'user-name':'Alexis'})
test_r(cli, '/bodie2/', 'a: 1; b: foo', 'post', data={'a-val':2, 'a_val':1})
test_r(cli, '/bodie2/?a-val=1&b=foo&nm=me', 'a: 1; b: foo')
test_r(cli, '/bodient/', '{"a":"1","b":"foo"}', 'post', data=d)
test_r(cli, '/bodietd/', '{"a":1,"b":"foo"}', 'post', data=d)

# Testing POST with Content-Type: application/json
@app.post("/")
def index(it: Bodie): return Titled("It worked!", P(f"{it.a}, {it.b}"))

s = json.dumps({"b": "Lorem", "a": 15})
response = cli.post('/', headers={"Content-Type": "application/json"}, data=s).text
assert "<title>It worked!</title>" in response and "<p>15, Lorem</p>" in response

# Testing POST with Content-Type: application/json
@app.post("/bodytext")
def index(body): return body

response = cli.post('/bodytext', headers={"Content-Type": "application/json"}, data=s).text
test_eq(response, '{"b": "Lorem", "a": 15}')

files = [('files', ('file1.txt', b'content1')), ('files', ('file2.txt', b'content2'))]

@rt("/uploads")
async def post(files:list[UploadFile]): return ','.join([(await file.read()).decode() for file in files])

res = cli.post('/uploads', files=files)
print(res.status_code)
print(res.text)

200
content1,content2

res = cli.post('/uploads', files=[files[0]])
test_eq(res.status_code, 200)
test_eq(res.text, 'content1')

@rt("/setsess")
def get(sess, foo:str=''):
    now = datetime.now()
    sess['auth'] = str(now)
    return f'Set to {now}'

@rt("/getsess")
def get(sess): return f'Session time: {sess["auth"]}'

print(cli.get('/setsess').text)
time.sleep(0.01)

cli.get('/getsess').text

Set to 2026-06-29 21:09:52.332526

'Session time: 2026-06-29 21:09:52.332526'

@rt("/sess-first")
def post(sess, name: str):
    sess["name"] = name
    return str(sess)

cli.post('/sess-first', data={'name': 2})

@rt("/getsess-all")
def get(sess): return sess['name']

test_eq(cli.get('/getsess-all').text, '2')

@rt("/upload")
async def post(uf:UploadFile): return (await uf.read()).decode()

with open('../../CHANGELOG.md', 'rb') as f: print(cli.post('/upload', files={'uf':f}, data={'msg':'Hello'}).text[:15])

# Release notes

@rt("/form-submit/{list_id}")
def options(list_id: str):
    headers = {'Access-Control-Allow-Origin': '*', 'Access-Control-Allow-Methods': 'POST', 'Access-Control-Allow-Headers': '*'}
    return Response(status_code=200, headers=headers)

h = cli.options('/form-submit/2').headers
test_eq(h['Access-Control-Allow-Methods'], 'POST')

from fasthtml.authmw import user_pwd_auth

def _not_found(req, exc): return Div('nope')

app,cli,rt = get_cli(FastHTML(exception_handlers={404:_not_found}))

txt = cli.get('/').text
assert '<div>nope</div>' in txt
assert '<!doctype html>' in txt

app,cli,rt = get_cli(FastHTML())

@rt("/{name}/{age}")
def get(name: str, age: int): return Titled(f"Hello {name.title()}, age {age}")

assert '<title>Hello Uma, age 5</title>' in cli.get('/uma/5').text
assert '404 Not Found' in cli.get('/uma/five').text

auth = user_pwd_auth(testuser='spycraft')
app,cli,rt = get_cli(FastHTML(middleware=[auth]))

@rt("/locked")
def get(auth): return 'Hello, ' + auth

test_eq(cli.get('/locked').text, 'not authenticated')
test_eq(cli.get('/locked', auth=("testuser","spycraft")).text, 'Hello, testuser')

auth = user_pwd_auth(testuser='spycraft')
app,cli,rt = get_cli(FastHTML(middleware=[auth]))

@rt("/locked")
def get(auth): return 'Hello, ' + auth

test_eq(cli.get('/locked').text, 'not authenticated')
test_eq(cli.get('/locked', auth=("testuser","spycraft")).text, 'Hello, testuser')

Check that afterware like toasts are called correctly.

from fasthtml.toaster import *

app,cli,rt = get_cli(FastHTML())

# Modified toast functions to test resp logic
def toast_after_(resp, req, sess):
    if sk in sess and (not resp or isinstance(resp, (tuple,FT,FtResponse))):
        sess['toast_duration'] = req.app.state.toast_duration
        # We will activate the toast depending on the content of the response
        if str('love') in str(resp): req.injects.append(render_toasts(sess))

def setup_toasts_(app, duration=5000):
    app.state.toast_duration = duration
    app.hdrs += [Style(toast_css), Script(js)]
    app.after.append(toast_after_)

setup_toasts_(app)

tm = 'Here is a toast!'

@rt("/")
def index(sess): 
    add_toast(sess, tm, "success")
    return P("I love toast!")

@rt("/no-toast")
def no_toast(sess):
    add_toast(sess, tm, "success")
    return P("I hate toast!")

r = cli.get('/no-toast')
assert not tm in r.text

r = cli.get('/')
assert tm in r.text

Here’s a quick sanity check to confirm that toasts are rendered in the toast container correctly. See issue.

assert 'id="fh-toast-container"' in r.text

app,cli,rt = get_cli(FastHTML(on_startup=[lambda: setattr(app, '_started', True)],
    on_shutdown=[lambda: setattr(app, '_stopped', True)]))

@rt
def index(): return str(app._started)

with cli: test_eq(cli.get('/').text, 'True')
test_eq(app._stopped, True)

app,cli,rt = get_cli(FastHTML())

@app.on_event('startup')
def _(): app._evt = 'fired'

@rt
def index(): return app._evt

with cli: test_eq(cli.get('/').text, 'fired')

A route-level before can set session data, short-circuit with a response, or raise to block access:

def check_user(sess, user:str=''):
    if user == 'admin':
        sess['foo'] = 'bar'
        return None
    if user == 'guest': return 'guest view'
    raise HTTPException(403, 'forbidden')

@rt('/dashboard', before=check_user)
def get(sess): return sess.get('foo')

test_eq(cli.get('/dashboard?user=admin').text, 'bar')
test_eq(cli.get('/dashboard?user=guest').text, 'guest view')
test_eq(cli.get('/dashboard?user=other').status_code, 403)

Multiple before functions run in order, each receiving the injected request params:

def set_a(sess): sess['a'] = 1
def set_b(sess): sess['b'] = 2

@rt('/multi', before=(set_a, set_b))
def get(sess): return f"{sess.get('a')},{sess.get('b')}"

test_eq(cli.get('/multi').text, '1,2')

A truthy return from any before becomes the response; remaining befores and the handler are skipped (return None to continue):

calls = []
def first(sess): 
    calls.append('first')
    return 'stop'
def second(sess): calls.append('second')

@rt('/chain', before=(first, second))
def get(sess): return 'handler'

test_eq(cli.get('/chain').text, 'stop')
test_eq(calls, ['first'])

cancel_on_disconnect handlers keep their normal parameters; the handler is cancelled if the client disconnects, or after timeout seconds if given:

app,cli,rt = get_cli(FastHTML())

@rt('/cq')
@cancel_on_disconnect
async def cq(q:str): return f'got {q}'

test_eq(cli.get('/cq?q=x').text, 'got x')

@rt('/sleepy')
@cancel_on_disconnect(timeout=0.01)
async def sleepy():
    await asyncio.sleep(1)
    return 'nope'

test_fail(lambda: cli.get('/sleepy'))

On disconnect the handler is cancelled and an empty response returned, which the server discards since the client is gone. Here we simulate the ASGI http.disconnect message directly:

async def _disc(): return {'type': 'http.disconnect'}
cancelled = []
async def _work():
    try: await asyncio.sleep(10)
    except asyncio.CancelledError:
        cancelled.append(True)
        raise

async def _t():
    scope = dict(type='http', method='GET', path='/', query_string=b'', headers=[])
    res = await until_disconnect(Request(scope, _disc), _work())
    await asyncio.sleep(0.01)
    return res

test_eq(run_sync(_t()).status_code, 204)
test_eq(cancelled, [True])

APIRouter

---

RouteFuncs

def RouteFuncs():

Attr-access store of named route functions (HTTP verb names excluded)

---

APIRouter

def APIRouter(
    prefix:str | None=None, body_wrap:function=noop_body
):

Add routes to an app

ar = APIRouter()

@ar("/hi")
def get(): return 'Hi there'
@ar("/hi")
def post(): return 'Postal'
@ar
def ho(): return 'Ho ho'
@ar("/hostie")
def show_host(req): return req.headers['host']
@ar
def yoyo(): return 'a yoyo'
@ar
def index(): return "home page"
@ar('/class-cards/{id}')
class CardMethods:
    async def get(id:str): return f'get {id}'
    async def patch(card:Card): return asdict(card)
    async def delete(id:str): return f'delete {id}'


@ar.ws("/ws")
def ws(self, msg:str): return f"Message text was: {msg}"

app,cli,_ = get_cli(FastHTML())
ar.to_app(app)

assert str(yoyo) == '/yoyo'
# ensure route functions are properly discoverable on `APIRouter` and `APIRouter.rt_funcs`
assert ar.prefix == ''
assert str(ar.rt_funcs.index) == '/'
assert str(ar.index) == '/'
with ExceptionExpected(): ar.blah()
with ExceptionExpected(): ar.rt_funcs.blah()
# ensure any route functions named using an HTTPMethod are not discoverable via `rt_funcs`
assert "get" not in ar.rt_funcs._funcs.keys()

test_eq(cli.get('/hi').text, 'Hi there')
test_eq(cli.post('/hi').text, 'Postal')
test_eq(cli.get('/hostie').text, 'testserver')
test_eq(cli.post('/yoyo').text, 'a yoyo')

test_eq(cli.get('/ho').text, 'Ho ho')
test_eq(cli.post('/ho').text, 'Ho ho')

test_eq(cli.get('/class-cards/abc').text, 'get abc')
test_r(cli, '/class-cards/abc', '{"id":"abc","title":"Foo"}', 'patch', data={'title':'Foo'})
test_eq(cli.delete('/class-cards/abc').text, 'delete abc')
test_eq(CardMethods.to(id='xyz'), '/class-cards/xyz')

with cli.websocket_connect('/ws') as ws:
    ws.send_text('{"msg":"Hi!"}')
    data = ws.receive_text()
    assert data == 'Message text was: Hi!'

ar2 = APIRouter("/products")

@ar2("/hi")
def get(): return 'Hi there'
@ar2("/hi")
def post(): return 'Postal'
@ar2
def ho(): return 'Ho ho'
@ar2("/hostie")
def show_host(req): return req.headers['host']
@ar2
def yoyo(): return 'a yoyo'
@ar2
def index(): return "home page"
@ar2
class Products:
    def get(id:str): return f'product {id}'


@ar2.ws("/ws")
def ws(self, msg:str): return f"Message text was: {msg}"

app,cli,_ = get_cli(FastHTML())
ar2.to_app(app)

assert str(yoyo) == '/products/yoyo'
assert ar2.prefix == '/products'
assert str(ar2.rt_funcs.index) == '/products/'
assert str(ar2.index) == '/products/'
assert str(ar.index) == '/'
with ExceptionExpected(): ar2.blah()
with ExceptionExpected(): ar2.rt_funcs.blah()
assert "get" not in ar2.rt_funcs._funcs.keys()

test_eq(cli.get('/products/hi').text, 'Hi there')
test_eq(cli.post('/products/hi').text, 'Postal')
test_eq(cli.get('/products/hostie').text, 'testserver')
test_eq(cli.post('/products/yoyo').text, 'a yoyo')

test_eq(cli.get('/products/ho').text, 'Ho ho')
test_eq(cli.post('/products/ho').text, 'Ho ho')

test_eq(cli.get('/products/products?id=7').text, 'product 7')
test_eq(Products.to(id=7), '/products/products?id=7')

with cli.websocket_connect('/products/ws') as ws:
    ws.send_text('{"msg":"Hi!"}')
    data = ws.receive_text()
    assert data == 'Message text was: Hi!'

@ar.get
def hi2(): return 'Hi there'
@ar.get("/hi3")
def _(): return 'Hi there'
@ar.post("/post2")
def _(): return 'Postal'

@ar2.get
def hi2(): return 'Hi there'
@ar2.get("/hi3")
def _(): return 'Hi there'
@ar2.post("/post2")
def _(): return 'Postal'

Extras

app,cli,rt = get_cli(FastHTML(secret_key='soopersecret'))

---

cookie

def cookie(
    key:str, value:str='', max_age:NoneType=None, expires:NoneType=None, path:str='/', domain:NoneType=None,
    secure:bool=False, httponly:bool=False, samesite:str='lax'
):

Create a ‘set-cookie’ HttpHeader

@rt("/setcookie")
def get(req): return cookie('now', datetime.now())

@rt("/getcookie")
def get(now:parsed_date): return f'Cookie was set at time {now.time()}'

print(cli.get('/setcookie').text)
time.sleep(0.01)
cli.get('/getcookie').text

<!doctype html>
<html>
  <head>
    <title>FastHTML page</title>
    <link rel="canonical" href="https://testserver/setcookie">
    <meta charset="utf-8">
    <meta name="viewport" content="width=device-width, initial-scale=1, viewport-fit=cover">
<script src="https://cdn.jsdelivr.net/npm/[email protected]/dist/htmx.js"></script><script src="https://cdn.jsdelivr.net/gh/answerdotai/[email protected]/fasthtml.js"></script><script src="https://cdn.jsdelivr.net/gh/answerdotai/surreal@main/surreal.js"></script><script src="https://cdn.jsdelivr.net/gh/gnat/css-scope-inline@main/script.js"></script>  </head>
  <body></body>
</html>

'Cookie was set at time 21:09:55.336114'

---

reg_re_param

def reg_re_param(
    m, s
):

Register a Starlette URL convertor named m matching regex s

---

FastHTML.static_route_exts

def static_route_exts(
    prefix:str='/', static_path:str='.', exts:str='static'
):

Add a static route at URL path prefix with files from static_path and exts defined by reg_re_param()

reg_re_param("imgext", "ico|gif|jpg|jpeg|webm|pdf")

@rt(r'/static/{path:path}{fn}.{ext:imgext}')
def get(fn:str, path:str, ext:str): return f"Getting {fn}.{ext} from /{path}"

test_r(cli, '/static/foo/jph.me.ico', 'Getting jph.me.ico from /foo/')

app.static_route_exts()
assert 'These are the source notebooks for FastHTML' in cli.get('/README.txt').text

---

FastHTML.static_route

def static_route(
    ext:str='', prefix:str='/', static_path:str='.'
):

Add a static route at URL path prefix with files from static_path and single ext (including the ‘.’)

app.static_route('.md', static_path='../..')
assert 'THIS FILE WAS AUTOGENERATED' in cli.get('/README.md').text

---

StaticNoCache

def StaticNoCache(
    *, directory:PathLike | None=None, packages:list[str | tuple[str, str]] | None=None, html:bool=False,
    check_dir:bool=True, follow_symlink:bool=False
)->None:

StaticFiles that sends Cache-Control: no-cache, so browsers revalidate on every use

---

StaticImmutable

def StaticImmutable(
    *, directory:PathLike | None=None, packages:list[str | tuple[str, str]] | None=None, html:bool=False,
    check_dir:bool=True, follow_symlink:bool=False
)->None:

Like StaticNoCache, but marks responses as cacheable forever: for versioned URLs such as vurl links or content-hashed filenames

These two classes are the poles of static-file caching. StaticNoCache makes the browser revalidate on every use: always fresh at the cost of a conditional request each time, right for files that change under a stable URL (and for dev, where you want edits visible on reload). StaticImmutable promises the URL’s content will never change, so the browser may cache it forever with no requests at all: only safe when a changed file gets a changed URL, which is what vurl or a bundler’s content-hashed filenames provide.

statd = Path(tempfile.mkdtemp())
(statd/'app.css').write_text('body{}')
stat_app = FastHTML(routes=[Mount('/nc', StaticNoCache(directory=statd)), Mount('/imm', StaticImmutable(directory=statd))])
stat_cli = TestClient(stat_app)
test_eq(stat_cli.get('/nc/app.css').headers['cache-control'], 'no-cache')
test_eq(stat_cli.get('/imm/app.css').headers['cache-control'], 'public, max-age=31536000, immutable')

---

vurl

def vurl(
    path, root:str='.'
):

Return path as a URL with a v query param hashing the file’s contents, so the URL changes when the file does

Browsers use the URL as the cache key, so a URL that changes whenever the file’s contents change can be cached forever: serve the mount with StaticImmutable and the asset costs no requests at all until it is next edited. This is the query-param version of the content-hashed filenames that asset pipelines and bundlers produce, without the build step. The hash is cached via flexicache with mtime_policy, so a call costs one stat and the file is only re-read after it changes. The param is inert under no-cache serving, so the same links work in dev with StaticNoCache, where the file is rechecked on each reload.

tmp = Path(tempfile.mkdtemp())
(tmp/'a.css').write_text('body{}')
u = vurl('a.css', tmp)
assert u.startswith('a.css?v=')
test_eq(u, vurl('a.css', tmp))
test_eq(vurl('/a.css', tmp).split('?')[0], '/a.css')
q = vurl('/a.css?theme=dark#top', tmp)
assert q.startswith('/a.css?theme=dark&v=') and q.endswith('#top')
(tmp/'a.css').write_text('body{color:red}')
os.utime(tmp/'a.css', (time.time()+1,)*2)
assert vurl('a.css', tmp) != u

---

add_sig_param

def add_sig_param(
    f, name, typ:type=NoneType, kind:inspect._ParameterKind=<_ParameterKind.KEYWORD_ONLY: 3>,
    default:fastcore.basics._SentinelEmpty=Parameter.empty
):

Add a parameter to a function’s signature

---

into

def into(
    func
):

Decorator to pass a route’s return value into func, with keyword params added to the route signature

into is a decorator that composes two functions: the decorated route function and an outer “wrapper” function (like a layout). The route’s return value is passed as positional args to the wrapper. Any keyword parameters listed in params are added to the route’s signature, so they become available as query/form parameters. This lets you write simple routes that return just their content, while the layout and its parameters are handled automatically.

Here hello returns a single P tag, but into(my_layout) passes it into my_layout, which wraps it in a Titled. The title parameter is injected into hello’s signature so it can be set via query params.

app,cli,rt = get_cli(FastHTML())

def my_layout(*c, title:str='Default'): return Titled(title, *c)

@rt
@into(my_layout)
def hello(name:str='World'): return P(f'Hello, {name}!')

With no query params, name defaults to 'World' and title defaults to 'Default' (from my_layout’s own default), giving us a page titled “Default” containing “Hello, World!”.

r = cli.get('/hello')
assert '<title>Default</title>' in r.text
assert '<p>Hello, World!</p>' in r.text

Here, both name and title are provided as query params. name='Jeremy' is handled by hello directly, while title='Greetings' is the injected param that gets forwarded to my_layout.

r = cli.get('/hello?name=Jeremy&title=Greetings')
assert '<title>Greetings</title>' in r.text
assert '<p>Hello, Jeremy!</p>' in r.text

---

MiddlewareBase

def MiddlewareBase(
    *args, **kwargs
):

Base for middleware: passes non-http/ws scopes through, else returns the HTTPConnection

---

FtResponse

def FtResponse(
    content, status_code:int=200, headers:NoneType=None, cls:type=HTMLResponse, media_type:str | None=None,
    background:starlette.background.BackgroundTask | None=None
):

Wrap an FT response with any Starlette Response

@rt('/ftr')
def get():
    cts = Title('Foo'),H1('bar')
    return FtResponse(cts, status_code=201, headers={'Location':'/foo/1'})

r = cli.get('/ftr')

test_eq(r.status_code, 201)
test_eq(r.headers['location'], '/foo/1')
txt = r.text
assert '<title>Foo</title>' in txt and '<h1>bar</h1>' in txt and '<html>' in txt

Test on a single background task:

def my_slow_task():
    print('Starting slow task')    
    time.sleep(0.001)
    print('Finished slow task')        

@rt('/background')
def get(): return P('BG Task'), BackgroundTask(my_slow_task)

r = cli.get('/background')

test_eq(r.status_code, 200)

Starting slow task
Finished slow task

Test multiple background tasks:

def increment(amount):
    amount = amount/1000
    print(f'Sleeping for {amount}s')    
    time.sleep(amount)
    print(f'Slept for {amount}s')

@rt
def backgrounds():
    tasks = BackgroundTasks()
    for i in range(3): tasks.add_task(increment, i)
    return P('BG Tasks'), tasks

r = cli.get('/backgrounds')
test_eq(r.status_code, 200)

Sleeping for 0.0s
Slept for 0.0s
Sleeping for 0.001s
Slept for 0.001s
Sleeping for 0.002s
Slept for 0.002s

@rt
def backgrounds2():
    tasks = [BackgroundTask(increment,i) for i in range(3)]
    return P('BG Tasks'), *tasks

r = cli.get('/backgrounds2')
test_eq(r.status_code, 200)

Sleeping for 0.0s
Slept for 0.0s
Sleeping for 0.001s
Slept for 0.001s
Sleeping for 0.002s
Slept for 0.002s

@rt
def backgrounds3():
    tasks = [BackgroundTask(increment,i) for i in range(3)]
    return {'status':'done'}, *tasks

r = cli.get('/backgrounds3')
test_eq(r.status_code, 200)
r.json()

Sleeping for 0.0s
Slept for 0.0s
Sleeping for 0.001s
Slept for 0.001s
Sleeping for 0.002s
Slept for 0.002s

{'status': 'done'}

---

unqid

def unqid(
    seeded:bool=False
):

*Random unique id, base64-encoded and prefixed with ’_’ so it’s usable as an HTML id*

---

FastHTML.setup_ws

def setup_ws(
    app:__main__.FastHTML, f:function=noop
):

setup_ws returns an async function that broadcasts FT elements to connected clients. Here a message handler broadcasts each incoming message. Clients can disconnect and reconnect:

push_app = FastHTML()
async def relay(msg:str): await push(Div(msg, id='status'))
push = push_app.setup_ws(relay)
with TestClient(push_app) as client:
    for msg in ('hello', 'again'):
        with client.websocket_connect('/ws') as ws:
            ws.send_json(dict(msg=msg))
            assert msg in ws.receive_text()

---

FastHTML.devtools_json

def devtools_json(
    path:NoneType=None, uuid:NoneType=None
):

---

FastHTML.get_client

def get_client(
    asink:bool=False, **kw
):

Get an httpx client with session cookes set from **kw

---

FastHTML.decode_session

def decode_session(
    cookie
):

Decode a signed session cookie

---

FastHTML.get_testclient

def get_testclient(
    **kw
):

Get a Starlette TestClient with session cookies set from **kw

cli = app.get_testclient(auth='user1')
test_eq(cli.session['auth'], 'user1')