railpy 0.2.0

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.

• ⚡ 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, Context
from railpy.middlewares import logger, cors

app = Railpy()

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

async def home(ctx: Context):
    return {"message": "Hello Railpy 🚀"}

# routing
app.get("/", home)

Run the application:

uvicorn app:app --reload

💡 Note:

• app:app → refers to the app.py file and the app instance
• --reload → automatically reloads the server when code changes
• This method is the same as FastAPI, standard for ASGI applications.

---

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 has built-in file download support.

from railpy import Railpy, Context

app = Railpy()

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

app.get("/download", download)

This automatically sets:

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

Browsers will download the file automatically.

---

Routing

from railpy import Controller, Get, Context, register_controller

@Controller("/")
class AppController:

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

# register controller
register_controller(app, AppController)

---

Route Groups

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

---

Validation (Pydantic)

from railpy import Controller, Post, Context, register_controller, ValidateBody

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


@Controller("/users")
class UserController:

    @Post("/")
    @ValidateBody(UserCreate)
    async def create_user(self, ctx: Context, body: UserCreate):
        return {
            "message": "User created",
            "data": body
        }


# register controller
register_controller(app, UserController)

---

Controller (Decorator)

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, (), {}),
]

---

Global Error Handling Example

Example: Global Error Handler

# app.py
from railpy import Railpy, Context

app = Railpy()

# =========================
# Global Error Handler
# =========================
@app.exception_handler
async def global_error_handler(ctx: Context, err: Exception):
    print(f"❌ [System Error]: {err}")

    ctx.status = 500
    ctx.body = {
        "success": False,
        "message": "Internal server error",
        "error_type": type(err).__name__
    }

# =========================
# Route that triggers an error
# =========================
async def fail_route(ctx: Context):
    # Simulate a crash
    1 / 0

app.get("/fail", fail_route)

Example Output

Request: GET /fail

{
    "success": false,
    "message": "Internal server error",
    "error_type": "ZeroDivisionError"
}

Console:

❌ [System Error]: division by zero

---

Lambda / Serverless

# lambda_test.py
from railpy import Railpy, RailpyLambda, Context

app = Railpy()

# Route example
async def hello(ctx: Context):
    return {"message": "Hello from Lambda 🚀"}

app.get("/api/v1/users", hello)

# Lambda handler
handler = RailpyLambda(app)

# Local test
if __name__ == "__main__":
    event = {
        "httpMethod": "GET",
        "path": "/api/v1/users",
        "headers": {},
        "queryStringParameters": None
    }
    response = handler(event, None)
    print(response)

# Run local: python lambda_test.py

Deploy to AWS Lambda

1. Install dependencies and prepare a requirements.txt: railpy

2. Directory structure:

   project/
    ├── lambda_test.py      # contains Railpy app + handler
    └── requirements.txt

3. Deploy to AWS Lambda (Python 3.10+ recommended):

   - Function handler:
     lambda_test.handler

   - Attach API Gateway (HTTP API) with route /api/v1/users pointing to your Lambda function.

4. Optional local testing with the same Lambda handler:

python lambda_test.py

Architect

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

Full Middleware

# app.py
import asyncio
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,
    helmet,
    session,
    query_parser,
    serve_static,
    normalize_headers,
    SessionOpts
)

# =========================
# App & Config
# =========================
app = Railpy()

JWT_SECRET = "super-secret-jwt"
SESSION_SECRET = "super-secret-session"
session_opts = SessionOpts(secret=SESSION_SECRET)

# =========================
# Global Middlewares
# =========================
app.use(logger)
app.use(error_handler)
app.use(rate_limit(60))
app.use(cors())
app.use(helmet)
app.use(normalize_headers())
app.use(body_parser)
app.use(query_parser())
app.use(session(session_opts))
app.use(serve_static("public"))

# =========================
# Public App-Level Route
# =========================
async def home(ctx: Context):
    return {"message": "Welcome to Railpy 🚀", "query": ctx.query}

app.get("/", home)

# =========================
# Protected App-Level Route with JWT
# =========================
async def secret_route(ctx: Context):
    user = ctx.data.get("user")
    return {"message": "Hello JWT", "user": user}

app.get("/secret", jwt_auth(JWT_SECRET), secret_route)

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

    async def send_email(u_id: str):
        await asyncio.sleep(1)
        print(f"Email sent to {u_id}")

    ctx.background_tasks.append((send_email, (user_id,), {}))
    return {"message": f"User {user_id} registered"}

app.get("/register/:id", register_user)

# =========================
# Controller Example
# =========================
@Controller("/api/v1")
@Use(jwt_auth(JWT_SECRET))  # JWT applied to all routes in controller
class UserController:

    @Get("/users/:id")
    @Param("id")
    async def get_user(self, user_id: str, ctx: Context):
        # Background task inside controller
        async def send_welcome_email(u_id: str):
            await asyncio.sleep(1)
            print(f"Welcome email sent to {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.get("session")}

# Register controller
register_controller(app, UserController)

---

Running

To run your Railpy application, you must use Uvicorn, just like FastAPI:

uvicorn app:app --reload

• app:app → app.py file, app instance
• --reload → enables automatic reload on file changes

Railpy follows the same pattern as FastAPI. This is the standard way to run ASGI apps.

Notes

• Function routes: app.get("/path", handler) or app.post("/path", handler)
• Controller routes: @Controller + @Get / @Post decorators
• Background tasks: ctx.background_tasks.append((fn, args, kwargs))
• JWT protected routes: jwt_auth(secret)
• Session management: session(SessionOpts(secret=...))

---

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

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

from railpy import Railpy, Context
from railpy.middlewares import session, SessionOpts

app = Railpy()
app.use(session(SessionOpts(secret="super-secret-session")))

async def set_session(ctx: Context):
    ctx.state["session"]["user"] = "admin"
    return {"session_set": True}

async def get_session(ctx: Context):
    return {"session": ctx.state.get("session")}

app.get("/set-session", set_session)
app.get("/get-session", get_session)

Query & Body Parsing

from railpy import Railpy, Context

app = Railpy()

# Query parsing
async def search(ctx: Context):
    q = ctx.query.get("q")
    page = ctx.query.get("page", 1)
    return {"query": q, "page": page}

app.get("/search", search)

# Body parsing
async def echo(ctx: Context):
    body = ctx.data.get("body")
    return {"you_sent": body}

app.post("/echo", echo)

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

async def hello(ctx):
    return {"hello": "world"}

app.get("/", hello)

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