Quickstart

Install

pip install neutron-py

Or with extras:

pip install "neutron-py[redis,otel]"

Create a Project

neutron new my-api --lang python
cd my-api
pip install -e .

Your First App

from pydantic import BaseModel
from neutron import App, Router
from neutron.nucleus import NucleusClient

app = App(title="My API", version="1.0.0")
router = Router()

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

class User(BaseModel):
    id: int
    name: str
    email: str

@router.get("/users/{user_id}")
async def get_user(user_id: int) -> User:
    user = await app.state.db.sql.query_one(User,
        "SELECT * FROM users WHERE id = $1", user_id)
    return user

@router.post("/users")
async def create_user(input: CreateUser) -> User:
    user = await app.state.db.sql.query_one(User,
        "INSERT INTO users (name, email) VALUES ($1, $2) RETURNING *",
        input.name, input.email)
    return user

@router.get("/users")
async def list_users() -> list[User]:
    return await app.state.db.sql.query(User, "SELECT * FROM users")

app.include_router(router, prefix="/api")

@app.on_event("startup")
async def startup():
    app.state.db = await NucleusClient.connect("postgres://localhost:5432/mydb")

@app.on_event("shutdown")
async def shutdown():
    await app.state.db.close()

if __name__ == "__main__":
    app.run(port=8000)

Run

# Development
neutron dev

# Or directly
python -m neutron run
# or
uvicorn app.main:app --reload --port 8000

Test It

curl http://localhost:8000/health
curl http://localhost:8000/api/users
curl -X POST http://localhost:8000/api/users \
  -H "Content-Type: application/json" \
  -d '{"name": "Alice", "email": "alice@example.com"}'

Middleware Stack

from neutron.middleware import (
    RequestIDMiddleware,
    LoggingMiddleware,
    CORSMiddleware,
    CompressionMiddleware,
    RateLimitMiddleware,
    TimeoutMiddleware,
)

app = App(
    title="My API",
    middleware=[
        RequestIDMiddleware,
        LoggingMiddleware,
        CORSMiddleware(allow_origins=["*"]),
        CompressionMiddleware(minimum_size=500),
        RateLimitMiddleware(rps=100, burst=200),
        TimeoutMiddleware(timeout=30),
    ],
)