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

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

Metadata

Release files for neutronapi 0.6.10

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for neutronapi 0.6.10
File	Size	Uploaded
neutronapi-0.6.10.tar.gz	133.2 kB	May 19, 2026	Details

Built distribution (wheel)

Table of built distributions (wheels) for neutronapi 0.6.10
File	Interpreter	ABI	Platform	Reset
neutronapi-0.6.10-py3-none-any.whl 156.2 kB May 19, 2026	Python 3	none	any	Details

Total release size: 289.4 kB

Release files / neutronapi-0.6.10.tar.gz

Download URL	neutronapi-0.6.10.tar.gz
Size	133.2 kB
Tags	Source
SHA-256 checksum How to use checksums	`f5e2e92e8e104d196ddbffd960fc5c9346fea0c5879ff6a9b2ea2ff536dda0f1`
BLAKE2b-256 checksum How to use checksums	`3640e6400d9db931f3d40b2334641013c1e1dae9692da1f4d9bf18dc6bbe5e47`
Upload date	May 19, 2026
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`twine/6.2.0 CPython/3.12.13`

Release files / neutronapi-0.6.10-py3-none-any.whl

Download URL	neutronapi-0.6.10-py3-none-any.whl
Size	156.2 kB
Tags	Python 3
SHA-256 checksum How to use checksums	`7a0e57a45b18d33c485ec242e04d8ccedcd66caa2ee61e0d969dcd2265afa528`
BLAKE2b-256 checksum How to use checksums	`fe4fcdaf6e096cbf20c448687d4a51a7f1e4e44dbcf16f65707a7352455e811c`
Upload date	May 19, 2026
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`twine/6.2.0 CPython/3.12.13`