ryuuseigun 0.1.1
A lightweight, type-safe, async web framework inspired by Flask
Requires Python: >=3.12

# ☄️ Ryūseigun

A lightweight, type-safe, async web framework inspired by [Flask.](https://flask.palletsprojects.com)

---

_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

```shell
pip install ryuuseigun
```

You will also need an ASGI server like [Uvicorn](https://uvicorn.dev/) or
[Granian](https://github.com/emmett-framework/granian). These two servers have been tested with
Ryūseigun, others have not.

## Minimal Example

```py
from ryuuseigun import Ryuuseigun

app = Ryuuseigun(__name__)
```

## Kitchen Sink Example

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

app = Ryuuseigun(
        __name__,
        # Optional. If omitted, `ctx.url_for(..., full_url=True)` derives origin from request headers.
        base_url='http://localhost:8000',
        # 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

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