"""
Veela: A magical, single-file ASGI API framework and web server.
Veela is designed for lightweight JSON APIs, offering a minimalistic way to build
ASGI-compatible web services. It prioritizes simplicity and performance over all.
Based on Wred: https://harrisonerd.com/wred
Veela website: https://harrisonerd.com/veela
Copyright (c) 2026, Harrison Erd. For license information see `LICENSE` AND
`THIRD_PARTY_LICENSES.txt`. Bundles `uvicorn`, `click`, and `h11`.
"""
__author__ = 'Harrison Erd'
__email__ = '[email protected]'
__version__ = '0.3.1'
__license__ = 'BSD-3-Clause'
from collections.abc import Awaitable, Callable
from asyncio import to_thread
from mimetypes import guess_type
from pathlib import Path
from typing import Any
from urllib.parse import parse_qsl
import json
try:
import orjson
except ImportError:
orjson = None
_loading_worker_app = False
def _load_worker_app(module, name):
"""Load an app without starting another server during its import."""
from importlib import import_module
global _loading_worker_app
_loading_worker_app = True
try:
return getattr(import_module(module), name)
except Exception as error:
import logging
from uvicorn.config import STARTUP_FAILURE
logging.getLogger('uvicorn.error').error('Cannot load application %s:%s: %s', module, name, error)
raise SystemExit(STARTUP_FAILURE) from None
finally:
_loading_worker_app = False
class VeelaResponseError(Exception):
"""Raised when a handler returns a non-JSON response."""
pass
class VeelaRequestError(Exception):
"""Raised when a request cannot be processed (e.g., invalid body)."""
pass
class Request:
"""A request object encapsulating ASGI scope, receive, and send."""
def __init__(self, scope: dict[str, Any], receive: Callable[[], Awaitable[dict[str, Any]]], send: Callable[[dict[str, Any]], Awaitable[None]]):
self.scope = scope
self.receive = receive
self.send = send
self.path = scope['path']
self.method = scope['method']
self.headers = {k.decode(): v.decode() for k, v in scope.get('headers', [])}
self._body = None
self._query = None
def query(self) -> dict[str, str]:
"""Parse and cache query parameters; repeated keys keep their last value."""
if self._query is None:
try:
self._query = dict(parse_qsl(self.scope.get('query_string', b'').decode('utf-8'), keep_blank_values=True))
except UnicodeDecodeError:
raise VeelaRequestError("Invalid query string")
return self._query
async def body(self) -> bytes:
"""Read the request body from the ASGI receive stream."""
if self._body is None:
self._body = b""
while True:
message = await self.receive()
if message['type'] != 'http.request':
raise VeelaRequestError("Client disconnected")
self._body += message.get('body', b'')
if not message.get('more_body', False):
break
return self._body
async def json(self) -> dict[str, Any]:
"""Parse the request body as JSON."""
try:
body = await self.body()
return (orjson.loads(body) if orjson else json.loads(body)) if body else {}
except (json.JSONDecodeError, UnicodeDecodeError):
raise VeelaRequestError("Invalid JSON body")
class App:
"""A minimal ASGI application framework."""
def __init__(self):
self.routes: dict[str, dict[str, Callable]] = {}
self.pattern_routes: dict[str, dict[str, Callable]] = {}
self.mounts: dict[str, Callable] = {}
def run(self, host="127.0.0.1", port=8000, workers=1):
"""Serve HTTP until interrupted. Workers need a module-level app, no main guard."""
import base64
import multiprocessing
import signal
import sys
import tempfile
import threading
from functools import partial
if _loading_worker_app or getattr(multiprocessing.current_process(), '_inheriting', False):
return
if type(workers) is not int or workers < 1:
raise ValueError('workers must be a positive integer')
application = self
if workers > 1:
if threading.current_thread() is not threading.main_thread():
raise RuntimeError('Multiple workers must be started from the main thread')
namespace = sys._getframe(1).f_globals
name = next((name for name, value in namespace.items() if value is self), None)
if name is None:
raise ValueError('Multiple workers require a module-level application')
module = namespace['__name__']
if module == '__main__':
spec = namespace.get('__spec__')
if spec is not None and spec.name.endswith('.__main__'):
module = spec.name
elif not Path(namespace.get('__file__', '')).is_file():
raise ValueError('Multiple workers require a Python script or importable module')
application = partial(_load_worker_app, module, name)
if any(name in sys.modules for name in ('uvicorn', 'h11', 'click')):
raise RuntimeError('App.run() requires uvicorn, h11, and click not to be imported already')
pending_signals = []
handlers = {}
if threading.current_thread() is threading.main_thread():
for sig in (signal.SIGINT, signal.SIGTERM):
handlers[sig] = signal.signal(sig, lambda sig, frame: pending_signals.append(sig))
try:
with tempfile.TemporaryDirectory(prefix='veela-') as directory:
archive = str(Path(directory) / 'server.zip')
Path(archive).write_bytes(base64.b64decode(_EMBEDDED_ZIP))
sys.path.insert(0, archive)
try:
import uvicorn
if pending_signals:
return
if workers > 1:
from uvicorn.supervisors.multiprocess import Multiprocess, SIGNALS
for sig in SIGNALS:
handlers.setdefault(sig, signal.getsignal(sig))
config = uvicorn.Config(application, host=host, port=port, http="h11", ws="none",
lifespan="off", loop="auto", interface="asgi3",
workers=workers, access_log=False, factory=True)
with config.bind_socket() as sock:
supervisor = Multiprocess(config, sockets=[sock])
supervisor.signal_queue.extend(pending_signals)
try:
supervisor.run()
except BaseException:
supervisor.terminate_all()
supervisor.join_all()
raise
else:
uvicorn.run(application, host=host, port=port, http="h11", ws="none",
lifespan="off", loop="auto", interface="asgi3",
workers=1, access_log=False)
finally:
sys.path.remove(archive)
for name, module in list(sys.modules.items()):
if (getattr(module, '__file__', None) or '').startswith((archive + '/', archive + '\\')):
del sys.modules[name]
for path in list(sys.path_importer_cache):
if path == archive or path.startswith((archive + '/', archive + '\\')):
del sys.path_importer_cache[path]
except KeyboardInterrupt:
pass
finally:
for sig, handler in handlers.items():
signal.signal(sig, handler)
async def __call__(self, scope: dict[str, Any], receive: Callable[[], Awaitable[dict[str, Any]]], send: Callable[[dict[str, Any]], Awaitable[None]]) -> None:
"""Handle ASGI requests."""
if scope['type'] != 'http':
raise RuntimeError(f"Unsupported ASGI scope type: {scope['type']}")
request = Request(scope, receive, send)
await self.handle_http(request)
async def handle_http(self, request: Request) -> None:
"""Handle HTTP requests with routing and JSON response support."""
path = request.path
root_path = request.scope.get('root_path', '').rstrip('/')
if root_path and (path == root_path or path.startswith(root_path + '/')):
path = path[len(root_path):] or '/'
method = request.method
try:
# Check exact matches first
if path in self.routes and method in self.routes[path]:
handler = self.routes[path][method]
response = await handler(request)
await self.dispatch_response(request, response)
return
# Check dynamic routes
for pattern, method_handlers in self.pattern_routes.items():
params = self.match_path(pattern, path)
if params is not None and method in method_handlers:
handler = method_handlers[method]
response = await handler(request, **params)
await self.dispatch_response(request, response)
return
except (VeelaResponseError, VeelaRequestError) as e:
status = 400 if isinstance(e, VeelaRequestError) else 500
await self.send_response(request.send, {"error": str(e)}, status=status)
return
# Delegate outside the response-error handler: the child owns its stream.
for prefix, application in self.mounts.items():
if not prefix or path == prefix or path.startswith(prefix + '/'):
scope = request.scope.copy()
scope['root_path'] = root_path + prefix
await application(scope, request.receive, request.send)
return
await self.send_response(request.send, {"error": "Not Found"}, status=404)
async def dispatch_response(self, request: Request, response: Any) -> None:
"""Invoke an ASGI response or serialize an ordinary handler result."""
if callable(response):
await response(request.scope, request.receive, request.send)
else:
await self.send_response(request.send, response)
async def send_response(self, send: Callable[[dict[str, Any]], Awaitable[None]], response: Any, status: int = 200) -> None:
"""Send JSON or (body, content_type, status[, headers]) with string header pairs."""
body, content_type = response, 'application/octet-stream'
headers = []
if isinstance(response, tuple) and len(response) in (3, 4):
body, content_type, status = response[:3]
if len(response) == 4:
headers = [(name.lower().encode('ascii'), value.encode('latin-1')) for name, value in response[3]]
if isinstance(body, (dict, list)):
body = (orjson.dumps(body, option=orjson.OPT_NON_STR_KEYS) if orjson else
json.dumps(body, ensure_ascii=False, separators=(',', ':')).encode('utf-8'))
content_type = 'application/json'
elif not isinstance(body, bytes):
raise VeelaResponseError("Response must be a dictionary, list, bytes, or response tuple")
await send({
'type': 'http.response.start',
'status': status,
'headers': [(b'content-type', content_type.encode('ascii'))] + headers,
})
await send({'type': 'http.response.body', 'body': body})
def match_path(self, pattern: str, path: str) -> dict[str, str] | None:
"""Match a path against a pattern and extract parameters."""
pattern_parts = pattern.strip('/').split('/')
path_parts = path.strip('/').split('/')
if len(pattern_parts) != len(path_parts):
return None
params = {}
for pattern_part, path_part in zip(pattern_parts, path_parts):
if pattern_part.startswith('{') and pattern_part.endswith('}'):
param_name = pattern_part[1:-1]
params[param_name] = path_part
elif pattern_part != path_part:
return None
return params
async def static(self, filename: str | Path, *, root: str | Path) -> tuple[bytes, str, int]:
"""Serve a small file inside root; read its entire body off the event loop."""
def read() -> tuple[bytes, str, int]:
try:
directory = Path(root).resolve()
path = (directory / filename).resolve()
if not path.is_relative_to(directory) or not path.is_file():
return b'Not Found', 'text/plain; charset=utf-8', 404
body = path.read_bytes()
except (OSError, ValueError, RuntimeError):
return b'Not Found', 'text/plain; charset=utf-8', 404
return body, guess_type(path.name)[0] or 'application/octet-stream', 200
return await to_thread(read)
def mount(self, path: str, application: Callable) -> None:
"""Delegate an HTTP subtree to an ASGI app, after ordinary routes."""
if not isinstance(path, str) or not path.startswith('/') or '{' in path or '}' in path:
raise ValueError('Mount path must be an absolute literal path')
if not callable(application):
raise TypeError('Mounted application must be callable')
self.mounts[path.rstrip('/')] = application
self.mounts = dict(sorted(self.mounts.items(), key=lambda item: len(item[0]), reverse=True))
def route(self, path: str, methods: list[str] = ['GET']) -> Callable:
"""Decorator to register routes with specific HTTP methods."""
def decorator(handler: Callable) -> Callable:
routes = self.pattern_routes if '{' in path and '}' in path else self.routes
routes.setdefault(path, {}).update({m.upper(): handler for m in methods})
return handler
return decorator