Metadata-Version: 2.4
Name: pyboot-web
Version: 0.1.15
Summary: Starlette web plugin for PyBoot
License-Expression: MIT
Requires-Python: >=3.11
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: pyboot-framework>=0.1.0
Requires-Dist: starlette>=0.40.0
Requires-Dist: pydantic>=2.0
Requires-Dist: python-dotenv>=1.0.0
Dynamic: license-file

# pyboot-web

`pyboot-web` is the Starlette web plugin for
[PyBoot](../../pyboot/README.md). Its `RestPlugin` adds:

- Starlette application creation and lifespan management
- Decorator-based controllers and HTTP routes
- IoC-managed web services, components, and middleware
- HTTP exception advice
- CORS configuration
- `.env` loading
- Interval, fixed-delay, and timezone-aware cron jobs
- Background-job exception advice

The current package registers only `RestPlugin` in the
`pyboot.plugins` entry-point group. `RestPlugin` performs the web
registrations itself.

## Requirements and installation

Python 3.11 or newer is required.

```bash
python -m pip install pyboot-framework pyboot-web uvicorn
```

From this repository:

```bash
python -m pip install -e pyboot
python -m pip install -e plugins/pyboot_web
```

The package depends on Starlette, Pydantic, and `python-dotenv`. Uvicorn is
needed to serve the ASGI application.

## Minimal application

Application classes must be below the package passed to `bootstrap()`:

```text
service/
├── main.py
└── app/
    ├── __init__.py
    └── controllers.py
```

`app/controllers.py`:

```python
from starlette.requests import Request
from starlette.responses import JSONResponse

from pyboot_web.rest_mapping import GetMapping, RestController


@RestController("/api")
class HealthController:
    @GetMapping("/health")
    async def health(self, request: Request) -> JSONResponse:
        return JSONResponse({"status": "ok"})
```

`main.py`:

```python
from pyboot import PyBoot
from pyboot_web.plugins.RestPlugin import RestPlugin


boot = PyBoot(debug=False)
boot.bootstrap(package="app")
app = boot.get_plugin(RestPlugin).app
```

Run it:

```bash
python -m uvicorn main:app --host 127.0.0.1 --port 8000
curl http://127.0.0.1:8000/api/health
```

`RestPlugin` creates the Starlette app. `PyBoot` does not accept an application
instance, and `bootstrap()` returns `None`.

## Controllers and routes

Route decorators are intended for methods inside `RestController` classes:

```python
from starlette.requests import Request
from starlette.responses import JSONResponse

from pyboot_web.rest_mapping import (
    DeleteMapping,
    GetMapping,
    PatchMapping,
    PostMapping,
    PutMapping,
    RestController,
)


@RestController("/api/users")
class UserController:
    @GetMapping("")
    async def list_users(self, request: Request) -> JSONResponse:
        return JSONResponse({"items": []})

    @GetMapping("/{user_id}")
    async def get_user(self, request: Request) -> JSONResponse:
        return JSONResponse({"id": request.path_params["user_id"]})

    @PostMapping("")
    async def create_user(self, request: Request) -> JSONResponse:
        return JSONResponse(await request.json(), status_code=201)

    @PutMapping("/{user_id}")
    async def replace_user(self, request: Request) -> JSONResponse:
        return JSONResponse({"id": request.path_params["user_id"]})

    @PatchMapping("/{user_id}")
    async def update_user(self, request: Request) -> JSONResponse:
        return JSONResponse({"id": request.path_params["user_id"]})

    @DeleteMapping("/{user_id}")
    async def delete_user(self, request: Request) -> JSONResponse:
        return JSONResponse({}, status_code=204)
```

`RequestMapping` stores HTTP methods as a list, so custom usage can map one
method decorator to the generated Starlette route. Path parameters, query
values, and headers come from Starlette's `Request`.

## Services and web components

Use `WebService()` for web-layer services:

```python
from pyboot_web.service import WebService


@WebService()
class UserService:
    def find_user(self, user_id: str) -> dict[str, str]:
        return {"id": user_id, "name": "Ada"}
```

Use `WebComponent()` for other web-plugin-managed classes:

```python
from pyboot_web.component import WebComponent


@WebComponent()
class RequestState:
    def __init__(self) -> None:
        self.value = "ready"
```

PyBoot's core `Component()` remains appropriate for general application
dependencies:

```python
from pyboot.core.builtins.component import Component


@Component()
class AuditService:
    def record(self, message: str) -> None:
        print(message)
```

All three kinds can be injected through annotated constructors:

```python
@RestController("/users")
class UserController:
    def __init__(self, users: UserService) -> None:
        self.users = users
```

## Middleware

```python
from starlette.requests import Request
from starlette.responses import Response

from pyboot_web.middlewares import (
    CallNext,
    MiddlewareBase,
    MiddlewareComponent,
)


@MiddlewareComponent()
class RequestIdMiddleware(MiddlewareBase):
    async def __call__(
        self,
        request: Request,
        call_next: CallNext,
    ) -> Response:
        response = await call_next(request)
        response.headers["x-request-id"] = request.headers.get(
            "x-request-id", "generated-request-id"
        )
        return response
```

Apply middleware to a controller or route with `UseMiddleware`:

```python
from pyboot_web.middlewares import UseMiddleware


@RestController("/api")
@UseMiddleware([RequestIdMiddleware])
class HealthController:
    @GetMapping("/health")
    async def health(self, request: Request) -> JSONResponse:
        return JSONResponse({"status": "ok"})
```

Pass middleware classes, not instances. Controller middleware wraps route
middleware, which wraps the endpoint.

## HTTP exception advice

```python
from starlette.requests import Request
from starlette.responses import JSONResponse

from pyboot_web.exception_handler import ExceptionHandler, RestControllerAdvice


class UserNotFoundError(Exception):
    pass


@RestControllerAdvice()
class ApiExceptionAdvice:
    @ExceptionHandler(UserNotFoundError)
    def user_not_found(
        self,
        request: Request,
        exception: UserNotFoundError,
    ) -> JSONResponse:
        return JSONResponse({"error": str(exception)}, status_code=404)
```

Handlers receive `(request, exception)` after `self`. Exact exception types
are checked first, followed by base classes. Unhandled exceptions are
re-raised.

## Environment values

`RestPlugin` loads a `.env` file from the current working directory before
class discovery:

```dotenv
SERVICE_NAME=user-api
PORT=8000
DEBUG=false
```

Read values with `Value`:

```python
from pyboot_web.value import Value


SERVICE_NAME = Value("SERVICE_NAME", "user-api")
PORT = int(Value("PORT", "8000"))
DEBUG = Value("DEBUG", "false").lower() == "true"
OPTIONAL = Value("OPTIONAL_SETTING", None, throw=False)
```

`Value(key, default, throw=True)` returns an environment value or default.
When both are `None`, the default behavior raises `MissingValueException`.
`throw=False` returns `None` instead. Convert values explicitly.

## CORS

Provide a `CorsConfiguration` bean:

```python
from pyboot.core.builtins.configuration import Bean, Configuration
from pyboot_web.cors import CorsConfiguration


@Configuration()
class WebConfiguration:
    @Bean()
    def cors(self) -> CorsConfiguration:
        return CorsConfiguration(
            allowed_origins=["http://localhost:3000"],
            allowed_methods=["GET", "POST"],
            allowed_headers=["content-type"],
            allow_credentials=True,
            max_age=600,
        )
```

`RestPlugin` detects this bean and adds Starlette's `CORSMiddleware`.

## Background jobs

```python
from pyboot_web.bg_service import BackgroundService, IntervalSchedule


@BackgroundService()
class CleanupJobs:
    @IntervalSchedule(interval_ms=60_000, initial_delay_ms=5_000)
    async def cleanup(self) -> None:
        print("cleanup")
```

For a fixed delay:

```python
from pyboot_web.bg_service import BackgroundService, FixedDelaySchedule


@BackgroundService()
class PollingJobs:
    @FixedDelaySchedule(fixed_delay_ms=30_000, initial_delay_ms=1_000)
    async def poll(self) -> None:
        print("poll")
```

For a timezone-aware cron job:

```python
from datetime import timezone, timedelta

from pyboot_web.bg_service import BackgroundService, CronJob


india = timezone(timedelta(hours=5, minutes=30))


@BackgroundService()
class ReportJobs:
    @CronJob("0 9 * * *", time_zone=india)
    async def daily_report(self) -> None:
        print("daily report")
```

Jobs are scheduled by APScheduler and start with the Starlette lifespan.
Interval/fixed-delay values must be positive. Exceptions are routed through
background exception advice when configured.

## Background exception advice

```python
from pyboot_web.bg_service_exception_handler import (
    BackgroundExceptionHandler,
    BackgroundServiceAdvice,
)


class ReportError(Exception):
    pass


@BackgroundServiceAdvice()
class BackgroundAdvice:
    @BackgroundExceptionHandler(ReportError)
    def report_error(self, exception: ReportError) -> None:
        print(f"background job failed: {exception}")
```

The handler receives the raised exception. Matching checks the exact type and
then base classes. If no handler matches, the exception is re-raised.

## Testing

```python
from starlette.testclient import TestClient

from pyboot import PyBoot
from pyboot_web.plugins.RestPlugin import RestPlugin


def test_health() -> None:
    boot = PyBoot()
    boot.bootstrap("app")
    client = TestClient(boot.get_plugin(RestPlugin).app)

    response = client.get("/api/health")
    assert response.status_code == 200
    assert response.json() == {"status": "ok"}
```

## Related documentation

- [Examples](./EXAMPLES.md)
- [Production starter](./STARTER.md)
- [Core PyBoot README](../../pyboot/README.md)
- [Core PyBoot examples](../../pyboot/examples.md)

## License

MIT. See [LICENSE](./LICENSE).
