neutronapi 0.6.8

A modern, high-performance ASGI-based Python web framework.

NeutronAPI

Async Python web framework for building APIs with first-class commands, models, migrations, background tasks, and ASGI support.

Source: github.com/aaronkazah/neutronapi

Install

Use the package

pip install neutronapi
python -m neutronapi --help

Work on the framework

python3.12 -m venv venv
source venv/bin/activate
python -m pip install -e .
python -m neutronapi --help

Quick Start

neutronapi startproject blog
cd blog
python manage.py check
python manage.py start --no-reload

Create an app:

python manage.py startapp posts

Add your API in apps/posts/api.py:

from neutronapi.base import API, endpoint


class PostAPI(API):
    resource = "/posts"
    name = "posts"

    @endpoint("/", methods=["GET"], name="list")
    async def list_posts(self, scope, receive, send, **kwargs):
        return await self.response(
            [
                {"id": "post_1", "title": "Hello"},
            ]
        )

Register it in apps/entry.py:

from neutronapi.application import Application
from apps.posts.api import PostAPI


app = Application(
    apis=[
        PostAPI(),
    ],
)

Then verify and run:

python manage.py check
python manage.py test
python manage.py start --no-reload

Project Layout

myproject/
├── manage.py
└── apps/
    ├── __init__.py
    ├── settings.py
    ├── entry.py
    └── posts/
        ├── __init__.py
        ├── api.py
        ├── models.py
        ├── commands/
        ├── migrations/
        └── tests/

Core Commands

python -m neutronapi --help
neutronapi startproject blog
python manage.py check
python manage.py start
python manage.py startapp posts
python manage.py makemigrations
python manage.py migrate
python manage.py test

Test database selection:

python manage.py test
python manage.py test --database sqlite
python manage.py test --database postgres

auto is the default:

• in the NeutronAPI source tree it uses SQLite
• in a real project it uses DATABASES["default"]

Settings

Minimal apps/settings.py:

import os
from pathlib import Path


BASE_DIR = Path(__file__).resolve().parent.parent
ENTRY = "apps.entry:app"

DATABASES = {
    "default": {
        "ENGINE": "aiosqlite",
        "NAME": ":memory:" if os.getenv("TESTING") == "1" else BASE_DIR / "db.sqlite3",
    }
}

SECRET_KEY = os.getenv("SECRET_KEY", "dev-key-change-me")
DEBUG = os.getenv("DEBUG", "true").lower() == "true"
ALLOWED_HOSTS = ["127.0.0.1", "localhost"]

Endpoints

Use the decorator aliases directly:

from neutronapi.base import API, endpoint, websocket


class HelloAPI(API):
    resource = "/hello"
    name = "hello"

    @endpoint("/", methods=["GET"], name="home")
    async def home(self, scope, receive, send, **kwargs):
        return await self.response({"message": "Hello from NeutronAPI"})

    @websocket("/stream")
    async def stream(self, scope, receive, send, **kwargs):
        await send({"type": "websocket.accept"})
        await send({"type": "websocket.send", "text": "connected"})
        await send({"type": "websocket.close", "code": 1000})

Models and Migrations

from neutronapi.db.fields import CharField, TextField
from neutronapi.db.models import Model


class Post(Model):
    title = CharField(max_length=255)
    body = TextField(null=True)

Generate and apply migrations:

python manage.py makemigrations
python manage.py migrate

Logging and Request IDs

NeutronAPI logs under the neutronapi.* namespace.

from neutronapi.logging import configure_logging, get_logger


configure_logging(level="INFO", fmt="json")

logger = get_logger(__name__)
logger.info("application booted")

You can also configure logging from apps/settings.py:

LOGGING = {
    "level": "INFO",
    "format": "json",
}

Every HTTP response gets X-Request-Id.

Events

from neutronapi.event_bus import events


@events.on("request.completed")
async def on_request(event):
    print(event.request_id, event.path, event.status)

Throttling

from neutronapi.base import API, endpoint
from neutronapi.throttling import BaseThrottle


class RateThrottle(BaseThrottle):
    async def allow_request(self, scope: dict) -> bool:
        return True

    async def wait(self) -> int | None:
        return None

    async def get_headers(self) -> dict[str, str]:
        return {
            "X-RateLimit-Limit": "100",
            "X-RateLimit-Remaining": "99",
            "X-RateLimit-Reset": "1717200000",
        }


class ItemAPI(API):
    resource = "/items"
    name = "items"

    @endpoint("/", methods=["POST"], throttle_classes=[RateThrottle])
    async def create_item(self, scope, receive, send, **kwargs):
        return await self.response({"ok": True})

Throttle headers are included on normal responses and 429 responses. Throttled responses also include Retry-After.

Idempotency

from neutronapi.application import Application
from neutronapi.idempotency import IdempotencyMiddleware, InMemoryIdempotencyStore


app = Application(
    apis=[PostAPI()],
    middlewares=[
        IdempotencyMiddleware(store=InMemoryIdempotencyStore(), ttl=86400),
    ],
)

Replay responses include:

• Idempotency-Key
• Idempotent-Replayed: true

Geo Middleware

from neutronapi.application import Application
from neutronapi.middleware import CloudflareGeoMiddleware, MaxMindGeoMiddleware


app = Application(
    apis=[PostAPI()],
    middlewares=[
        MaxMindGeoMiddleware(),
        CloudflareGeoMiddleware(),
    ],
)

Middleware order is the fallback chain. The first middleware that sets scope["_neutronapi_geo"] wins, so custom geo middleware can run ahead of the built-ins.

MaxMindGeoMiddleware uses GEOIP_DATABASE_PATH from apps/settings.py and enriches the scope with country_code, region, city, latitude, and longitude. CloudflareGeoMiddleware is a lightweight fallback that only fills country_code from cf-ipcountry.

Background Tasks

from neutronapi.application import Application
from neutronapi.background import Task, TaskFrequency


class CleanupTask(Task):
    name = "cleanup"
    frequency = TaskFrequency.HOURLY

    async def run(self, **kwargs):
        return None


app = Application(
    apis=[PostAPI()],
    tasks={"cleanup": CleanupTask()},
)

Custom Commands

Create apps/posts/commands/greet.py:

from typing import List


class Command:
    def __init__(self):
        self.help = "Greet a user"

    async def handle(self, args: List[str]) -> None:
        name = args[0] if args else "World"
        print(f"Hello, {name}!")

Run it:

python manage.py greet Alice
python manage.py greet --help

Development

Run the framework test suite from the repo root:

source venv/bin/activate
python manage.py test
python manage.py test --database sqlite
python manage.py test --database postgres