railpy 0.1.0.dev0

Deterministic high-performance ASGI engine for modern Python backends

🌐 railpy

LIVE EXAMPLE

A deterministic, high-performance ASGI engine for modern Python backends.

Railpy is not just a framework --- it is an execution engine for building scalable backend architectures.

---

Why railpy?

Most Python frameworks force trade-offs.

Framework	Problem
Flask	Simple but messy at scale
Django	Powerful but heavy
FastAPI	Great DX but opinionated

👉 Railpy gives you control + performance + predictable execution.

---

Advantages

⚡ Radix-tree router (O(k))
🧠 Deterministic middleware execution
🧩 Pipeline-based orchestration
🔌 Middleware-first architecture
🪶 Lightweight ASGI core
📝 Pydantic validation support
🌐 Works with Uvicorn / ASGI / Lambda

---

Mental Model

Request
   ↓
Context (ctx)
   ↓
Middleware Pipeline
   ↓
Router (Radix)
   ↓
Handler
   ↓
Response

---

Installation

pip install railpy

---

Quick

from railpy import Railpy
from railpy.middleware import logger, cors

app = Railpy()

app.use(logger)
app.use(cors())


@app.get("/")
async def hello(ctx):
    return {"message": "Hello Railpy 🚀"}

app.run()

# uvicorn main:app

---

Context API

Railpy provides a powerful request context.

Method	Description
ctx.params	URL parameters
ctx.query	Query string
ctx.data["body"]	Parsed request body
ctx.state	Request state storage
ctx.json()	Send JSON
ctx.text()	Send text
ctx.html()	Send HTML
ctx.file()	Send / download file
ctx.redirect()	Redirect
ctx.ok()	200 response
ctx.created()	201 response
ctx.no_content()	204 response
ctx.bad_request()	400 response
ctx.unauthorized()	401 response
ctx.forbidden()	403 response
ctx.not_found()	404 response
ctx.internal_error()	500 response

---

File Download (Railpy Feature)

Railpy has built-in file download support.

@app.get("/download")
async def download(ctx):
    ctx.file(
        "./files/report.pdf",
        filename="monthly-report.pdf"
    )

This automatically sets:

Content-Disposition: attachment
Content-Type: application/pdf

Browsers will download the file automatically.

---

Routing

@app.get("/users")
async def users(ctx):
    return {"users": []}

@app.get("/dashboard")
async def user(ctx):
    return {"dashboard": []}

---

Route Groups

app.group("/admin", lambda r: (
    r.get("/dashboard", dashboard_handler),
    r.get("/users/:id", admin_user_handler)
))

---

Validation (Pydantic)

from pydantic import BaseModel
from railpy import ValidateBody

class UserCreate(BaseModel):
    name: str
    email: str


@app.post("/users")
@ValidateBody(UserCreate)
async def create_user(ctx, body: UserCreate):
    return body

---

Controller (Decorator Style)

from railpy import Controller, Get

@Controller("/users")
class UserController:

    @Get("/")
    async def list(self, ctx):
        return {"users": []}

    @Get("/:id")
    async def get(self, ctx, id: int):
        return {"id": id}

Register controller:

from railpy import register_controller

register_controller(app, UserController)

---

Swagger / OpenAPI

from railpy import setup_swagger

setup_swagger(app)

Docs:

    /docs

OpenAPI JSON:

    /openapi.json

---

Background Tasks

Railpy allows you to run asynchronous background tasks after the response is sent.
This is useful for sending emails, logging, or other post-response operations without delaying the client.

Usage

from railpy import Railpy, Context
import asyncio

app = Railpy()

async def send_email(user_id: int):
    await asyncio.sleep(1)  # simulate async operation
    print(f"Email sent to user {user_id}")

async def register_user(ctx: Context):
    user_id = ctx.params["id"]

    # Define background task
    async def send_email(u_id):
        await asyncio.sleep(1)  # simulate async operation
        print(f"Email sent to {u_id}")

    # Add the background task
    ctx.background_tasks.append((send_email, (user_id,), {}))

    return {"message": "User registered"}

# Register the route with Railpy
app.get("/register/:id", register_user)

How it works

• During request processing, append tasks to ctx.background_tasks. Each task is a tuple: (callable, args, kwargs).
• After the response is finalized, Railpy automatically runs each task with asyncio.create_task().
• Tasks run concurrently and do not block the client.

ctx.background_tasks = [
    (some_async_fn, (arg1, arg2), {"kwarg1": "value"}),
    # (another_async_fn, (), {}),
]

---

Lambda Support

from railpy import Railpy, RailpyLambda

app = Railpy()

@app.get("/")
async def hello(ctx):
    return {"message": "Hello from Lambda 🚀"}

handler = RailpyLambda(app)

Deploy with:

AWS Lambda
+ API Gateway

Your handler:

handler.handler

Directory structure:

project/
 ├── main.py
 └── requirements.txt

Serverless Architecture

API Gateway
     ↓
AWS Lambda
     ↓
RailpyLambda Adapter
     ↓
Railpy Core
     ↓
Router → Handler

---

Railpy Architecture

    ASGI
     ↓
    Railpy Core
     ↓
    Middleware Pipeline
     ↓
    Radix Router
     ↓
    Handler
     ↓
    Response

---

Ecosystem Middleware

Railpy uses a pipeline-first middleware architecture. Key middleware include:

Middleware	Purpose
logger()	Logs requests and response times
error_handler()	Global error boundary
cors()	Adds CORS headers
body_parser()	Parses JSON / form / multipart bodies
json_parser()	JSON body parser with size limit and strict validation
query_parser()	Parses query strings into ctx.query
rate_limit()	IP-based request rate limiting
jwt_auth(secret)	JWT authentication and user injection into ctx.data["user"]
session(SessionOpts)	Session management with HMAC signing
serve_static(path)	Serve static files safely from disk
helmet()	Adds standard security headers
normalize_headers()	Lowercase all headers for consistency

Quick Example: Full Middleware Stack

from railpy import Railpy, Controller, Get, Post, Param, Context, register_controller, Use
from railpy.middlewares import (
    logger, error_handler, cors, body_parser,
    rate_limit, jwt_auth, session, SessionOpts,
    query_parser, serve_static
)

app = Railpy()

# =========================
# Middleware Stack
# =========================
SESSION_SECRET = "super-secret-session"
JWT_SECRET = "super-secret-jwt"

app.use(logger)                     # Logs requests/responses
app.use(error_handler)              # Global error handling
app.use(rate_limit(limit=100))      # Rate limit per IP
app.use(cors())                      # Allow all origins
app.use(body_parser)                # Parse JSON/form bodies
app.use(query_parser())             # Parse query string into ctx.query
app.use(session(SessionOpts(secret=SESSION_SECRET))) # Session management
app.use(serve_static("./downloads"))  # Serve static files from ./downloads

# =========================
# Public Route
# =========================
@app.get("/")
async def home(ctx: Context):
    return {"message": "Welcome to Railpy 🚀"}

# =========================
# Protected Route
# =========================
@app.get("/secret", jwt_auth(JWT_SECRET))
async def secret_route(ctx: Context):
    user = ctx.data.get("user")
    return {"message": "Hello JWT", "user": user}

# =========================
# Controller Example
# =========================
@Controller("/api/v1")
@Use(jwt_auth(JWT_SECRET))
class UserExpress:

    @Get("/users/:id")
    @Param("id")
    async def get_user(self, user_id: str, ctx: Context):
        async def get_user(self, user_id: str, ctx: Context):
        user = ctx.data.get("user")

        # Add a background task inside controller
        async def send_welcome_email(u_id: str):
            await asyncio.sleep(1)  # simulate async operation
            print(f"Email sent to user {u_id}")

        ctx.background_tasks.append((send_welcome_email, (user_id,), {}))

        return {
            "user_id": user_id,
            "logged_in_user": ctx.data.get("user")
        }

    @Post("/users")
    async def create_user(self, ctx: Context):
        body = ctx.data.get("body")
        return {
            "message": "User created",
            "body_received": body,
            "logged_in_user": ctx.data.get("user")
        }

    @Get("/profile")
    async def profile(self, ctx: Context):
        return {
            "message": "Your profile",
            "user": ctx.data.get("user"),
            "session": ctx.state["session"]
        }

register_controller(app, UserExpress)

# =========================
# Run Server
# =========================
if __name__ == "__main__":
    app.start(port=3000)

---

Features demoed:

• Logging (logger)
• Error handling (error_handler)
• Rate limiting (rate_limit)
• CORS (cors)
• Body parsing (body_parser)
• Query parsing (query_parser)
• JWT auth (jwt_auth)
• Session management (session)
• Static file serving (serve_static)
• Controller routes (@Controller)
• Static File Example

---

Static File Example

Put two files in ./downloads/ folder:

./downloads/readme_download1.txt
./downloads/readme_download2.txt

Then access via browser or curl:

curl http://localhost:3000/readme_download1.txt
curl http://localhost:3000/readme_download2.txt

Session Example

@app.get("/set-session")
async def set_session(ctx: Context):
    ctx.state["session"]["user"] = "admin"
    return {"session_set": True}

@app.get("/get-session")
async def get_session(ctx: Context):
    return {"session": ctx.state["session"]}

JWT Example

from jwt import encode

token = encode({"sub": "1234"}, JWT_SECRET, algorithm="HS256")

# Send in Authorization header: "Bearer <token>"

Query Parsing Example

@app.get("/search")
async def search(ctx: Context):
    # ?q=python&page=2
    q = ctx.query.get("q")
    page = ctx.query.get("page", 1)
    return {"query": q, "page": page}

Body Parsing Example

@app.post("/echo")
async def echo(ctx: Context):
    body = ctx.data.get("body")
    return {"you_sent": body}

Downloadable README Example Files

./downloads/readme_download1.txt:

Railpy Download File 1

This is a sample file for testing Railpy static file serving.

./downloads/readme_download2.txt:

Railpy Download File 2

Another sample file for testing static file downloads.

---

Comparison

Criteria	Railpy	FastAPI	Flask	Django
Core concept	Execution engine	API framework	Micro framework	Full framework
Performance	⚡ High	⚡ High	⚠️ Medium	⚠️ Medium
Routing	Radix	Starlette	Werkzeug	Django
Middleware	Deterministic	Stack	Stack	Stack
Architecture	Flexible	Opinionated	Flexible	Structured
Serverless	✅ Native	⚠️ Adapter	⚠️ Adapter	❌

---

Benchmark

Railpy is designed for minimal overhead and deterministic execution.

Basic benchmark comparison (simple JSON response):

Framework	Req/sec	Notes
Flask	~5k	WSGI
Django	~7k	Full framework
FastAPI	~18k	Starlette + Pydantic
Starlette	~22k	Minimal ASGI
Railpy	~25k+	Radix router + compiled middleware

Benchmark example:

from railpy import Railpy

app = Railpy()

@app.get("/")
async def hello(ctx):
    return {"hello": "world"}

Run benchmark:

uvicorn main:app --workers 1

Test using wrk:

wrk -t4 -c100 -d30s http://localhost:8000/

Example output:

Running 30s test @ http://localhost:8000
4 threads and 100 connections

Requests/sec: 25000+
Latency: ~3ms

Performance varies depending on hardware and middleware stack.

---

When to Use

✔ High-performance APIs.
✔ Microservices.
✔ Serverless backends.
✔ Custom frameworks.

---

Philosophy

You control:
- architecture
- middleware
- data

Railpy controls:
- execution
- routing
- lifecycle

---

License

MIT