ryuuseigun 0.2.0

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',
        # 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',
        # 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)

# ---------- #
# 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()

# ---------------- #
# 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,
)

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

@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='public, max-age=300',
    )

# If the client sends matching `If-None-Match` or `If-Modified-Since`,
# `apply_conditional_response` returns HTTP 304 automatically.

#----------#
# 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.
# 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'

Testing

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