import time
from IPython import display
from enum import Enum
from pprint import pprint
from fastcore.test import *
from starlette.testclient import TestClient
from starlette.requests import Headers
FastHTML tests
is_typeddict
is_typeddict (cls:type)
Check if cls
is a TypedDict
class MyDict(TypedDict): name:str
assert is_typeddict(MyDict)
assert not is_typeddict({'a':1})
is_namedtuple
is_namedtuple (cls)
True
is cls
is a namedtuple type
assert is_namedtuple(namedtuple('tst', ['a']))
assert not is_namedtuple(tuple)
date
date (s:str)
Convert s
to a datetime
'2pm') date(
datetime.datetime(2024, 6, 22, 14, 0)
snake2hyphens
snake2hyphens (s:str)
Convert s
from snake case to hyphenated and capitalised
"snake_case") snake2hyphens(
'Snake-Case'
HtmxHeaders
HtmxHeaders (boosted:str|None=None, current_url:str|None=None, history_restore_request:str|None=None, prompt:str|None=None, request:str|None=None, target:str|None=None, trigger_name:str|None=None, trigger:str|None=None)
def test_request(url: str='/', headers: dict={}, method: str='get') -> Request:
= {
scope '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),
}= lambda: {"body": b"", "more_body": False}
receive return Request(scope, receive)
= test_request(headers=Headers({'HX-Request':'1'}))
h _get_htmx(h)
HtmxHeaders(boosted=None, current_url=None, history_restore_request=None, prompt=None, request='1', target=None, trigger_name=None, trigger=None)
str2int
str2int (s)
Convert s
to an int
'1'),str2int('none') str2int(
(1, 0)
def _mk_list(t, v): return [t(o) for o in v]
str,None]),_fix_anno(float) _fix_anno(Union[
(str, float)
int)('1') _fix_anno(
1
list[int])(['1','2']) _fix_anno(
[1, 2]
list[int])('1') _fix_anno(
[1]
= dict(k=int, l=List[int])
d 'k', "1", d) _form_arg(
1
'l', "1", d) _form_arg(
[1]
'l', ["1","2"], d) _form_arg(
[1, 2]
HttpHeader
HttpHeader (k:str, v:str)
form2dict
form2dict (form:starlette.datastructures.FormData)
Convert starlette form data to a dict
= [('a',1),('a',2),('b',0)]
d = FormData(d)
fd = form2dict(fd)
res 'a'], [1,2])
test_eq(res['b'], 0) test_eq(res[
async def f(req):
def _f(p:HttpHeader): ...
= first(signature(_f).parameters.values())
p = await _from_body(req, p)
result return JSONResponse(result.__dict__)
= Starlette(routes=[Route('/', f, methods=['POST'])])
app = TestClient(app)
client
= dict(k='value1',v=['value2','value3'])
d = client.post('/', data=d)
response print(response.json())
{'k': 'value1', 'v': "['value2', 'value3']"}
def g(req, this:Starlette, a:str, b:HttpHeader): ...
async def f(req):
= await _wrap_req(req, signature(g).parameters)
a return Response(str(a))
= Starlette(routes=[Route('/', f, methods=['POST'])])
app = TestClient(app)
client
= client.post('/?a=1', data=d)
response print(response.text)
[<starlette.requests.Request object>, <starlette.applications.Starlette object>, '1', HttpHeader(k='value1', v="['value2', 'value3']")]
flat_xt
flat_xt (lst)
Flatten lists, except for XT
s
= XT('a',1)
x flat_xt([x, x, [x,x]])
[['a', 1, {}], ['a', 1, {}], ['a', 1, {}], ['a', 1, {}]]
Beforeware
Beforeware (f, skip=None)
Initialize self. See help(type(self)) for accurate signature.
WS_RouteX
WS_RouteX (path:str, endpoint, name=None, middleware=None, hdrs=None, before=None, **bodykw)
Initialize self. See help(type(self)) for accurate signature.
RouteX
RouteX (path:str, endpoint, methods=None, name=None, include_in_schema=True, middleware=None, hdrs=None, before=None, **bodykw)
Initialize self. See help(type(self)) for accurate signature.
RouterX
RouterX (routes=None, redirect_slashes=True, default=None, on_startup=None, on_shutdown=None, lifespan=None, middleware=None, hdrs=None, before=None, **bodykw)
Initialize self. See help(type(self)) for accurate signature.
get_key
get_key (key=None, fname='.sesskey')
get_key()
'08b63b51-be3a-4f54-8d26-4cd27eb17c0d'
def _list(o): return [] if not o else o if isinstance(o, (tuple,list)) else [o]
FastHTML
FastHTML (debug=False, routes=None, middleware=None, exception_handlers=None, on_startup=None, on_shutdown=None, lifespan=None, hdrs=None, before=None, default_hdrs=True, secret_key=None, session_cookie='session_', max_age=31536000, sess_path='/', same_site='lax', sess_https_only=False, sess_domain=None, key_fname='.sesskey', **bodykw)
*Creates an application instance.
Parameters:
- debug - Boolean indicating if debug tracebacks should be returned on errors.
- routes - A list of routes to serve incoming HTTP and WebSocket requests.
- middleware - A list of middleware to run for every request. A starlette application will always automatically include two middleware classes.
ServerErrorMiddleware
is added as the very outermost middleware, to handle any uncaught errors occurring anywhere in the entire stack.ExceptionMiddleware
is added as the very innermost middleware, to deal with handled exception cases occurring in the routing or endpoints. - exception_handlers - A mapping of either integer status codes, or exception class types onto callables which handle the exceptions. Exception handler callables should be of the form
handler(request, exc) -> response
and may be either standard functions, or async functions. - on_startup - A list of callables to run on application startup. Startup handler callables do not take any arguments, and may be either standard functions, or async functions.
- on_shutdown - A list of callables to run on application shutdown. Shutdown handler callables do not take any arguments, and may be either standard functions, or async functions.
- lifespan - A lifespan context function, which can be used to perform startup and shutdown tasks. This is a newer style that replaces the
on_startup
andon_shutdown
handlers. Use one or the other, not both.*
reg_re_param
reg_re_param (m, s)
MiddlewareBase
MiddlewareBase ()
Initialize self. See help(type(self)) for accurate signature.
def get_cli(app): return app,TestClient(app),app.route
= get_cli(FastHTML(secret_key='soopersecret')) app,cli,rt
@rt("/hi")
def get(): return 'Hi there'
= cli.get('/hi')
r r.text
'Hi there'
@rt("/hi")
def post(): return 'Postal'
'/hi').text cli.post(
'Postal'
@app.get("/")
def show_host(req): return req.headers['host']
'/').text cli.get(
'testserver'
@rt('/user/{nm}', name='gday')
def get(nm:str): return f"Good day to you, {nm}!"
'/user/Alexis').text cli.get(
'Good day to you, Alexis!'
'gday', nm='Jeremy'), '/user/Jeremy') test_eq(app.router.url_path_for(
= {'headers':{'hx-request':"1"}}
hxhdr
@rt('/xt')
def get(): return Title('Foo'),H1('bar')
= cli.get('/xt').text
txt assert '<title>Foo</title>' in txt and '<h1>bar</h1>' in txt and '<html>' in txt
@rt('/xt2')
def get(): return H1('bar')
= cli.get('/xt2').text
txt 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')))
= cli.get('/xt3').text
txt assert '<title>FastHTML page</title>' not in txt and '<title>hi</title>' in txt and '<p>there</p>' in txt
def test_r(cli, path, exp, meth='get', hx=False, **kwargs):
if hx: kwargs['headers'] = {'hx-request':"1"}
getattr(cli, meth)(path, **kwargs).text, exp)
test_eq(
= 'foo'
app.chk = str_enum('ModelName', "alexnet", "resnet", "lenet")
ModelName = [{"name": "Foo"}, {"name": "Bar"}] fake_db
@rt('/html/{idx}')
async def get(idx:int): return Body(H4(f'Next is {idx+1}.'))
"imgext", "ico|gif|jpg|jpeg|webm")
reg_re_param(
@rt(r'/static/{path:path}{fn}.{ext:imgext}')
def get(fn:str, path:str, ext:str): return f"Getting {fn}.{ext} from /{path}"
@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]
'/html/1', '<body>\n <h4>Next is 2.</h4>\n</body>\n', hx=True)
test_r(cli, '/static/foo/jph.ico', 'Getting jph.ico from /foo/')
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"}') test_r(cli,
@app.get("/booly/")
def _(coming:bool=True): return 'Coming' if coming else 'Not coming'
@app.get("/datie/")
def _(d: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")
'/booly/?coming=true', 'Coming')
test_r(cli, '/booly/?coming=no', 'Not coming')
test_r(cli, = "17th of May, 2024, 2p"
date_str 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') test_r(cli,
= cli.get('/app2', **hxhdr)
r 'foo\n')
test_eq(r.text, 'mykey'], 'myval') test_eq(r.headers[
@dataclass
class Bodie: a:int;b:str
@rt("/bodie/{nm}/")
async def post(nm:str, data:Bodie):
= asdict(data)
res 'nm'] = nm
res[return res
@app.post("/bodied/")
async def bodied(nm:str, data:dict): return data
= namedtuple('Bodient', ['a','b'])
nt
@app.post("/bodient/")
async def bodient(data:nt): return data._asdict()
class BodieTD(TypedDict): a:int;b:str='foo'
@app.post("/bodietd/")
async def bodient(data:BodieTD): return data
class Bodie2:
int|None; b:str
a:def __init__(self, a, b='foo'): store_attr()
@app.post("/bodie2/")
async def bodie(d:Bodie2): return f"a: {d.a}; b: {d.b}"
= dict(a=1, b='foo')
d
'/bodie/me', '{"a":1,"b":"foo","nm":"me"}', 'post', data=d)
test_r(cli, '/bodied/', '{"a":"1","b":"foo"}', 'post', data=d)
test_r(cli, '/bodie2/', 'a: 1; b: foo', 'post', data={'a':1})
test_r(cli, '/bodient/', '{"a":"1","b":"foo"}', 'post', data=d)
test_r(cli, '/bodietd/', '{"a":1,"b":"foo"}', 'post', data=d) test_r(cli,
@rt("/setcookie")
async def get(req):
= datetime.now()
now = Response(f'Set to {now}')
res 'now', str(now))
res.set_cookie(return res
@rt("/getcookie")
async def get(now:date): return f'Cookie was set at time {now.time()}'
print(cli.get('/setcookie').text)
0.01)
time.sleep('/getcookie').text cli.get(
Set to 2024-06-15 09:33:00.521810
'Cookie was set at time 09:33:00.521810'
@rt("/setsess")
async def get(sess):
= datetime.now()
now 'noo'] = str(now)
sess[return f'Set to {now}'
@rt("/getsess")
async def get(noo:date): return f'Session time: {noo.time()}'
print(cli.get('/setsess').text)
0.01)
time.sleep('/getsess').text cli.get(
Set to 2024-06-15 09:33:01.830672
'Session time: 09:33:01.830672'
@rt("/upload")
async def post(uploadfile:str): return (await uploadfile.read()).decode()
= '../CHANGELOG.md'
fn = {'message': 'Hello, world!'}
data with open(fn, 'rb') as f:
print(cli.post('/upload', files={'uploadfile': f}, data=data).text[:80])
# Release notes
<!-- do not remove -->
## 0.0.13
### New Features
- Add `htm
= user_pwd_auth(testuser='spycraft')
auth = get_cli(FastHTML(middleware=[auth]))
app,cli,rt
@rt("/locked")
def get(auth): return 'Hello, ' + auth
'/locked').text, 'not authenticated')
test_eq(cli.get('/locked', auth=("testuser","spycraft")).text, 'Hello, testuser') test_eq(cli.get(
= app.router.hdrs, app.routes hdrs, routes
= get_cli(FastHTMLWithLiveReload())
app,cli,rt
@rt("/hi")
def get(): return 'Hi there'
'/hi').text, "Hi there")
test_eq(cli.get(
= app.router.hdrs, app.routes
lr_hdrs, lr_routes len(lr_hdrs), len(hdrs)+1)
test_eq(assert app.LIVE_RELOAD_HEADER in lr_hdrs
len(lr_routes), len(routes)+1)
test_eq(assert app.LIVE_RELOAD_ROUTE in lr_routes