patx/veela
Simplify example app with mounted applications
Commit 3118562 · patx · 2026-10-14T20:46:31-04:00
Comments
No comments yet.
Diff
diff --git a/example_app/app.py b/example_app/app.py
index 32e7440..e7412d2 100644
--- a/example_app/app.py
+++ b/example_app/app.py
@@ -7,14 +7,11 @@ Login is restored with an HttpOnly refresh cookie; API calls use bearer tokens.
Storage and rate limits are process-local. Production needs shared storage and HTTPS.
"""
-from veela import App
-
from . import auth, notes, pages
-app = App()
-pages.register_routes(app)
-auth.register_routes(app)
-notes.register_routes(app)
+app = pages.app
+app.mount('/api', auth.app)
+app.mount('/api/notes', notes.app)
if __name__ == '__main__':
app.run()
diff --git a/example_app/auth.py b/example_app/auth.py
index 1625d40..edfaba0 100644
--- a/example_app/auth.py
+++ b/example_app/auth.py
@@ -7,10 +7,12 @@ from http.cookies import CookieError, SimpleCookie
from secrets import compare_digest, token_bytes, token_urlsafe
from time import monotonic
-from veela import VeelaRequestError
+from veela import App, VeelaRequestError
from .rate_limit import rate_limit
from .storage import users, tokens, refresh_tokens
+app = App()
+
token_lifetime = 3600
refresh_lifetime = 7 * 24 * 3600
refresh_cookie_name = 'notebook_refresh'
@@ -133,88 +135,89 @@ def protected(handler):
return wrapper
-def register_routes(app):
- """Register authentication endpoints on the supplied Veela app."""
-
- @app.route('/api/signup', methods=['POST'])
- @rate_limit(requests=10, seconds=60)
- async def signup(request):
- username, password = await read_credentials(request)
- salt = token_bytes(16)
- password_hash = await hash_password(password, salt)
- # Check after the await so simultaneous signups cannot overwrite the same account.
- if username in users:
- return {'error': 'Username already exists'}, 'application/json', 409
- users[username] = {'salt': salt, 'password_hash': password_hash}
- return {'id': username}, 'application/json', 201
-
- @app.route('/api/login', methods=['POST'])
- @rate_limit(requests=5, seconds=60)
- async def login(request):
- if not same_origin_request(request):
- return (
- {'error': 'A same-origin request with X-Notebook-Request: 1 is required'},
- 'application/json',
- 403,
- )
- username, password = await read_credentials(request)
- # Use a dummy record for unknown users so failed logins still perform password hashing.
- record = users.get(username, {'salt': bytes(16), 'password_hash': bytes(32)})
- password_hash = await hash_password(password, record['salt'])
- if username not in users or not compare_digest(password_hash, record['password_hash']):
- return {'error': 'Invalid credentials'}, 'application/json', 401
- login_id = token_urlsafe(16)
- refresh = token_urlsafe(32)
- # Store only its digest and a fixed absolute expiry; refreshes do not extend this deadline.
- refresh_tokens[sha256(refresh.encode()).hexdigest()] = {
- 'user_id': username,
- 'login_id': login_id,
- 'expires_at': monotonic() + refresh_lifetime,
- }
[email protected]('/signup', methods=['POST'])
+@rate_limit(requests=10, seconds=60)
+async def signup(request):
+ username, password = await read_credentials(request)
+ salt = token_bytes(16)
+ password_hash = await hash_password(password, salt)
+ # Check after the await so simultaneous signups cannot overwrite the same account.
+ if username in users:
+ return {'error': 'Username already exists'}, 'application/json', 409
+ users[username] = {'salt': salt, 'password_hash': password_hash}
+ return {'id': username}, 'application/json', 201
+
+
[email protected]('/login', methods=['POST'])
+@rate_limit(requests=5, seconds=60)
+async def login(request):
+ if not same_origin_request(request):
return (
- issue_access_token(username, login_id),
+ {'error': 'A same-origin request with X-Notebook-Request: 1 is required'},
'application/json',
- 200,
- [('Cache-Control', 'no-store'), refresh_cookie(request, refresh, refresh_lifetime)],
+ 403,
)
-
- @app.route('/api/refresh', methods=['POST'])
- @rate_limit(requests=30, seconds=60)
- async def refresh(request):
- if not same_origin_request(request):
- return (
- {'error': 'A same-origin request with X-Notebook-Request: 1 is required'},
- 'application/json',
- 403,
- )
- digest = sha256(read_refresh_cookie(request).encode()).hexdigest()
- record = refresh_tokens.get(digest)
- if record is None or record['expires_at'] <= monotonic():
- if record is not None:
- revoke_login(record['login_id'])
- return (
- {'error': 'Unauthorized'},
- 'application/json',
- 401,
- [('Cache-Control', 'no-store'), refresh_cookie(request, '', 0)],
- )
- # Reuse the cookie while issuing a fresh memory-only access token; concurrent tabs work.
+ username, password = await read_credentials(request)
+ # Use a dummy record for unknown users so failed logins still perform password hashing.
+ record = users.get(username, {'salt': bytes(16), 'password_hash': bytes(32)})
+ password_hash = await hash_password(password, record['salt'])
+ if username not in users or not compare_digest(password_hash, record['password_hash']):
+ return {'error': 'Invalid credentials'}, 'application/json', 401
+ login_id = token_urlsafe(16)
+ refresh = token_urlsafe(32)
+ # Store only its digest and a fixed absolute expiry; refreshes do not extend this deadline.
+ refresh_tokens[sha256(refresh.encode()).hexdigest()] = {
+ 'user_id': username,
+ 'login_id': login_id,
+ 'expires_at': monotonic() + refresh_lifetime,
+ }
+ return (
+ issue_access_token(username, login_id),
+ 'application/json',
+ 200,
+ [('Cache-Control', 'no-store'), refresh_cookie(request, refresh, refresh_lifetime)],
+ )
+
+
[email protected]('/refresh', methods=['POST'])
+@rate_limit(requests=30, seconds=60)
+async def refresh(request):
+ if not same_origin_request(request):
return (
- issue_access_token(record['user_id'], record['login_id']),
+ {'error': 'A same-origin request with X-Notebook-Request: 1 is required'},
'application/json',
- 200,
- [('Cache-Control', 'no-store')],
+ 403,
)
-
- @app.route('/api/me')
- @protected
- async def me(request):
- return request.user
-
- @app.route('/api/logout', methods=['POST'])
- @protected
- async def logout(request):
- digest = sha256(request.access_token.encode('utf-8')).hexdigest()
- login_id = tokens[digest]['login_id']
- revoke_login(login_id)
- return {'logged_out': True}, 'application/json', 200, [refresh_cookie(request, '', 0)]
+ digest = sha256(read_refresh_cookie(request).encode()).hexdigest()
+ record = refresh_tokens.get(digest)
+ if record is None or record['expires_at'] <= monotonic():
+ if record is not None:
+ revoke_login(record['login_id'])
+ return (
+ {'error': 'Unauthorized'},
+ 'application/json',
+ 401,
+ [('Cache-Control', 'no-store'), refresh_cookie(request, '', 0)],
+ )
+ # Reuse the cookie while issuing a fresh memory-only access token; concurrent tabs work.
+ return (
+ issue_access_token(record['user_id'], record['login_id']),
+ 'application/json',
+ 200,
+ [('Cache-Control', 'no-store')],
+ )
+
+
[email protected]('/me')
+@protected
+async def me(request):
+ return request.user
+
+
[email protected]('/logout', methods=['POST'])
+@protected
+async def logout(request):
+ digest = sha256(request.access_token.encode('utf-8')).hexdigest()
+ login_id = tokens[digest]['login_id']
+ revoke_login(login_id)
+ return {'logged_out': True}, 'application/json', 200, [refresh_cookie(request, '', 0)]
diff --git a/example_app/notes.py b/example_app/notes.py
index b6b01a7..61e15d0 100644
--- a/example_app/notes.py
+++ b/example_app/notes.py
@@ -3,10 +3,12 @@
from datetime import UTC, datetime
from secrets import token_urlsafe
-from veela import VeelaRequestError
+from veela import App, VeelaRequestError
from .auth import protected
from .storage import notes
+app = App()
+
async def read_note(request, partial=False):
data = await request.json()
@@ -36,92 +38,94 @@ def owned_note(note_id, user):
return note
-def register_routes(app):
- """Register note endpoints; each handler requires a valid bearer token."""
-
- @app.route('/api/notes', methods=['GET'])
- @protected
- async def list_notes(request):
- # Keep this demo's query budget at 20 fields, including repeated keys.
- if request.scope.get('query_string', b'').count(b'&') >= 20:
- raise VeelaRequestError('Invalid query string')
- query = request.query()
- try:
- limit = int(query.get('limit') or '5')
- offset = int(query.get('offset') or '0')
- except ValueError:
- raise VeelaRequestError('Limit and offset must be integers')
- if not 1 <= limit <= 50 or offset < 0:
- raise VeelaRequestError('Limit must be 1–50; offset must be zero or greater')
- search = query.get('q', '').strip().casefold()
- pinned = query.get('pinned', '')
- if pinned not in ('', 'true', 'false'):
- raise VeelaRequestError('Pinned filter must be true or false')
- # Apply ownership before searching or counting so other users' data never leaks.
- items = [note for note in notes.values() if note['owner'] == request.user['id']]
- items = [
- note for note in items if search in (note['title'] + '\n' + note['body']).casefold()
- ]
- if pinned:
- items = [note for note in items if note['pinned'] == (pinned == 'true')]
- # Show pinned notes first, then most recently edited notes within each group.
- items.sort(key=lambda note: (note['pinned'], note['updated_at'], note['id']), reverse=True)
- return {
- 'items': items[offset : offset + limit],
- 'total': len(items),
- 'limit': limit,
- 'offset': offset,
- }
-
- @app.route('/api/notes', methods=['POST'])
- @protected
- async def create_note(request):
- data = await read_note(request)
- note_id = token_urlsafe(9)
- now = datetime.now(UTC).isoformat(timespec='microseconds')
- note = {
- 'id': note_id,
- 'owner': request.user['id'],
- **data,
- 'created_at': now,
- 'updated_at': now,
- }
- notes[note_id] = note
- return note, 'application/json', 201, [('Location', f'/api/notes/{note_id}')]
-
- @app.route('/api/notes/{note_id}', methods=['GET'])
- @protected
- async def get_note(request, note_id):
- return owned_note(note_id, request.user)
-
- @app.route('/api/notes/{note_id}', methods=['PUT', 'PATCH'])
- @protected
- async def update_note(request, note_id):
- note = owned_note(note_id, request.user)
- if isinstance(note, tuple):
- return note
- data = await read_note(request, partial=request.method == 'PATCH')
- note.update(data)
- note['updated_at'] = datetime.now(UTC).isoformat(timespec='microseconds')
[email protected]('/', methods=['GET'])
+@protected
+async def list_notes(request):
+ # Keep this demo's query budget at 20 fields, including repeated keys.
+ if request.scope.get('query_string', b'').count(b'&') >= 20:
+ raise VeelaRequestError('Invalid query string')
+ query = request.query()
+ try:
+ limit = int(query.get('limit') or '5')
+ offset = int(query.get('offset') or '0')
+ except ValueError:
+ raise VeelaRequestError('Limit and offset must be integers')
+ if not 1 <= limit <= 50 or offset < 0:
+ raise VeelaRequestError('Limit must be 1–50; offset must be zero or greater')
+ search = query.get('q', '').strip().casefold()
+ pinned = query.get('pinned', '')
+ if pinned not in ('', 'true', 'false'):
+ raise VeelaRequestError('Pinned filter must be true or false')
+ # Apply ownership before searching or counting so other users' data never leaks.
+ items = [note for note in notes.values() if note['owner'] == request.user['id']]
+ items = [
+ note for note in items if search in (note['title'] + '\n' + note['body']).casefold()
+ ]
+ if pinned:
+ items = [note for note in items if note['pinned'] == (pinned == 'true')]
+ # Show pinned notes first, then most recently edited notes within each group.
+ items.sort(key=lambda note: (note['pinned'], note['updated_at'], note['id']), reverse=True)
+ return {
+ 'items': items[offset : offset + limit],
+ 'total': len(items),
+ 'limit': limit,
+ 'offset': offset,
+ }
+
+
[email protected]('/', methods=['POST'])
+@protected
+async def create_note(request):
+ data = await read_note(request)
+ note_id = token_urlsafe(9)
+ now = datetime.now(UTC).isoformat(timespec='microseconds')
+ note = {
+ 'id': note_id,
+ 'owner': request.user['id'],
+ **data,
+ 'created_at': now,
+ 'updated_at': now,
+ }
+ notes[note_id] = note
+ return note, 'application/json', 201, [('Location', f'/api/notes/{note_id}')]
+
+
[email protected]('/{note_id}', methods=['GET'])
+@protected
+async def get_note(request, note_id):
+ return owned_note(note_id, request.user)
+
+
[email protected]('/{note_id}', methods=['PUT', 'PATCH'])
+@protected
+async def update_note(request, note_id):
+ note = owned_note(note_id, request.user)
+ if isinstance(note, tuple):
+ return note
+ data = await read_note(request, partial=request.method == 'PATCH')
+ note.update(data)
+ note['updated_at'] = datetime.now(UTC).isoformat(timespec='microseconds')
+ return note
+
+
[email protected]('/{note_id}', methods=['DELETE'])
+@protected
+async def delete_note(request, note_id):
+ note = owned_note(note_id, request.user)
+ if isinstance(note, tuple):
return note
+ del notes[note_id]
+ return b'', 'application/json', 204
+
- @app.route('/api/notes/{note_id}', methods=['DELETE'])
- @protected
- async def delete_note(request, note_id):
- note = owned_note(note_id, request.user)
- if isinstance(note, tuple):
- return note
- del notes[note_id]
- return b'', 'application/json', 204
-
- # This fixed path wins over /api/notes/{note_id}, even though it is registered afterward.
- @app.route('/api/notes/export', methods=['GET'])
- @protected
- async def export_notes(request):
- items = [note for note in notes.values() if note['owner'] == request.user['id']]
- return (
- items,
- 'application/json',
- 200,
- [('Content-Disposition', 'attachment; filename="notes.json"')],
- )
+# This fixed path wins over /{note_id}, even though it is registered afterward.
[email protected]('/export', methods=['GET'])
+@protected
+async def export_notes(request):
+ items = [note for note in notes.values() if note['owner'] == request.user['id']]
+ return (
+ items,
+ 'application/json',
+ 200,
+ [('Content-Disposition', 'attachment; filename="notes.json"')],
+ )
diff --git a/example_app/pages.py b/example_app/pages.py
index 128b826..6a6d973 100644
--- a/example_app/pages.py
+++ b/example_app/pages.py
@@ -2,33 +2,37 @@
from pathlib import Path
+from veela import App
+
from .rate_limit import rate_limit
+app = App()
static_root = Path(__file__).parent / 'static'
-def register_routes(app):
- """Register the browser UI and public demonstration endpoints."""
[email protected]('/')
+async def home(request):
+ body, content_type, status = await app.static('index.html', root=static_root)
+ return body, content_type, status, [('Cache-Control', 'no-cache')]
+
+
[email protected]('/static/{filename}')
+async def assets(request, filename):
+ body, content_type, status = await app.static(filename, root=static_root)
+ return body, content_type, status, [('Cache-Control', 'no-cache')]
+
- @app.route('/')
- async def home(request):
- body, content_type, status = await app.static('index.html', root=static_root)
- return body, content_type, status, [('Cache-Control', 'no-cache')]
[email protected]('/ui')
+async def old_ui(request):
+ return b'', 'text/plain', 303, [('Location', '/')]
- @app.route('/static/{filename}')
- async def assets(request, filename):
- body, content_type, status = await app.static(filename, root=static_root)
- return body, content_type, status, [('Cache-Control', 'no-cache')]
- @app.route('/ui')
- async def old_ui(request):
- return b'', 'text/plain', 303, [('Location', '/')]
[email protected]('/api/health')
+async def health(request):
+ return {'status': 'ok'}
- @app.route('/api/health')
- async def health(request):
- return {'status': 'ok'}
- @app.route('/api/rate-limit-demo')
- @rate_limit(requests=3, seconds=10)
- async def rate_limit_demo(request):
- return {'message': 'Request accepted'}
[email protected]('/api/rate-limit-demo')
+@rate_limit(requests=3, seconds=10)
+async def rate_limit_demo(request):
+ return {'message': 'Request accepted'}