Settings in Selva are handled through YAML files. +Internally it uses strictyaml to parse the +yaml files in order to do the parsing in a safe and predictable way.
+Settings files are located by default in the configuration directory with the
+base name settings.yaml:
project/
+├── application/
+│ └── ...
+└── configuration/
+ ├── settings.yaml
+ ├── settings_dev.yaml
+ └── settings_prod.yaml
+The configuration values can be accessed by injecting selva.configuration.Settings.
from typing import Annotated
+from selva.configuration import Settings
+from selva.di import Inject, service
+
+
+@service
+class MyService:
+ settings: Annotated[Settings, Inject]
+The selva.configuration.Settings is a dict like object that can also be accessed
+using property syntax:
from selva.configuration import Settings
+
+settings = Settings({"config": "value"})
+assert settings["config"] == "value"
+assert settings.config == "value"
+Since strictyaml is used to parse the yaml files, all values strs. However, we
+can use pydantic and Selva dependency injection system to provide access to the
+settings in a typed manner:
from pydantic import BaseModel
+from selva.configuration import Settings
+from selva.di import service
+
+
+class MySettings(BaseModel):
+ int_property: int
+ bool_property: bool
+
+
+@service
+def my_settings(settings: Settings) -> MySettings:
+ return MySettings.model_validate(settings.my_settings)
+The settings files can include references to environment variables that takes the
+format ${ENV_VAR:default_value}. The default value is optional and an error will
+be raised if neither the environment variable nor the default value are defined.
required: ${ENV_VAR} # required environment variable
+optional: ${OPT_VAR:default} # optional environment variable
+Optional profiles can be activated by settings the environment variable SELVA_PROFILE.
+The framework will look for a file named settings_${SELVA_PROFILE}.yaml and merge
+the values with the main settings.yaml. Values from the profile settings take
+precedence over the values from the main settings.
As an example, if we define SELVA_PROFILE=dev, the file settings_dev.yaml will
+be loaded. If instead we define SELVA_PROFILE=prod, then the file settings_prod.yaml
+will be loaded.
Settings can also be defined with environment variables whose names start with SELVA__,
+where subsequent double undercores (__) indicates nesting (variable is a mapping).
+Also, variable names will be lowercased.
For example, consider the following environment variables:
+ +Those variables will be collected as the following:
+ +If running you project using selva.run.app, for example uvicorn selva.run:app,
+environment variables can be loaded from a .env file. The parsing is done using
+the python-dotenv library.
By default, a .env file in the current working directory will be loaded, but it
+can be customized with the environment variable SELVA_DOTENV pointing to a .env file.
Controllers are classes responsible for handling requests through handler methods.
+They are defined using the @controller on the class and @get, @post, @put,
+@patch, @delete and @websocket on each of the handler methods.
Handler methods must receive, at least, two parameters: Request and Response.
+It is not needed to annotate the request and response parameters, but they should
+be the first two parameters.
from asgikit.requests import Request, read_json
+from asgikit.responses import respond_text, respond_redirect
+from selva.web import controller, get, post
+from loguru import logger
+
+
+@controller
+class IndexController:
+ @get
+ async def index(self, request: Request):
+ await respond_text(request.response, "application root")
+
+
+@controller("admin")
+class AdminController:
+ @post("send")
+ async def handle_data(self, request: Request):
+ logger.info(await read_json(request))
+ await respond_redirect(request.response, "/")
+Note
+Defining a path on @controller or @get @post etc... is optional and
+defaults to an empty string "".
Handler methods can be defined with path parameters, which can be bound to the
+handler with the annotation FromPath:
from typing import Annotated
+from selva.web.converter import FromPath
+from selva.web.routing.decorator import get
+
+
+@get("/:path_param")
+def handler(request, path_param: Annotated[str, FromPath]):
+ ...
+It is also possible to explicitly declare from which parameter the value will +be retrieved from:
+ +The routing section provides more information about path parameters
+Controllers themselves are services, and therefore can have services injected.
+from typing import Annotated
+from selva.di import service, Inject
+from selva.web import controller
+
+
+@service
+class MyService:
+ pass
+
+
+@controller
+class MyController:
+ my_service: Annotated[MyService, Inject]
+Handler methods receive an object of type asgikit.requests.Request as the first
+parameter that provides access to request information (path, method, headers, query
+string, request body). It also provides the asgikit.responses.Response or
+asgikit.websockets.WebSocket objects to either respond the request or interact
+with the client via websocket.
Attention
+For http requests, Request.websocket will be None, and for
+websocket requests, Request.response will be None
from http import HTTPMethod, HTTPStatus
+from asgikit.requests import Request
+from asgikit.responses import respond_json
+from selva.web import controller, get, websocket
+
+
+@controller
+class MyController:
+ @get
+ async def handler(self, request: Request):
+ assert request.response is not None
+ assert request.websocket is None
+
+ assert request.method == HTTPMethod.GET
+ assert request.path == "/"
+ await respond_json(request.response, {"status": HTTPStatus.OK})
+
+ @websocket
+ async def ws_handler(self, request: Request):
+ assert request.response is None
+ assert request.websocket is not None
+
+ ws = request.websocket
+ await ws.accept()
+ while True:
+ data = await ws.receive()
+ await ws.send(data)
+asgikit provides several functions to retrieve the request body:
async def read_body(request: Request) -> bytes
+async def read_text(request: Request, encoding: str = None) -> str
+async def read_json(request: Request) -> dict | list
+async def read_form(request: Request) -> dict[str, str | multipart.File]:
+For websocket, there are the following methods:
+async def accept(subprotocol: str = None, headers: Iterable[tuple[bytes, bytes]] = None)
+async def receive(self) -> str | bytes
+async def send(self, data: bytes | str)
+async def close(self, code: int = 1000, reason: str = "")
+Handler methods can receive additional parameters, which will be extracted from
+the request using an implementation of selva.web.FromRequest[Type].
+If there is no direct implementation of FromRequest[Type], Selva will iterate
+over the base types of Type until an implementation is found or an error will
+be returned if there is none.
You can use the register_from_request decorator to register an FromRequest implementation.
from asgikit.requests import Request
+from asgikit.responses import respond_text
+from selva.web import controller, get
+from selva.web.converter.decorator import register_from_request
+
+
+class Param:
+ def __init__(self, path: str):
+ self.request_path = path
+
+
+@register_from_request(Param)
+class ParamFromRequest:
+ def from_request(
+ self,
+ request: Request,
+ original_type,
+ parameter_name,
+ metadata = None,
+ ) -> Param:
+ return Param(request.path)
+
+
+@controller
+class MyController:
+ @get
+ async def handler(self, request: Request, param: Param):
+ await respond_text(request.response, param.request_path)
+If the FromRequest implementation raise an error, the handler is not called.
+And if the error is a subclass of selva.web.error.HTTPError, for instance
+UnathorizedError, a response will be produced according to the error.
from selva.web.exception import HTTPUnauthorizedException
+
+
+@register_from_request(Param)
+class ParamFromRequest:
+ def from_request(
+ self,
+ request: Request,
+ original_type,
+ parameter_name,
+ metadata = None,
+ ) -> Param:
+ if "authorization" not in request.headers:
+ raise HTTPUnauthorizedException()
+ return Param(context.path)
+Selva already implements FromRequest[pydantic.BaseModel] by reading the request
+body and parsing the input into the pydantic model, if the content type is json
+or form, otherwise raising an HTTPError with status code 415. It is also implemented
+for list[pydantic.BaseModel].
Inheriting the asgikit.responses.Response from asgikit, the handler methods
+do not return a response, instead they write to the response.
Selva is a tool for creating ASGI applications that are easy to build and maintain.
+It is built on top of asgikit and comes with +a dependency injection system built upon Python type annotations. It is compatible with python 3.11+.
+Install selva and uvicorn:
Create file application.py:
from asgikit.requests import Request
+from asgikit.responses import respond_text
+from selva.web import controller, get
+
+
+@controller
+class Controller:
+ @get
+ async def hello(self, request: Request):
+ await respond_text(request.response, "Hello, World")
+Run application with uvicorn. Selva will automatically load application.py:
Selva uses loguru for logging, but provides +some facilities on top of it to make its usage a bit closer to other frameworks +like Spring Boot.
+First, an interceptor to the standard logging module is configured by default,
+as suggested in https://github.com/Delgan/loguru#entirely-compatible-with-standard-logging.
Second, a custom logging filter is provided in order to set the logging level for +each package independently.
+Logging is configured in the Selva configuration:
+logging:
+ root: WARNING
+ level:
+ application: INFO
+ application.service: TRACE
+ sqlalchemy: DEBUG
+ enable:
+ - packages_to_activate_logging
+ disabled:
+ - packages_to_deactivate_logging
+The root property is the root level. It is used if no other level is set for the
+package where the log comes from.
The level property defines the logging level for each package independently.
The enable and disable properties lists the packages to enable or disable logging.
+This comes from loguru, as can be seen in https://github.com/Delgan/loguru#suitable-for-scripts-and-libraries.
If you want full control of how loguru is configured, you can provide a logger setup +function and reference it in the configuration file:
+The setup function receives a parameter of type selva.configuration.Settings,
+so you can have access to the whole settings.
The middleware pipeline is configured with the MIDDLEWARE configuration property. It must contain a list of classes
+that inherit from selva.web.middleware.Middleware.
To demonstrate the middleware system, we will create a timing middleware that will output to the console the time spent +in the processing of the request:
+from collections.abc import Callable
+from datetime import datetime
+
+from asgikit.requests import Request
+from selva.web.middleware import Middleware
+from loguru import logger
+
+
+class TimingMiddleware(Middleware):
+ async def __call__(self, chain: Callable, request: Request):
+ request_start = datetime.now()
+ await chain(request) # (1)
+ request_end = datetime.now()
+
+ delta = request_end - request_start
+ logger.info("Request time: {}", delta)
+Middleware instances are created using the same machinery as services, and therefore
+can have services of their own. Our TimingMiddleware, for instance, could persist
+the timings using a service instead of printing to the console:
from collections.abc import Callable
+from datetime import datetime
+
+from asgikit.requests import Request
+from selva.di import Inject
+from selva.web.middleware import Middleware
+
+from application.service import TimingService
+
+
+class TimingMiddleware(Middleware):
+ timing_service: TimingService = Inject()
+
+ async def __call__(self, chain: Callable, request: Request):
+ request_start = datetime.now()
+ await chain(request)
+ request_end = datetime.now()
+
+ await self.timing_service.save(request_start, request_end)
+Routing is defined by the decorators in the controllers and handlers.
+Parameters can be defined in the handler's path using the syntax :parameter_name,
+where parameter_name must be the name of the argument on the handler's signature.
from typing import Annotated
+from asgikit.requests import Request
+from asgikit.responses import respond_text
+from selva.web import controller, get, FromPath
+
+
+@controller
+class Controller:
+ @get("hello/:name")
+ async def handler(self, request: Request, name: Annotated[str, FromPath]):
+ await respond_text(request.response, f"Hello, {name}!")
+Here was used Annotated and FromPath to indicated that the handler argument
+is to be bound to the parameter from the request path. More on that will be explained
+in the following sections.
The default behavior is for a path parameter to match a single path segment.
+If you want to match the whole path, or a subpath of the request path,
+use the syntax *parameter_name.
from typing import Annotated
+from asgikit.requests import Request
+from asgikit.responses import respond_text
+from selva.web import controller, get, FromPath
+
+
+@controller
+class Controller:
+ @get("hello/*path")
+ async def handler(self, request: Request, path: Annotated[str, FromPath]):
+ name = " ".join(path.split("/"))
+ await respond_text(request.response, f"Hello, {name}!")
+For a request like GET hello/Python/World, the handler will output
+Hello, Python World!.
You can mix both types of parameters with no problem:
+*path*path/literal_segment:normal_param/*path:normal_param/*path/:other_pathParameter conversion is done through the type annotation on the parameter. Selva +will try to find a converter suitable for the parameter type and then convert +the value before calling the handler method.
+from typing import Annotated
+from asgikit.requests import Request
+from asgikit.responses import respond_json
+from selva.web import controller, get, FromPath
+
+
+@controller
+class Controller:
+ @get("repeat/:amount")
+ async def handler(self, request: Request, amount: Annotated[int, FromPath]):
+ await respond_json(request.response, {f"repeat {i}": i for i in range(amount)})
+The type annotation indicates that we want a value of type int that should be
+taken from the request path.
Selva already provide converters for the types str, int, float, bool and pathlib.PurePath.
Conversion can be customized by providing an implementing of selva.web.converter.param_converter.ParamConverter.
+You normally use the shortcut decorator selva.web.converter.decodator.register_param_converter.
from dataclasses import dataclass
+from typing import Annotated
+
+from asgikit.requests import Request
+from asgikit.responses import respond_text
+from selva.web import controller, get, FromPath
+from selva.web.converter.decorator import register_param_converter
+
+
+@dataclass
+class MyModel:
+ name: str
+
+
+@register_param_converter(MyModel)
+class MyModelParamConverter:
+ def from_str(self, value: str) -> MyModel:
+ return MyModel(value)
+
+
+@controller
+class MyController:
+ @get("/:model")
+ async def handler(self, request: Request, model: Annotated[MyModel, FromPath]):
+ await respond_text(request.response, str(model))
+If the ParamConverter implementation raise an error, the handler is not called.
+And if the error is a subclass of selva.web.error.HTTPError, for instance
+UnathorizedError, a response will be produced according to the error.
The ParamConverter can also be provided a method called into_str(self, obj) -> str
+that is used to convert the object back. This is used to build urls from routes.
+If not implemented, the default calls str on the object.
Selva is a tool for creating ASGI applications that are easy to build and maintain.
It is built on top of asgikit and comes with a dependency injection system built upon Python type annotations. It is compatible with python 3.11+.
"},{"location":"#quickstart","title":"Quickstart","text":"Install selva and uvicorn:
pip install selva uvicorn[standard]\nCreate file application.py:
from asgikit.requests import Request\nfrom asgikit.responses import respond_text\nfrom selva.web import controller, get\n@controller\nclass Controller:\n@get\nasync def hello(self, request: Request):\nawait respond_text(request.response, \"Hello, World\")\nRun application with uvicorn. Selva will automatically load application.py:
uvicorn selva.run:app\nINFO: Started server process [18664]\nINFO: Waiting for application startup.\nINFO: Application startup complete.\nINFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)\nSettings in Selva are handled through YAML files. Internally it uses strictyaml to parse the yaml files in order to do the parsing in a safe and predictable way.
Settings files are located by default in the configuration directory with the base name settings.yaml:
project/\n\u251c\u2500\u2500 application/\n\u2502 \u2514\u2500\u2500 ...\n\u2514\u2500\u2500 configuration/\n \u251c\u2500\u2500 settings.yaml\n \u251c\u2500\u2500 settings_dev.yaml\n \u2514\u2500\u2500 settings_prod.yaml\nThe configuration values can be accessed by injecting selva.configuration.Settings.
from typing import Annotated\nfrom selva.configuration import Settings\nfrom selva.di import Inject, service\n@service\nclass MyService:\nsettings: Annotated[Settings, Inject]\nThe selva.configuration.Settings is a dict like object that can also be accessed using property syntax:
from selva.configuration import Settings\nsettings = Settings({\"config\": \"value\"})\nassert settings[\"config\"] == \"value\"\nassert settings.config == \"value\"\nSince strictyaml is used to parse the yaml files, all values strs. However, we can use pydantic and Selva dependency injection system to provide access to the settings in a typed manner:
from pydantic import BaseModel\nfrom selva.configuration import Settings\nfrom selva.di import service\nclass MySettings(BaseModel):\nint_property: int\nbool_property: bool\n@service\ndef my_settings(settings: Settings) -> MySettings:\nreturn MySettings.model_validate(settings.my_settings)\nmy_settings:\nint_property: 1\nbool_property: true\nThe settings files can include references to environment variables that takes the format ${ENV_VAR:default_value}. The default value is optional and an error will be raised if neither the environment variable nor the default value are defined.
required: ${ENV_VAR} # required environment variable\noptional: ${OPT_VAR:default} # optional environment variable\nOptional profiles can be activated by settings the environment variable SELVA_PROFILE. The framework will look for a file named settings_${SELVA_PROFILE}.yaml and merge the values with the main settings.yaml. Values from the profile settings take precedence over the values from the main settings.
As an example, if we define SELVA_PROFILE=dev, the file settings_dev.yaml will be loaded. If instead we define SELVA_PROFILE=prod, then the file settings_prod.yaml will be loaded.
Settings can also be defined with environment variables whose names start with SELVA__, where subsequent double undercores (__) indicates nesting (variable is a mapping). Also, variable names will be lowercased.
For example, consider the following environment variables:
SELVA__PROPERTY=1\nSELVA__MAPPING__PROPERTY=2\nSELVA__MAPPING__ANOTHER_PROPERTY=3\nThose variables will be collected as the following:
{\n\"property\": \"1\",\n\"mapping\": {\n\"property\": \"2\",\n\"another_property\": \"3\",\n},\n}\nIf running you project using selva.run.app, for example uvicorn selva.run:app, environment variables can be loaded from a .env file. The parsing is done using the python-dotenv library.
By default, a .env file in the current working directory will be loaded, but it can be customized with the environment variable SELVA_DOTENV pointing to a .env file.
Controllers are classes responsible for handling requests through handler methods. They are defined using the @controller on the class and @get, @post, @put, @patch, @delete and @websocket on each of the handler methods.
Handler methods must receive, at least, two parameters: Request and Response. It is not needed to annotate the request and response parameters, but they should be the first two parameters.
from asgikit.requests import Request, read_json\nfrom asgikit.responses import respond_text, respond_redirect\nfrom selva.web import controller, get, post\nfrom loguru import logger\n@controller\nclass IndexController:\n@get\nasync def index(self, request: Request):\nawait respond_text(request.response, \"application root\")\n@controller(\"admin\")\nclass AdminController:\n@post(\"send\")\nasync def handle_data(self, request: Request):\nlogger.info(await read_json(request))\nawait respond_redirect(request.response, \"/\")\nNote
Defining a path on @controller or @get @post etc... is optional and defaults to an empty string \"\".
Handler methods can be defined with path parameters, which can be bound to the handler with the annotation FromPath:
from typing import Annotated\nfrom selva.web.converter import FromPath\nfrom selva.web.routing.decorator import get\n@get(\"/:path_param\")\ndef handler(request, path_param: Annotated[str, FromPath]):\n...\nIt is also possible to explicitly declare from which parameter the value will be retrieved from:
@get(\"/:path_param\")\ndef handler(req, res, value: Annotated[str, FromPath(\"path_param\")]):\n...\nThe routing section provides more information about path parameters
"},{"location":"controllers/#dependencies","title":"Dependencies","text":"Controllers themselves are services, and therefore can have services injected.
from typing import Annotated\nfrom selva.di import service, Inject\nfrom selva.web import controller\n@service\nclass MyService:\npass\n@controller\nclass MyController:\nmy_service: Annotated[MyService, Inject]\nHandler methods receive an object of type asgikit.requests.Request as the first parameter that provides access to request information (path, method, headers, query string, request body). It also provides the asgikit.responses.Response or asgikit.websockets.WebSocket objects to either respond the request or interact with the client via websocket.
Attention
For http requests, Request.websocket will be None, and for websocket requests, Request.response will be None
from http import HTTPMethod, HTTPStatus\nfrom asgikit.requests import Request\nfrom asgikit.responses import respond_json\nfrom selva.web import controller, get, websocket\n@controller\nclass MyController:\n@get\nasync def handler(self, request: Request):\nassert request.response is not None\nassert request.websocket is None\nassert request.method == HTTPMethod.GET\nassert request.path == \"/\"\nawait respond_json(request.response, {\"status\": HTTPStatus.OK})\n@websocket\nasync def ws_handler(self, request: Request):\nassert request.response is None\nassert request.websocket is not None\nws = request.websocket\nawait ws.accept()\nwhile True:\ndata = await ws.receive()\nawait ws.send(data)\nasgikit provides several functions to retrieve the request body:
async def read_body(request: Request) -> bytes\nasync def read_text(request: Request, encoding: str = None) -> str\nasync def read_json(request: Request) -> dict | list\nasync def read_form(request: Request) -> dict[str, str | multipart.File]:\nFor websocket, there are the following methods:
async def accept(subprotocol: str = None, headers: Iterable[tuple[bytes, bytes]] = None)\nasync def receive(self) -> str | bytes\nasync def send(self, data: bytes | str)\nasync def close(self, code: int = 1000, reason: str = \"\")\nHandler methods can receive additional parameters, which will be extracted from the request using an implementation of selva.web.FromRequest[Type]. If there is no direct implementation of FromRequest[Type], Selva will iterate over the base types of Type until an implementation is found or an error will be returned if there is none.
You can use the register_from_request decorator to register an FromRequest implementation.
from asgikit.requests import Request\nfrom asgikit.responses import respond_text\nfrom selva.web import controller, get\nfrom selva.web.converter.decorator import register_from_request\nclass Param:\ndef __init__(self, path: str):\nself.request_path = path\n@register_from_request(Param)\nclass ParamFromRequest:\ndef from_request(\nself,\nrequest: Request,\noriginal_type,\nparameter_name,\nmetadata = None,\n) -> Param:\nreturn Param(request.path)\n@controller\nclass MyController:\n@get\nasync def handler(self, request: Request, param: Param):\nawait respond_text(request.response, param.request_path)\nIf the FromRequest implementation raise an error, the handler is not called. And if the error is a subclass of selva.web.error.HTTPError, for instance UnathorizedError, a response will be produced according to the error.
from selva.web.exception import HTTPUnauthorizedException\n@register_from_request(Param)\nclass ParamFromRequest:\ndef from_request(\nself,\nrequest: Request,\noriginal_type,\nparameter_name,\nmetadata = None,\n) -> Param:\nif \"authorization\" not in request.headers:\nraise HTTPUnauthorizedException()\nreturn Param(context.path)\nSelva already implements FromRequest[pydantic.BaseModel] by reading the request body and parsing the input into the pydantic model, if the content type is json or form, otherwise raising an HTTPError with status code 415. It is also implemented for list[pydantic.BaseModel].
Inheriting the asgikit.responses.Response from asgikit, the handler methods do not return a response, instead they write to the response.
from asgikit.requests import Request\nfrom asgikit.responses import respond_text\nfrom selva.web import controller, get\n@controller\nclass Controller:\n@get\nasync def handler(self, request: Request):\nawait respond_text(request.response, \"Ok\")\nSelva uses loguru for logging, but provides some facilities on top of it to make its usage a bit closer to other frameworks like Spring Boot.
First, an interceptor to the standard logging module is configured by default, as suggested in https://github.com/Delgan/loguru#entirely-compatible-with-standard-logging.
Second, a custom logging filter is provided in order to set the logging level for each package independently.
"},{"location":"logging/#configuring-logging","title":"Configuring logging","text":"Logging is configured in the Selva configuration:
logging:\nroot: WARNING\nlevel:\napplication: INFO\napplication.service: TRACE\nsqlalchemy: DEBUG\nenable:\n- packages_to_activate_logging\ndisabled:\n- packages_to_deactivate_logging\nThe root property is the root level. It is used if no other level is set for the package where the log comes from.
The level property defines the logging level for each package independently.
The enable and disable properties lists the packages to enable or disable logging. This comes from loguru, as can be seen in https://github.com/Delgan/loguru#suitable-for-scripts-and-libraries.
If you want full control of how loguru is configured, you can provide a logger setup function and reference it in the configuration file:
application/logging.pyconfiguration/settings.yamlfrom loguru import logger\ndef setup(settings):\nlogger.configure(...)\nlogging:\nsetup: application.logging.setup\nThe setup function receives a parameter of type selva.configuration.Settings, so you can have access to the whole settings.
The middleware pipeline is configured with the MIDDLEWARE configuration property. It must contain a list of classes that inherit from selva.web.middleware.Middleware.
To demonstrate the middleware system, we will create a timing middleware that will output to the console the time spent in the processing of the request:
application/controller.pyapplication/middleware.pyconfiguration/settings.yamlfrom asgikit.requests import Request\nfrom asgikit.responses import respond_json\nfrom selva.web import controller, get\n@controller\nclass HelloController:\n@get\nasync def hello(self, request: Request):\nawait respond_json(request.response, {\"greeting\": \"Hello, World!\"})\nfrom collections.abc import Callable\nfrom datetime import datetime\nfrom asgikit.requests import Request\nfrom selva.web.middleware import Middleware\nfrom loguru import logger\nclass TimingMiddleware(Middleware):\nasync def __call__(self, chain: Callable, request: Request):\nrequest_start = datetime.now()\nawait chain(request) # (1)\nrequest_end = datetime.now()\ndelta = request_end - request_start\nlogger.info(\"Request time: {}\", delta)\nmiddleware:\n- application.middleware.TimingMiddleware\nMiddleware instances are created using the same machinery as services, and therefore can have services of their own. Our TimingMiddleware, for instance, could persist the timings using a service instead of printing to the console:
from datetime import datetime\nfrom selva.di import service\n@service\nclass TimingService:\nasync def save(start: datetime, end: datetime):\n...\nfrom collections.abc import Callable\nfrom datetime import datetime\nfrom asgikit.requests import Request\nfrom selva.di import Inject\nfrom selva.web.middleware import Middleware\nfrom application.service import TimingService\nclass TimingMiddleware(Middleware):\ntiming_service: TimingService = Inject()\nasync def __call__(self, chain: Callable, request: Request):\nrequest_start = datetime.now()\nawait chain(request)\nrequest_end = datetime.now()\nawait self.timing_service.save(request_start, request_end)\nRouting is defined by the decorators in the controllers and handlers.
"},{"location":"routing/#path-parameters","title":"Path parameters","text":"Parameters can be defined in the handler's path using the syntax :parameter_name, where parameter_name must be the name of the argument on the handler's signature.
from typing import Annotated\nfrom asgikit.requests import Request\nfrom asgikit.responses import respond_text\nfrom selva.web import controller, get, FromPath\n@controller\nclass Controller:\n@get(\"hello/:name\")\nasync def handler(self, request: Request, name: Annotated[str, FromPath]):\nawait respond_text(request.response, f\"Hello, {name}!\")\nHere was used Annotated and FromPath to indicated that the handler argument is to be bound to the parameter from the request path. More on that will be explained in the following sections.
The default behavior is for a path parameter to match a single path segment. If you want to match the whole path, or a subpath of the request path, use the syntax *parameter_name.
from typing import Annotated\nfrom asgikit.requests import Request\nfrom asgikit.responses import respond_text\nfrom selva.web import controller, get, FromPath\n@controller\nclass Controller:\n@get(\"hello/*path\")\nasync def handler(self, request: Request, path: Annotated[str, FromPath]):\nname = \" \".join(path.split(\"/\"))\nawait respond_text(request.response, f\"Hello, {name}!\")\nFor a request like GET hello/Python/World, the handler will output Hello, Python World!.
You can mix both types of parameters with no problem:
*path*path/literal_segment:normal_param/*path:normal_param/*path/:other_pathParameter conversion is done through the type annotation on the parameter. Selva will try to find a converter suitable for the parameter type and then convert the value before calling the handler method.
from typing import Annotated\nfrom asgikit.requests import Request\nfrom asgikit.responses import respond_json\nfrom selva.web import controller, get, FromPath\n@controller\nclass Controller:\n@get(\"repeat/:amount\")\nasync def handler(self, request: Request, amount: Annotated[int, FromPath]):\nawait respond_json(request.response, {f\"repeat {i}\": i for i in range(amount)})\nThe type annotation indicates that we want a value of type int that should be taken from the request path.
Selva already provide converters for the types str, int, float, bool and pathlib.PurePath.
Conversion can be customized by providing an implementing of selva.web.converter.param_converter.ParamConverter. You normally use the shortcut decorator selva.web.converter.decodator.register_param_converter.
from dataclasses import dataclass\nfrom typing import Annotated\nfrom asgikit.requests import Request\nfrom asgikit.responses import respond_text\nfrom selva.web import controller, get, FromPath\nfrom selva.web.converter.decorator import register_param_converter\n@dataclass\nclass MyModel:\nname: str\n@register_param_converter(MyModel)\nclass MyModelParamConverter:\ndef from_str(self, value: str) -> MyModel:\nreturn MyModel(value)\n@controller\nclass MyController:\n@get(\"/:model\")\nasync def handler(self, request: Request, model: Annotated[MyModel, FromPath]):\nawait respond_text(request.response, str(model))\nIf the ParamConverter implementation raise an error, the handler is not called. And if the error is a subclass of selva.web.error.HTTPError, for instance UnathorizedError, a response will be produced according to the error.
The ParamConverter can also be provided a method called into_str(self, obj) -> str that is used to convert the object back. This is used to build urls from routes. If not implemented, the default calls str on the object.
Let's dig a little deeper and learn the basic concepts of Selva.
We will create a greeting api that logs the greet requests.
"},{"location":"tutorial/#installing-selva","title":"Installing Selva","text":"Before going any further, we need to install Selva and Uvicorn.
pip install selva uvicorn\nA selva application is structured like the following:
project/\n\u251c\u2500\u2500 application/\n\u2502 \u251c\u2500\u2500 __init__.py\n\u2502 \u251c\u2500\u2500 controller.py\n\u2502 \u251c\u2500\u2500 repository.py\n\u2502 \u2514\u2500\u2500 service.py\n\u251c\u2500\u2500 configuration/\n\u2502 \u2514\u2500\u2500 settings.yaml\n\u2514\u2500\u2500 resources/\nAnd... that's it! A module or package named application will automatically be imported and scanned for controllers and services.
We will use uvicorn to run the application and automatically reload when we make changes to the code:
$ uvicorn selva.run:app --reload\nINFO: Will watch for changes in these directories: ['/home/user/projects/selva-tutorial']\nINFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)\nINFO: Started reloader process [23568] using WatchFiles\nINFO: Started server process [19472]\nINFO: Waiting for application startup.\nINFO: Application startup complete.\nController classes hold handler methods that will respond to HTTP or WebSocket requests. They can receive services through dependency injection.
application/controller.pyfrom typing import Annotated\nfrom asgikit.requests import Request\nfrom asgikit.responses import respond_json\nfrom selva.web import controller, get, FromPath\n@controller # (1)\nclass GreetingController:\n@get(\"hello/:name\") # (2)\nasync def hello(self, request: Request, name: Annotated[str, FromPath]):\nawait respond_json(request.response, {\"greeting\": f\"Hello, {name}!\"})\n@controller marks a class as a controller. It can optionally receive a path (e.g. @controller(\"path\")) that will be prepended to the handlers' path.
@get(\"hello/:name\") defines the method as a handler on the given path. If no path is given, the path from the controller will be used.
:name defines a path parameter that will be bound to the name parameter on the handler, indicated by Annotated[str, FromPath]
And now we test if our controller is working:
$ curl localhost:8000/hello/World\n{\"greeting\": \"Hello, World!\"}\nRight now our controller just get a name from the query string and return a dict. When a handler returns a dict or a list it will be automatically converted to JSON.
Our service will have a method that receives a name and returns a greeting. It will be injected into the controller we created previously.
/application/service.pyapplication/controller.pyfrom selva.di import service\n@service # (1)\nclass Greeter:\ndef greet(self, name: str) -> str:\nreturn f\"Hello, {name}!\"\n@service registers the class in the dependency injection system so it can be injected in other classesfrom typing import Annotated\nfrom asgikit.requests import Request\nfrom asgikit.responses import respond_json\nfrom selva.di import Inject\nfrom selva.web import controller, get\nfrom .service import Greeter\n@controller\nclass GreetingController:\ngretter: Annotated[Gretter, Inject] # (1)\n@get(\"/hello/:name\")\nasync def hello(self, request: Request, name: Annotated[str, FromPath]):\ngreeting = self.greeter.greet(name)\nawait respond_json(request.response, {\"greeting\": greeting})\nGreeter serviceOur greeting application is working fine, but we might want to add register the greeting requests in a persistent database, for auditing purposes.
To do this we need to create the database service and inject it into the Greeter service. For this we can use the Databases library with SQLite support:
pip install databases[aiosqlite]\nDatabases provides a class called Database. However, we can not decorate it with @service, so in this case we need to create a factory function for it:
from datetime import datetime\nfrom typing import Annotated\nfrom databases import Database\nfrom selva.di import service, Inject\n@service # (1)\nasync def database_factory() -> Database:\ndatabase = Database(\"sqlite:///database.sqlite3\")\nquery = \"\"\"\n create table if not exists greeting_log(\n greeting text not null,\n datetime text not null\n );\n \"\"\"\nawait database.execute(query)\nreturn database\n@service\nclass GreetingRepository:\ndatabase: Annotated[Database, Inject] # (2)\nasync def initialize(self): # (3)\nawait self.database.connect()\nasync def finalize(self): # (4)\nawait self.database.disconnect()\nasync def save_greeting(self, greeting: str, date: datetime):\nquery = \"\"\"\n insert into greeting_log (greeting, datetime)\n values (:greeting, datetime(:datetime))\n \"\"\"\nparams = {\"greeting\": greeting, \"datetime\": date}\nawait self.database.execute(query, params)\nA function decorated with @service is used to create a service when you need to provide types you do not own
Inject the Database service in the GreetingRepository
A method called initialize will be invoked after the service is constructed in order to run any initialization logic
A method called finalize will be invoked before the service is destroyed in order to run any cleanup logic
from typing import Annotated\nfrom datetime import datetime\nfrom asgikit.requests import Request\nfrom asgikit.responses import respond_json\nfrom selva.di import Inject\nfrom selva.web import controller, get, FromPath\nfrom .repository import GreetingRepository\nfrom .service import Greeter\n@controller\nclass GreetingController:\ngreeter: Annotated[Greeter, Inject]\nrepository: Annotated[GreetingRepository, Inject]\n@get(\"hello/:name\")\nasync def hello_name(self, request: Request, name: Annotated[str, FromPath]):\ngreeting = self.greeter.greet(name)\nawait self.repository.save_greeting(greeting, datetime.now())\nawait respond_json(request.response, {\"greeting\": greeting})\nThe greetings are being saved to the database, but now we have a problem: the user has to wait until the greeting is saved before receiving it.
To solve this problem and improve the user experience, we can use save the greeging after the request is completed:
application/controller.pyfrom datetime import datetime\nfrom typing import Annotated\nfrom asgikit.requests import Request\nfrom asgikit.responses improt respond_json\nfrom selva.di import Inject\nfrom selva.web import controller, get, FromPath\nfrom .repository import GreetingRepository\nfrom .service import Greeter\n@controller\nclass GreetingController:\ngreeter: Annotated[Greeter, Inject]\nrepository: Annotated[GreetingRepository, Inject]\n@get(\"hello/:name\")\nasync def hello_name(self, request: Request, name: Annotated[str, FromPath]):\ngreeting = self.greeter.greet(name)\nawait respond_json(request.response, {\"greeting\": greeting}) # (1)\nawait self.repository.save_greeting(greeting, datetime.now()) # (2)\nThe call to respond_json completes the response
The greeting is saved after the response is completed
To see the greetings saved to the database, we just need to add a route to get the logs and return them:
application/repository.pyapplication/controllers.py@service\nclass GreetingRepository:\n# ...\nasync def get_greetings(self) -> list[tuple[str, str]]:\nquery = \"\"\"\n select l.greeting, datetime(l.datetime) from greeting_log l\n order by rowid desc\n \"\"\"\nresult = await self.database.fetch_all(query)\nreturn [{\"greeting\": r.greeting, \"datetime\": r.datetime} for r in result]\n@controller\nclass GreetingController:\n# ...\n@get(\"/logs\")\nasync def greeting_logs(self, request: Request):\ngreetings = await self.repository.get_greetings()\nawait respond_json(request.response, greetings)\nNow let us try requesting some greetings and retrieving the logs:
$ curl localhost:8000/hello/Python\n{\"greeting\": \"Hello, Python!\"}\n$ curl localhost:8000/hello/World\n{\"greeting\": \"Hello, World!\"}\n$ curl -s localhost:8000/logs | python -m json.tool\n[\n{\n\"greeting\": \"Hello, World!\",\n \"datetime\": \"2022-07-06 14:23:14\"\n},\n {\n\"greeting\": \"Hello, Python!\",\n \"datetime\": \"2022-07-06 14:23:08\"\n},\n]\nWe can also send the name in the body of the request, instead of the url, and use Pydantic to parse the request body:
application/models.pyapplication/controller.pyfrom pydantic import BaseModel\nclass GreetingRequest(BaseModel):\nname: str\n# ...\nfrom .model import GreetingRequest\n@controller\nclass GreetingController:\ngreeter: Annotated[Greeter, Inject]\nrepository: Annotated[GreetingRepository, Inject]\n# ...\n@post(\"hello\")\nasync def hello_post(self, request: Request, greeting_request: GreetingRequest):\nname = greeting_request.name\ngreeting = self.greeter.greet(name)\nawait respond_json(request.response, {\"greeting\": greeting})\nawait self.repository.save_greeting(greeting, datetime.now())\nAnd to test it:
$ curl -H 'Content-Type: application/json' -d '{\"name\": \"World\"}' localhost:8000/hello\n{\"greeting\": \"Hello, World!\"}\nLet's dig a little deeper and learn the basic concepts of Selva.
+We will create a greeting api that logs the greet requests.
+Before going any further, we need to install Selva and Uvicorn.
+ +A selva application is structured like the following:
+project/
+├── application/
+│ ├── __init__.py
+│ ├── controller.py
+│ ├── repository.py
+│ └── service.py
+├── configuration/
+│ └── settings.yaml
+└── resources/
+And... that's it! A module or package named application will automatically
+be imported and scanned for controllers and services.
We will use uvicorn to run the application and automatically reload when we
+make changes to the code:
$ uvicorn selva.run:app --reload
+INFO: Will watch for changes in these directories: ['/home/user/projects/selva-tutorial']
+INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
+INFO: Started reloader process [23568] using WatchFiles
+INFO: Started server process [19472]
+INFO: Waiting for application startup.
+INFO: Application startup complete.
+Controller classes hold handler methods that will respond to HTTP or WebSocket +requests. They can receive services through dependency injection.
+from typing import Annotated
+from asgikit.requests import Request
+from asgikit.responses import respond_json
+from selva.web import controller, get, FromPath
+
+
+@controller # (1)
+class GreetingController:
+ @get("hello/:name") # (2)
+ async def hello(self, request: Request, name: Annotated[str, FromPath]):
+ await respond_json(request.response, {"greeting": f"Hello, {name}!"})
+@controller marks a class as a controller. It can optionally receive
+ a path (e.g. @controller("path")) that will be prepended to the handlers' path.
@get("hello/:name") defines the method as a handler on the given path.
+ If no path is given, the path from the controller will be used.
:name defines a path parameter that will be bound to the name
+parameter on the handler, indicated by Annotated[str, FromPath]
And now we test if our controller is working:
+ +Right now our controller just get a name from the query string and return a
+dict. When a handler returns a dict or a list it will be automatically
+converted to JSON.
Our service will have a method that receives a name and returns a greeting. It +will be injected into the controller we created previously.
+from selva.di import service
+
+
+@service # (1)
+class Greeter:
+ def greet(self, name: str) -> str:
+ return f"Hello, {name}!"
+@service registers the class in the dependency injection system so it
+ can be injected in other classesfrom typing import Annotated
+from asgikit.requests import Request
+from asgikit.responses import respond_json
+from selva.di import Inject
+from selva.web import controller, get
+from .service import Greeter
+
+
+@controller
+class GreetingController:
+ gretter: Annotated[Gretter, Inject] # (1)
+
+ @get("/hello/:name")
+ async def hello(self, request: Request, name: Annotated[str, FromPath]):
+ greeting = self.greeter.greet(name)
+ await respond_json(request.response, {"greeting": greeting})
+Greeter serviceOur greeting application is working fine, but we might want to add register +the greeting requests in a persistent database, for auditing purposes.
+To do this we need to create the database service and inject it into the +Greeter service. For this we can use the Databases +library with SQLite support:
+ +Databases provides a class called Database. However, we can not decorate it
+with @service, so in this case we need to create a factory function for it:
from datetime import datetime
+from typing import Annotated
+from databases import Database
+from selva.di import service, Inject
+
+@service # (1)
+async def database_factory() -> Database:
+ database = Database("sqlite:///database.sqlite3")
+ query = """
+ create table if not exists greeting_log(
+ greeting text not null,
+ datetime text not null
+ );
+ """
+ await database.execute(query)
+ return database
+
+
+@service
+class GreetingRepository:
+ database: Annotated[Database, Inject] # (2)
+
+ async def initialize(self): # (3)
+ await self.database.connect()
+
+ async def finalize(self): # (4)
+ await self.database.disconnect()
+
+ async def save_greeting(self, greeting: str, date: datetime):
+ query = """
+ insert into greeting_log (greeting, datetime)
+ values (:greeting, datetime(:datetime))
+ """
+ params = {"greeting": greeting, "datetime": date}
+ await self.database.execute(query, params)
+A function decorated with @service is used to create a service when
+ you need to provide types you do not own
Inject the Database service in the GreetingRepository
A method called initialize will be invoked after the service is
+ constructed in order to run any initialization logic
A method called finalize will be invoked before the service is
+ destroyed in order to run any cleanup logic
from typing import Annotated
+from datetime import datetime
+from asgikit.requests import Request
+from asgikit.responses import respond_json
+from selva.di import Inject
+from selva.web import controller, get, FromPath
+from .repository import GreetingRepository
+from .service import Greeter
+
+
+@controller
+class GreetingController:
+ greeter: Annotated[Greeter, Inject]
+ repository: Annotated[GreetingRepository, Inject]
+
+ @get("hello/:name")
+ async def hello_name(self, request: Request, name: Annotated[str, FromPath]):
+ greeting = self.greeter.greet(name)
+ await self.repository.save_greeting(greeting, datetime.now())
+ await respond_json(request.response, {"greeting": greeting})
+The greetings are being saved to the database, but now we have a problem: the +user has to wait until the greeting is saved before receiving it.
+To solve this problem and improve the user experience, we can use save the greeging +after the request is completed:
+from datetime import datetime
+from typing import Annotated
+from asgikit.requests import Request
+from asgikit.responses improt respond_json
+from selva.di import Inject
+from selva.web import controller, get, FromPath
+from .repository import GreetingRepository
+from .service import Greeter
+
+
+@controller
+class GreetingController:
+ greeter: Annotated[Greeter, Inject]
+ repository: Annotated[GreetingRepository, Inject]
+
+ @get("hello/:name")
+ async def hello_name(self, request: Request, name: Annotated[str, FromPath]):
+ greeting = self.greeter.greet(name)
+ await respond_json(request.response, {"greeting": greeting}) # (1)
+
+ await self.repository.save_greeting(greeting, datetime.now()) # (2)
+The call to respond_json completes the response
The greeting is saved after the response is completed
+To see the greetings saved to the database, we just need to add a route to get +the logs and return them:
+@service
+class GreetingRepository:
+ # ...
+ async def get_greetings(self) -> list[tuple[str, str]]:
+ query = """
+ select l.greeting, datetime(l.datetime) from greeting_log l
+ order by rowid desc
+ """
+ result = await self.database.fetch_all(query)
+ return [{"greeting": r.greeting, "datetime": r.datetime} for r in result]
+Now let us try requesting some greetings and retrieving the logs:
+$ curl localhost:8000/hello/Python
+{"greeting": "Hello, Python!"}
+
+$ curl localhost:8000/hello/World
+{"greeting": "Hello, World!"}
+
+$ curl -s localhost:8000/logs | python -m json.tool
+[
+ {
+ "greeting": "Hello, World!",
+ "datetime": "2022-07-06 14:23:14"
+ },
+ {
+ "greeting": "Hello, Python!",
+ "datetime": "2022-07-06 14:23:08"
+ },
+]
+We can also send the name in the body of the request, instead of the url, and +use Pydantic to parse the request body:
+# ...
+from .model import GreetingRequest
+
+
+@controller
+class GreetingController:
+ greeter: Annotated[Greeter, Inject]
+ repository: Annotated[GreetingRepository, Inject]
+
+ # ...
+
+ @post("hello")
+ async def hello_post(self, request: Request, greeting_request: GreetingRequest):
+ name = greeting_request.name
+ greeting = self.greeter.greet(name)
+ await respond_json(request.response, {"greeting": greeting})
+ await self.repository.save_greeting(greeting, datetime.now())
+And to test it:
+ + + + + + + +