- Easy: intuitive, clear logic
- Simple: small code base, no external dependencies
- Modular: application mounting, custom route behavior
- Flexible: unopinionated, paradigm-free
- Fast: minimal overhead
- Safe: small attack surface
µHTTP is on PyPI.
pip install uhttpAlso, an ASGI server might be needed.
pip install uvicornfrom uhttp import Application
app = Application()
@app.get('/')
def hello(request):
return f'Hello, {request.ip}!'
if __name__ == '__main__':
import uvicorn
uvicorn.run('__main__:app')An ASGI application. Called once per request by the server.
Application(*, routes=None, startup=None, shutdown=None, before=None, after=None, max_content=1048576)E.g.:
app = Application(
startup=[open_db],
before=[counter, auth],
routes={
'/': {
'GET': lambda request: 'HI!',
'POST': new
},
'/users/': {
'GET': users,
'PUT': users
}
},
after=[logger],
shutdown=[close_db]
)Mounts another application at the specified prefix.
app.mount(another, prefix='')E.g.:
utils = Application()
@utils.before
def incoming(request):
print(f'Incoming from {request.ip}')
app.mount(utils)Append the decorated function to the list of functions called at the beginning of the Lifespan protocol.
@app.startup
[async] def func(state)E.g.:
@app.startup
async def open_db(state):
state['db'] = await aiosqlite.connect('db.sqlite')Appends the decorated function to the list of functions called at the end of the Lifespan protocol.
@app.shutdown
[async] def func(state)E.g.:
@app.shutdown
async def close_db(state):
await state['db'].close()Appends the decorated function to the list of functions called before a response is made.
@app.before
[async] def func(request)E.g.:
@app.before
def restrict(request):
user = request.state['session'].get('user')
if user != 'admin':
raise Response(401)Appends the decorated function to the list of functions called after a response is made.
@app.after
[async] def func(request, response)E.g.:
@app.after
def logger(request, response):
print(request, '-->', response)Inserts the decorated function to the routing table.
@app.route(path, methods=('GET',))
[async] def func(request)Paths are compiled at startup as regular expression patterns. Named groups define path parameters.
If the request path doesn't match any route pattern, a 404 Not Found response is returned.
If the request method isn't in the route methods, a 405 Method Not Allowed response is returned.
Decorators for the standard methods are also available.
E.g.:
@app.route('/', methods=('GET', 'POST'))
def index(request):
return f'{request.method}ing from {request.ip}'
@app.get(r'/user/(?P<id>\d+)')
def profile(request):
user = request.state['db'].get_or_404(request.params['id'])
return f'{user.name} has {user.friends} friends!'An HTTP request. Created every time the application is called on the HTTP protocol with a shallow copy of the state.
Request(method, path, *, ip='', params=None, args=None, headers=None, cookies=None, body=b'', json=None, form=None, state=None)An HTTP Response. May be raised or returned at any time in middleware or route functions.
Response(status, *, headers=None, cookies=None, body=b'')E.g.:
@app.startup
def open_db(state):
state['db'] = {
1: {
'name': 'admin',
'likes': ['terminal', 'old computers']
},
2: {
'name': 'john',
'likes': ['animals']
}
}
def get_or_404(db, id):
if user := db.get(id):
return user
else:
raise Response(404)
@app.get(r'/user/(?P<id>\d+)')
def profile(request):
user = get_or_404(request.state['db'], request.params['id'])
if request.args.get('json'):
return user
else:
return f"{user['name']} likes {', '.join(user['likes'])}"Session implementation based on JavaScript Web Signatures. Sessions are stored in the client's browser as a tamper-proof cookie. Depends on PyJWT.
import os
import time
import jwt
from uhttp import Application, Response
app = Application()
secret = os.getenv('APP_SECRET', 'dev')
@app.before
def get_token(request):
session = request.cookies.get('session')
if session and session.value:
try:
request.state['session'] = jwt.decode(
jwt=session.value,
key=secret,
algorithms=['HS256']
)
except jwt.exceptions.PyJWTError:
request.state['session'] = {'exp': 0}
raise Response(400)
else:
request.state['session'] = {}
@app.after
def set_token(request, response):
if session := request.state.get('session'):
session.setdefault('exp', int(time.time()) + 604800)
response.cookies['session'] = jwt.encode(
payload=session,
key=secret,
algorithm='HS256'
)
response.cookies['session']['expires'] = time.strftime(
'%a, %d %b %Y %T GMT', time.gmtime(session['exp'])
)
response.cookies['session']['samesite'] = 'Lax'
response.cookies['session']['httponly'] = True
response.cookies['session']['secure'] = TrueSupport for multipart forms. Depends on python-multipart.
from multipart.multipart import FormParser, parse_options_header
from multipart.exceptions import FormParserError
from uhttp import Application, MultiDict, Response
app = Application()
def parse_form(request):
form = MultiDict()
def on_field(field):
form[field.field_name.decode()] = field.value.decode()
def on_file(file):
if file.field_name:
form[file.field_name.decode()] = file.file_object
content_type, options = parse_options_header(
request.headers.get('content-type', '')
)
try:
parser = FormParser(
content_type.decode(),
on_field,
on_file,
boundary=options.get(b'boundary'),
config={'MAX_MEMORY_FILE_SIZE': float('inf')} # app._max_content
)
parser.write(request.body)
parser.finalize()
except FormParserError:
raise Response(400)
return form
@app.before
def handle_multipart(request):
if 'multipart/form-data' in request.headers.get('content-type'):
request.form = parse_form(request)Static files for development.
import os
from mimetypes import guess_type
from uhttp import Application, Response
app = Application()
def send_file(path):
if not os.path.isfile(path):
raise RuntimeError('Invalid file')
mime_type = guess_type(path)[0] or 'application/octet-stream'
with open(path, 'rb') as file:
content = file.read()
return Response(
status=200,
headers={'content-type': mime_type},
body=content
)
@app.get('/assets/(?P<path>.*)')
def assets(request):
directory = 'assets'
path = os.path.realpath(
os.path.join(directory, request.params['path'])
)
if os.path.commonpath([directory, path]) == directory:
if os.path.isfile(path):
return send_file(path)
if os.path.isdir(path):
index = os.path.join(path, 'index.html')
if os.path.isfile(index):
return send_file(index)
return 404All contributions are welcomed.
Released under the MIT license.
