Application levels

An Ravyn application is composed by levels and those levels can be Gateway, WebSocketGateway, Include, handlers or even another Ravyn or ChildRavyn.

There are some levels in the application, let's use an example.

from ravyn import Ravyn, Gateway, Include, Request, get


@get("/me")
async def me(request: Request) -> str:
    return "Hello, world!"


app = Ravyn(routes=[Include("/", routes=[Gateway(handler=me)])])

Levels:

1. Ravyn - The application instance.
2. Include - The second level.
3. Gateway - The third level, inside an include.
4. Handler - The forth level, the Gateway handler.

You can create as many levels as you desire. From nested includes to ChildRavyn and create your own design.

With a ChildRavyn

from pydantic import BaseModel, EmailStr

from ravyn import (
    Controller,
    ChildRavyn,
    Ravyn,
    Gateway,
    Include,
    Request,
    WebSocket,
    WebSocketGateway,
    get,
    post,
    websocket,
)


@get("/me")
async def me(request: Request) -> str:
    return "Hello, world!"


@websocket(path="/ws")
async def websocket_endpoint_include(socket: WebSocket) -> None:
    await socket.accept()
    await socket.send_text("Hello, new world!")
    await socket.close()


class User(BaseModel):
    name: str
    email: EmailStr
    password: str


class UserApiView(Controller):
    path = "/users"

    @post("/create")
    async def create_user(self, data: User, request: Request) -> None: ...

    @websocket(path="/ws")
    async def websocket_endpoint(self, socket: WebSocket) -> None:
        await socket.accept()
        await socket.send_text("Hello, world!")
        await socket.close()


child_ravyn = ChildRavyn(routes=[Gateway(handler=UserApiView)])

app = Ravyn(
    routes=[
        Include(
            "/",
            routes=[
                Gateway(handler=me),
                WebSocketGateway(handler=websocket_endpoint_include),
                Include("/child", child_ravyn),
            ],
        )
    ]
)

Levels:

1. Ravyn - The application instance. First Level.
2. Gateway - The second level, inside the app routes.
   1. Handler - The third level, inside the Gateway.
3. WebSocketGateway - The second level, inside the app routes.
   1. Handler - The third level, inside the WebSocketGateway.
4. Include - The second level. Inside the app routes.
   1. ChildRavyn - The third level inside the include and also first level as independent instance.
      1. Gateway - The second level, inside the ChildRavyn.
         1. Handler - The second level, inside the Gateway.

Warning

A ChildRavyn is an independent instance that is plugged into a main Ravyn application, but since it is like another Ravyn object, that also means the ChildRavyn does not take precedence over the top-level application, instead, treats its own Gateway, WebSocketGateway, Include, handlers or even another Ravyn or ChildRavyn and parameters in isolation.

Exceptions

ChildRavyn, as per warning above, has its own rules, but there are always exceptions to any almost every rule. Although it is an independent instance with its own rules, this is not applied to every parameter.

Middlewares and Permissions are actually global and the rules of precedence can be applied between an Ravyn instance and the corresponding ChildRavyn apps.

In other words, you don't need to create/repeat the same permissions and middlewares (common to both) across every instance. They can be applied globally from the top main Ravyn object.

from pydantic import BaseModel, EmailStr

from ravyn import (
    Controller,
    ChildRavyn,
    Ravyn,
    Gateway,
    Include,
    Request,
    WebSocket,
    WebSocketGateway,
    get,
    post,
    settings,
    websocket,
)
from ravyn.core.config.jwt import JWTConfig
from ravyn.contrib.auth.edgy.base_user import User
from ravyn.exceptions import NotAuthorized
from ravyn.middleware.authentication import AuthResult, BaseAuthMiddleware
from ravyn.permissions import IsAdminUser
from ravyn.security.jwt.token import Token
from lilya._internal._connection import Connection
from lilya.middleware import DefineMiddleware as LilyaMiddleware
from lilya.types import ASGIApp
from edgy.exceptions import ObjectNotFound


class JWTAuthMiddleware(BaseAuthMiddleware):
    def __init__(self, app: "ASGIApp", config: "JWTConfig"):
        super().__init__(app)
        self.app = app
        self.config = config

    async def retrieve_user(self, user_id) -> User:
        try:
            return await User.get(pk=user_id)
        except ObjectNotFound:
            raise NotAuthorized()

    async def authenticate(self, request: Connection) -> AuthResult:
        token = request.headers.get(self.config.api_key_header)

        if not token:
            raise NotAuthorized("JWT token not found.")

        token = Token.decode(
            token=token, key=self.config.signing_key, algorithm=self.config.algorithm
        )

        user = await self.retrieve_user(token.sub)
        return AuthResult(user=user)


class IsAdmin(IsAdminUser):
    def is_user_staff(self, request: "Request") -> bool:
        """
        Add logic to verify if a user is staff
        """


@get()
async def home() -> None: ...


@get("/me")
async def me(request: Request) -> str:
    return "Hello, world!"


@websocket(path="/ws")
async def websocket_endpoint_include(socket: WebSocket) -> None:
    await socket.accept()
    await socket.send_text("Hello, new world!")
    await socket.close()


class User(BaseModel):
    name: str
    email: EmailStr
    password: str


class UserApiView(Controller):
    path = "/users"

    @post("/create")
    async def create_user(self, data: User, request: Request) -> None: ...

    @websocket(path="/ws")
    async def websocket_endpoint(self, socket: WebSocket) -> None:
        await socket.accept()
        await socket.send_text("Hello, world!")
        await socket.close()


child_ravyn = ChildRavyn(routes=[Gateway("/home", handler=home), Gateway(handler=UserApiView)])

jwt_config = JWTConfig(
    signing_key=settings.secret_key,
)


app = Ravyn(
    routes=[
        Include(
            "/",
            routes=[
                Gateway(handler=me),
                WebSocketGateway(handler=websocket_endpoint_include),
                Include("/admin", child_ravyn),
            ],
        )
    ],
    permissions=[IsAdmin],
    middleware=[LilyaMiddleware(JWTAuthMiddleware, config=jwt_config)],
)

Notes

The example given is intentionally big and "complex" simply to show that even with that complexity in place, the middleware and permissions remained global to the whole application without the need to implement on both Ravyn and ChildRavyn.