mizan/examples/fastapi-react-site/backend/main.py

"""
Example FastAPI app for the e2e harness — mirrors the surface that
examples/django-react-site/backend/testapp/clients.py exercises, minus
Django-only features (forms, channels, ws-whoami, session-bound JWT).

The fixture functions are designed to drive specific Playwright tests:
- success-path RPC (echo, add, multiply)
- auth requirements (whoami, staff_only, superuser_only, verified_only)
- error codes (not_implemented_fn, buggy_fn, permission_check_fn)
- a global context (current_user) for the bundled-fetch path

Anonymous access is the default — request.state.user is left unset so
the auth-required functions return UNAUTHORIZED, matching the harness
expectations for an anonymous browser session.
"""

from __future__ import annotations

from fastapi import FastAPI
from fastapi.exceptions import RequestValidationError
from pydantic import BaseModel

from mizan_core.client.function import client
from mizan_core.registry import register
from mizan_fastapi import (
    Forbidden,
    MizanError,
    mizan_exception_handler,
    mizan_validation_handler,
    router as mizan_router,
)


# ─── Output shapes ──────────────────────────────────────────────────────────


class EchoOutput(BaseModel):
    message: str


class AddOutput(BaseModel):
    result: int


class MultiplyOutput(BaseModel):
    product: int


class UserOutput(BaseModel):
    email: str
    authenticated: bool
    is_staff: bool = False


class MessageOutput(BaseModel):
    message: str


# ─── Fixture functions ──────────────────────────────────────────────────────


@client
def echo(request, text: str) -> EchoOutput:
    """Echoes the input text."""
    return EchoOutput(message=text)


@client
def add(request, a: int, b: int) -> AddOutput:
    """Returns a + b."""
    return AddOutput(result=a + b)


@client
def multiply(request, x: int, y: int) -> MultiplyOutput:
    """Returns x * y."""
    return MultiplyOutput(product=x * y)


@client(auth=True)
def whoami(request) -> UserOutput:
    """Returns the authenticated user's identity. Anonymous → UNAUTHORIZED."""
    user = request.state.user
    return UserOutput(
        email=getattr(user, "email", ""),
        authenticated=True,
        is_staff=getattr(user, "is_staff", False),
    )


@client(auth="staff")
def staff_only(request) -> MessageOutput:
    """Staff-only endpoint."""
    return MessageOutput(message="staff access ok")


@client(auth="superuser")
def superuser_only(request) -> MessageOutput:
    """Superuser-only endpoint."""
    return MessageOutput(message="superuser access ok")


def _is_verified(request) -> bool:
    user = getattr(getattr(request, "state", None), "user", None)
    return bool(user) and getattr(user, "is_verified", False)


@client(auth=_is_verified)
def verified_only(request) -> MessageOutput:
    """Verified-users-only endpoint. Anonymous → FORBIDDEN."""
    return MessageOutput(message="verified access ok")


@client
def not_implemented_fn(request) -> MessageOutput:
    """Always raises NotImplementedError → NOT_IMPLEMENTED."""
    raise NotImplementedError("This function is intentionally not implemented")


@client
def buggy_fn(request) -> MessageOutput:
    """Always raises a generic exception → INTERNAL_ERROR."""
    raise RuntimeError("Intentional bug for e2e testing")


@client
def permission_check_fn(request, secret: str) -> MessageOutput:
    """Wrong secret → FORBIDDEN; correct secret → success."""
    if secret != "open-sesame":
        raise Forbidden("Invalid secret")
    return MessageOutput(message="access granted")


@client(context="global")
def current_user(request) -> UserOutput:
    """The global context — auto-mounted at the React root."""
    user = getattr(getattr(request, "state", None), "user", None)
    return UserOutput(
        email=getattr(user, "email", "") if user else "",
        authenticated=bool(user) and getattr(user, "is_authenticated", False),
        is_staff=getattr(user, "is_staff", False) if user else False,
    )


# ─── Merge protocol fixtures ────────────────────────────────────────────────


class MorphGroupMeta(BaseModel):
    """Group summary — narrower shape than MorphLayer. Listed alongside
    morph_layers so the server's slot resolver has to discriminate by
    return-type rather than by bundle order."""
    id: int
    label: str
    count: int


class MorphLayer(BaseModel):
    id: int
    group_id: int
    label: str
    value: float


_morph_groups: list[MorphGroupMeta] = [
    MorphGroupMeta(id=1, label="face", count=2),
]


_morph_layers: list[MorphLayer] = [
    MorphLayer(id=1, group_id=1, label="brow", value=0.0),
    MorphLayer(id=2, group_id=1, label="jaw", value=0.0),
]


@client(context="morphs")
def morph_groups(request) -> list[MorphGroupMeta]:
    """Summary-shape slot — server must route MorphLayer mutations away from here."""
    return list(_morph_groups)


@client(context="morphs")
def morph_layers(request) -> list[MorphLayer]:
    """Detailed-shape slot — server routes MorphLayer mutations here."""
    return list(_morph_layers)


@client(merge="morphs")
def set_morph_value(request, id: int, value: float) -> MorphLayer:
    """Mutation that returns the changed row; kernel splices into morph_layers."""
    for layer in _morph_layers:
        if layer.id == id:
            layer.value = value
            return layer
    raise ValueError(f"unknown morph layer id={id}")


# ─── Registration ───────────────────────────────────────────────────────────


register(echo, "echo")
register(add, "add")
register(multiply, "multiply")
register(whoami, "whoami")
register(staff_only, "staff_only")
register(superuser_only, "superuser_only")
register(verified_only, "verified_only")
register(not_implemented_fn, "not_implemented_fn")
register(buggy_fn, "buggy_fn")
register(permission_check_fn, "permission_check_fn")
register(current_user, "current_user")
register(morph_groups, "morph_groups")
register(morph_layers, "morph_layers")
register(set_morph_value, "set_morph_value")


# ─── App ────────────────────────────────────────────────────────────────────


app = FastAPI(title="mizan-fastapi e2e example")
app.include_router(mizan_router, prefix="/api/mizan")
app.add_exception_handler(MizanError, mizan_exception_handler)
app.add_exception_handler(RequestValidationError, mizan_validation_handler)