ryuuseigun 0.5.1

A lightweight, type-safe, async web framework inspired by Flask

☄️ Ryūseigun

A lightweight, type-safe, async web framework inspired by Flask.

---

Ryūseigun is deliberately minimal, embracing a “bring-your-own-X” philosophy. It provides just enough to give you something that works and leaves plenty of room to do things how you want to.

Installation

pip install ryuuseigun

You will also need an ASGI server like Uvicorn or Granian. These two servers have been tested with Ryūseigun, others have not.

Minimal Example

from ryuuseigun import Ryuuseigun

app = Ryuuseigun(__name__)

Kitchen Sink Example

from json import loads, dumps
from ryuuseigun import Context, Response, Ryuuseigun

app = Ryuuseigun(
        __name__,
        # Optional. If omitted, `ctx.url_for(..., full_url=True)` uses Host header from the request.
        base_url='http://localhost:8000',
        # Optional. Centralized renderer for framework-generated errors.
        error_renderer=None,
        # Trust X-Forwarded-* / Forwarded for URL generation (enable only behind trusted proxies).
        trust_proxy_headers=False,
        # A route ending with a slash is treated the same as a route without a slash
        strict_slashes=False,
        # Serve files from this path (e.g. public/favicon.ico -> website.com/favicon.ico)
        public_dir='./public',
        # Serve precompressed static assets (for example app.js.br / app.js.gz) when accepted.
        static_precompressed=False,
        # Preferred precompressed static encodings in match order.
        static_precompressed_encodings=('br', 'gzip'),
        # Optional dev-server proxy target (for example Vite).
        dev_proxy_target=None,  # e.g. 'http://127.0.0.1:5173'
        # Prefixes that should never be proxied (API routes by default).
        dev_proxy_exclude_prefixes=('/api',),
        # If True, proxy only when no route/static file matches.
        dev_proxy_fallback_only=True,
        # Optional SPA history fallback target. Disabled by default.
        spa_fallback_index=None,  # e.g. '/index.html'
        # Optional path prefixes that should never use SPA fallback.
        spa_fallback_exclude_prefixes=('/api',),
        # Require `Accept: text/html` for SPA fallback.
        spa_fallback_require_html_accept=True,
        # Optional default secret for encrypted cookie helpers in `ryuuseigun.utils`.
        cookie_encryption_secret=None,  # e.g. 'replace-me-in-prod'
        # Reject request bodies over this limit with HTTP 413 (set None to disable)
        max_request_body_size=16 * 1024 * 1024,
        # JSON parser/serializer hooks (swap with orjson.loads/orjson.dumps if desired)
        loads=loads,
        dumps=dumps,
)

# -------------- #
# Error handlers #
# -------------- #
@app.error_handler(Exception)  # More specific exception classes will be prioritized
async def handle_exceptions(ctx: Context, e: Exception) -> Response:
    return Response(str(e), status=500)

# Optional: central renderer used for framework-generated errors
# (for example malformed UTF-8 in headers/query, body-size preflight errors,
# and unhandled exceptions without a matching error handler).
from ryuuseigun import ErrorRenderContext, KnownContentType

def error_renderer(error: ErrorRenderContext):
    return (
        {
            'ok': False,
            'error': {
                'code': error.code,
                'status': error.status,
                'message': error.message,
                'stage': error.stage,
            },
        },
        error.status,
        {'content-type': KnownContentType.JSON},
    )
# Then pass `error_renderer=error_renderer` when creating `Ryuuseigun(...)`.

# ---------- #
# Blueprints #
# ---------- #
from ryuuseigun import Blueprint

api_bp = Blueprint('api', url_prefix='/api')

@api_bp.get('/')
async def api_index() -> str:
    return 'API docs'

@api_bp.error_handler(Exception)  # Blueprint-specific error handlers
async def handle_api_errors(ctx: Context, e: Exception) -> dict:
    return {
        'success': True,
        'message': str(e),
    }

users_bp = Blueprint('users', url_prefix='/users')

@users_bp.get('/<username>')
async def get_user(ctx: Context) -> str:
    return ctx.request.route_params['username']

api_bp.register_blueprint(users_bp)
app.register_blueprint(api_bp)  # GET /api/users/caim -> 'caim'

# ------------------------------------ #
# Request globals & lifecycle handlers #
# ------------------------------------ #
from typing import cast

@app.before_request
async def before_all_requests(ctx: Context):
    ctx.g['value'] = 123

    @ctx.after_this_request  # Called after *this* request
    async def after_all_requests(response: Response):
        response.set_header('X-My-Value', str(ctx.g['value']))
        return response

    return None

@app.route('/', methods=['GET'])
async def index(ctx: Context) -> str:  # Route handlers can return `str`, `dict`, or `Response` (`Response.stream(...)` for streaming)
    my_value = cast(int, ctx.g['value'])
    return str(my_value)

# -------------------- #
# ASGI lifespan events #
# -------------------- #
@app.on_startup
async def startup() -> None:
    await connect_db()

@app.on_shutdown
async def shutdown() -> None:
    await disconnect_db()

# ----------------- #
# WebSocket routes  #
# ----------------- #
@app.websocket('/ws/chat')
async def chat(ws) -> None:
    while True:
        text = await ws.receive_text()
        if text is None:
            return
        await ws.send_text(f'echo:{text}')

# WebSocket handlers can also receive request context for route params, cookies, etc.
@app.websocket('/ws/rooms/<int:room_id>')
async def room_socket(ws, ctx: Context) -> None:
    room_id = ctx.request.route_param('room_id', as_type=int)
    await ws.send_text(f'room:{room_id}')

# ---------------- #
# Route parameters #
# ---------------- #
@app.post('/multiply/<int:num>/<int:factor>')  # HTTP method shortcuts for convenience
async def index(ctx: Context) -> dict:
    num = ctx.request.route_param('num', as_type=int)
    factor = ctx.request.route_param('factor', as_type=int)
    return {
        'num': num,
        'factor': factor,
        'product': num * factor,
    }

# ---------------- #
# Route converters #
# ---------------- #
# Built-in converters: `int`, `float`, and `path`.
# Route matching is specificity-aware: static segments and typed converters win over broader `path` captures.
# Register custom route converters with parse/format behavior:
app.register_converter(
    'hex',
    regex=r'[0-9a-fA-F]+',
    parse=lambda raw: int(raw, 16),
    format=lambda value: format(int(value), 'x'),
)

@app.get('/colors/<hex:color>')
async def show_color(ctx: Context) -> dict:
    color = ctx.request.route_param('color')  # int (parsed from hex)
    return {'decimal': color}

@app.get('/links/color')
async def color_link(ctx: Context) -> dict:
    # Uses converter `format`: -> /colors/ff
    return {'url': ctx.url_for('show_color', color=255)}

# Converters can also be scoped to blueprints:
assets_bp = Blueprint('assets', url_prefix='/assets')
assets_bp.register_converter(
    'slugpath',
    regex=r'.+',
    parse=str,
    format=str,
    allows_slash=True,  # allow values like "images/icons/logo.svg"
)

@assets_bp.get('/<slugpath:key>')
async def get_asset(ctx: Context) -> dict:
    return {'key': ctx.request.route_param('key')}

app.register_blueprint(assets_bp)

# -------------------- #
# Request body (async) #
# -------------------- #
@app.post('/upload')
async def upload(ctx: Context) -> dict:
    total = 0
    async for chunk in ctx.request.iter_body():
        total += len(chunk)
    return {'bytes': total}

@app.post('/json')
async def json_endpoint(ctx: Context) -> dict:
    payload = await ctx.request.json_async()
    return {'ok': True, 'payload': payload}

# In ASGI request flow, body parsing is stream-first.
# Use `await request.read()/json_async()/form_async()/payload_async()` for body access.

# ------------------------------ #
# Parsing/coercion customization #
# ------------------------------ #
# Register custom request payload parsers by content-type match:
app.add_request_payload_parser(
    'text/csv',
    lambda request: [row.split(',') for row in request.body.decode('utf-8').splitlines() if row],
    first=True,  # check before built-in parsers
)

# Register custom response coercers for arbitrary return types:
class Box:
    def __init__(self, value: str):
        self.value = value

app.add_response_coercer(
    lambda result: Response(f'box:{result.value}', status=201) if isinstance(result, Box) else None,
    first=True,
)

# -------------------------- #
# Precompressed static files #
# -------------------------- #
# If your build emits `.br` / `.gz` variants next to assets, Ryūseigun can
# serve them automatically based on `Accept-Encoding`.
app = Ryuuseigun(
    __name__,
    public_dir='./public',
    static_precompressed=True,
    static_precompressed_encodings=('br', 'gzip'),
)

# Example: request `/assets/app.js`
# - serves `/assets/app.js.br` when `Accept-Encoding` allows `br`
# - else serves `/assets/app.js.gz` when `gzip` is allowed
# - else serves `/assets/app.js`
# When enabled, static responses include `Vary: accept-encoding`.

# ----------------------- #
# Dev proxy helper (Vite) #
# ----------------------- #
# During SPA development, proxy unresolved requests to a dev server.
app = Ryuuseigun(
    __name__,
    dev_proxy_target='http://127.0.0.1:5173',
    dev_proxy_exclude_prefixes=('/api',),
    dev_proxy_fallback_only=True,
)

# Behavior:
# - with `dev_proxy_fallback_only=True`, app routes/static files win first
# - unmatched paths are proxied to the dev server
# - excluded prefixes (like `/api`) are never proxied
# - proxy failures return HTTP 502

# ------------------------------ #
# Conditional caching (optional) #
# ------------------------------ #
from datetime import datetime, timezone
from ryuuseigun.utils import (
    make_etag,
    cache_control,
    etag_for_file,
    last_modified_for_file,
    apply_conditional_response,
)

@app.get('/assets/app.js')
async def app_js(ctx: Context) -> Response:
    body = b'console.log("hello")\n'
    response = Response(
        body=body,
        headers={'content-type': 'application/javascript'},
    )
    return apply_conditional_response(
        ctx,
        response,
        etag=make_etag(body),
        last_modified=datetime(2026, 2, 20, tzinfo=timezone.utc),
        cache_control=cache_control(public=True, max_age=300, immutable=True),
    )

# If the client sends matching `If-None-Match` or `If-Modified-Since`,
# `apply_conditional_response` returns HTTP 304 automatically.
# For file-backed responses, helpers are available:
# etag = etag_for_file('./public/assets/app.js')
# last_modified = last_modified_for_file('./public/assets/app.js')

# -------------------- #
# SPA history fallback #
# -------------------- #
# For SPAs using client-side routing, you can serve `index.html` for unresolved
# GET/HEAD routes while keeping API paths excluded.
app = Ryuuseigun(
    __name__,
    public_dir='./public',
    spa_fallback_index='/index.html',
    spa_fallback_exclude_prefixes=('/api',),
    spa_fallback_require_html_accept=True,
)

# Behavior:
# - Disabled when `spa_fallback_index` is None (default)
# - Only considered for unresolved GET/HEAD requests
# - Never used for 405 method-mismatch paths
# - Skips file-like paths (for example `/app.js`)
# - Respects excluded prefixes such as `/api`
# - Optionally requires `Accept: text/html`

# -------- #
# Sessions #
# -------- #
# Sessions default to an in-memory engine (per process, cookie-identified, not shared across workers).
# You can tune it with `session_ttl`, `session_purge_interval`, and `session_max_entries`, or fully
# replace storage by passing a custom `session_engine` object that implements: `load`, `create`,
# `save`, and `destroy` as async methods.
# You can also customize the session cookie key name with `session_cookie_name`.
# Session access is async-first: use `await ctx.session_async()` when you need to create/load a
# session.
@app.get('/login')
async def login(ctx: Context) -> str:
    session = await ctx.session_async()
    session['user_id'] = 123
    return 'ok'

@app.get('/me')
async def me(ctx: Context) -> str:
    session = await ctx.session_async()
    user_id = session.get('user_id')
    return str(user_id or 'anonymous')

@app.get('/logout')
async def logout(ctx: Context) -> str:
    session = await ctx.session_async()
    session.destroy()
    return 'bye'

# ------------------------- #
# Utility helper highlights #
# ------------------------- #
from ryuuseigun.utils import (
    problem,
    signed_token,
    json_response,
    safe_filename,
    encrypted_token,
    get_signed_cookie,
    set_signed_cookie,
    verify_signed_token,
    set_encrypted_cookie,
    get_encrypted_cookie,
    decrypt_encrypted_token,
)

@app.get('/profile')
async def profile(ctx: Context):
    # Uses app.dumps from `Ryuuseigun(..., dumps=...)`.
    return json_response(ctx, {'ok': True, 'user': 'alice'})

@app.get('/problem-demo')
async def problem_demo(ctx: Context):
    return problem(
        ctx,
        404,
        detail='Resource not found',
        extra={'code': 'resource_not_found'},
    )

# Signed tokens are tamper-evident (not encrypted).
token = signed_token({'user_id': 123}, ttl=300, secret='signing-secret')
payload = verify_signed_token(token, 'signing-secret')

# Encrypted tokens provide confidentiality + integrity.
enc = encrypted_token('alice', ttl=300, secret='enc-secret')
value = decrypt_encrypted_token(enc, 'enc-secret')

@app.get('/set-signed-cookie')
async def set_signed_cookie_route():
    response = Response('ok')
    # Response method:
    response.set_signed_cookie(
        'auth',
        'alice',
        secret='signing-secret',
        ttl=3600,
        httponly=True,
    )
    # Utility wrapper (same behavior):
    # set_signed_cookie(response, 'auth', 'alice', secret='signing-secret', ttl=3600, httponly=True)
    return response

@app.get('/read-signed-cookie')
async def read_signed_cookie_route(ctx: Context):
    # Key rotation: new secret first, old secrets after.
    value = get_signed_cookie(
        ctx,
        'auth',
        secret=('new-signing-secret', 'old-signing-secret'),
    )
    return value or 'anonymous'

@app.get('/set-encrypted-cookie')
async def set_encrypted_cookie_route(ctx: Context):
    response = Response('ok')
    # Response method:
    response.set_encrypted_cookie(
        'vault',
        'alice',
        secret=app.cookie_encryption_secret,
        ttl=3600,
        httponly=True,
    )
    # Utility wrapper (uses app secret when `secret` is omitted):
    # set_encrypted_cookie(ctx, response, 'vault', 'alice', ttl=3600, httponly=True)
    return response

@app.get('/read-encrypted-cookie')
async def read_encrypted_cookie_route(ctx: Context):
    # Can also pass explicit rotation secrets:
    value = get_encrypted_cookie(
        ctx,
        'vault',
        secret=(app.cookie_encryption_secret, b'older-secret-bytes'),
    )
    return value or 'anonymous'

safe_name = safe_filename('../../etc/passwd')  # -> 'passwd'

Testing

pip install -e ".[test]"
python -m pytest