Metadata-Version: 2.4
Name: ryuuseigun
Version: 0.3.3
Summary: A lightweight, type-safe, async web framework inspired by Flask
Project-URL: Source, https://github.com/depthbomb/ryuuseigun
Project-URL: Documentation, https://github.com/depthbomb/ryuuseigun?tab=readme-ov-file#kitchen-sink-example
Project-URL: Changelog, https://github.com/depthbomb/ryuuseigun/blob/master/CHANGELOG.md
Project-URL: Issues, https://github.com/depthbomb/ryuuseigun/issues
Author-email: depthbomb <depthbomb@super.fish>
License-Expression: Apache-2.0
License-File: LICENSE
Classifier: Development Status :: 4 - Beta
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Natural Language :: English
Classifier: Operating System :: MacOS
Classifier: Operating System :: Microsoft :: Windows
Classifier: Operating System :: OS Independent
Classifier: Operating System :: POSIX :: Linux
Classifier: Programming Language :: Python
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Software Development
Classifier: Topic :: Software Development :: Libraries
Classifier: Topic :: Software Development :: Libraries :: Application Frameworks
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.10
Provides-Extra: test
Requires-Dist: pytest>=9.0; extra == 'test'
Description-Content-Type: text/markdown

# ☄️ Ryūseigun

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

[![View on PyPi](https://img.shields.io/badge/View_on-PyPi-0073b7?logo=pypi)](https://pypi.org/project/ryuuseigun/)

---

_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)` 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,
        # 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 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.

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

```

## Testing

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