Metadata-Version: 2.4
Name: neutronapi
Version: 0.6.5
Summary: A modern, high-performance ASGI-based Python web framework.
Author-email: Aaron Kazah <aaron@neutronapi.com>
License-Expression: MIT
Project-URL: Homepage, https://github.com/aaronkazah/neutronapi
Project-URL: Repository, https://github.com/aaronkazah/neutronapi
Keywords: api,framework,async,fastapi,uvicorn,orm,migrations,background-tasks,dependency-injection,typing
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Internet :: WWW/HTTP :: HTTP Servers
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Framework :: AsyncIO
Requires-Python: >=3.12
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: uvicorn[standard]>=0.23.0
Requires-Dist: uvloop>=0.17.0
Requires-Dist: aiosqlite>=0.19.0
Requires-Dist: orjson>=3.10.0
Requires-Dist: maxminddb>=2.8.0
Requires-Dist: python-multipart>=0.0.9
Requires-Dist: PyJWT>=2.8.0
Requires-Dist: watchdog>=3.0.0
Provides-Extra: postgres
Requires-Dist: asyncpg>=0.29.0; extra == "postgres"
Provides-Extra: dev
Requires-Dist: httpx>=0.24.0; extra == "dev"
Dynamic: license-file

# NeutronAPI

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

Source: [github.com/aaronkazah/neutronapi](https://github.com/aaronkazah/neutronapi)

## Install

### Use the package

```bash
pip install neutronapi
python -m neutronapi --help
```

### Work on the framework

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

## Quick Start

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

Create an app:

```bash
python manage.py startapp posts
```

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

```python
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`:

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


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

Then verify and run:

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

## Project Layout

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

## Core Commands

```bash
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:

```bash
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`:

```python
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:

```python
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

```python
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:

```bash
python manage.py makemigrations
python manage.py migrate
```

## Logging and Request IDs

NeutronAPI logs under the `neutronapi.*` namespace.

```python
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`:

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

Every HTTP response gets `X-Request-Id`.

## Events

```python
from neutronapi.event_bus import events


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

## Throttling

```python
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

```python
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

```python
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

```python
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`:

```python
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:

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

## Development

Run the framework test suite from the repo root:

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