Écrit par Pierre Chat

Depuis quelques années, je poursuis un projet personnel, développer un framework en Python. Un framework de quoi ? Bonne question !

Le cœur de Bolinette

Le package principal du framework, sobrement baptisé core contient plusieurs éléments:

• Un système d’injection de dépendances
• Un mapper d’objet pour cloner et transformer des instances de classes ou convertir en types Python primitifs
• Une gestion des fichiers d’environnement écrits en YAML
• Un CLI facile à utiliser

Base de données et ORM

Le second package de Bolinette s’appelle data. Il s’agit d’un wrapper de SQLAlchemy avec quelques ajouts qui intègrent l’ORM en respectant la syntaxe du package core.

Endpoints web et sockets

Le troisième package est web. Il permet de créer des API web en mappant les endpoints à des fonctions. En s’appuyant sur le mapper du package core, il permet de recevoir directement en entrée du contrôleur des instance de classes, plutôt que des types JSON primitifs (list, dict, int, etc.)

Open Source Software

Bolinette est publié sous licence MIT. Pour l’instant, les contributions ne sont pas acceptées, j’estime que le projet n’est pas prêt. Mais il est disponible à l’utilisation sur GitHub et le dépôt PyPI dès maintenant !

bolinette.core

Injection de dépendances

from bolinette.core.injection import Injection

class Service:
  def hello() -> str:
    return "hello"

inject = Injection()
inject.add(Service, "singleton")

service = inject.require(Service)
assert service.hello() == "hello"

Voici l’example le plus simple. On crée un service, on l’ajoute au registre de l’injection et, plus tard, on demande une instance. Il faut aussi préciser la stratégie d’injection:

• singleton: une seule instance est maintenue et donnée à chaque require.
• transcient: une nouvelle instance est créée à chaque require.
• scoped: cette stratégie n’est disponible qu’en mode Scoped, voir plus tard.

Le système d’injection de Bolinette s’appuie sur le typage optionnel de Python, que le framework rend obligatoire. Les dépendances de la méthode __init__ d’un service sont résolues avec les services renseignés dans le contexte d’injection.

from bolinette.core.injection import Injection

class Service:
  def hello() -> str:
    return "hello"

class Controller:
  def __init__(self, service: Service) -> None:
    self._service = service

  def get_hello(self) -> str:
    return self._service.hello()

def say_hello(ctrl: Controller) -> str:
  return ctrl.get_hello()

inject = Injection()
inject.add(Service, "singleton")
inject.add(Controller, "singleton")

ctrl = inject.require(Controller)
assert ctrl.get_hello() == "hello"
assert inject.call(say_hello) == "hello"

Dans ces exemples, le service d’injection est créé manuellement, mais dans un contexte réel d’utilisation du framework (commandes CLI, endpoint API, etc.) tout est transparent. Les services injectables sont déclarés à l’aide de décorateurs.

L’instantiation des dépendances ne se fait que lors du premier accès. En attendant, l’attribut qui contient le service est un proxy qui attend d’être instancié. Dans l’example, tant que self._service.hello() n’est pas appelé, self._service n’est pas instancié.

from bolinette.core.injection import injectable

@injectable("singleton")
class Service:
  def hello() -> str:
    return "hello"

# Nous verrons plus tard comment utiliser les commandes.
def cmd(service: Service) -> None:
  assert service.hello() == "hello"

⚠️ Les services injectés dans la méthode __init__ ne peuvent pas être utilisés dans cette même méthode, une erreur d’accès sera levée. Cela permet de pouvoir injecter des services en dépendance circulaire. Utilisez les méthodes d’initialisation pour faire des traitements à l’instanciation.

Méthodes d’initialisation

Ces méthodes sont appelées dès la sortie de la méthode __init__ lors de l’instanciation d’un service. Elles sont appelées dans l’ordre de déclaration, en commançant par les classes parentes, elles-mêmes dans l’ordre.

Le décorateur @init_method permet de déclarer une méthode en tant que méthode d’initialisation.

Une méthode d’initialisation peut demander des services dans sa signature. Cela permet de demander un service seulement utile pour cette méthode et ne pas le stocker en variable d’instance dans l’__init__.

from bolinette.core.injection import injectable, init_method

from my_app.services import AuthService, CredentialsService

@injectable("scoped")
class BusinessService():
  def __init__(self, auth: AuthService) -> None:
    self.auth_service = auth

  @init_method
  def _init_service(self, creds: CredentialsService) -> None:
    self.auth_service.set_credentials(creds.get_credentials())

  async def process(self) -> None:
    await self.auth_service.login()
    do_stuff()
    await self.auth_service.logout()

⚠️ Les dépendances circulaires ne sont pas autorisées dans les méthodes d’initialisation. Si un cycle d’appel est détecté, une erreur sera levée.

⚠️ Le système d’injection ne supporte pas les méthodes d’initialisation asynchrones. Veuillez utiliser un context manager asynchrone en mode scopé (cf. plus bas), ou une fonction de démarrage (cf. plus bas).

Injection en scope

Certains services doivent avoir une durée de vie, comme une transaction de base de données par exemple. Pour cela, il existe une version d’injection à durée de vie limitée.

Un service peut être déclaré comme scoped, un mode qui est réservé à l’injection scopée. Si un service singleton ou transcient requiert un service scoped, une erreur est levée. Les instances scoped existent seulement le temps de vie de l’injection et ne sont pas réutilisées lors d’une prochaine scope.

Dans un contexte de production, en mode commande ou API web, une scope est automatiquement créée avant de donner la main au code de l’utilisateur.

Il existe deux types d’injections scopées: synchrone et asynchrone. Les sessions asynchrones permettent d’utiliser des context managers asynchrones, mais les méthodes d’initialisation asynchrones ne sont pas permises.

from bolinette.core.injection import injectable, Injection

GLOBAL = {"count": 0}

@injectable("scoped")
class ScopedService:
  def __init__(self) -> None:
    GLOBAL["count"] += 1

async def cmd(inject: Injection) -> None:
  with inject.get_scoped_session() as sub_inject:
    sub_inject.require(ScopedService)
  async with inject.get_async_scoped_session() as asub_inject:
    asub_inject.require(ScopedService)
  assert GLOBAL["count"] == 2  # Le service a été instancié deux fois

Context manager

L’injection supporte des instances qui implémentent l’API context manager de Python, à savoir être utilisées avec la syntaxte with ... as ...:. Les deux méthodes à implémenter, pour un contexte synchrone, sont:

• def __enter__(self) -> Self:
• def __exit__(self, *args: tuple[type[BaseException], Exception, TracebackType] | tuple[None, None, None]) -> Self:

Et pour un contexte asynchrone:

• async def __aenter__(self) -> Self:
• async def __aexit__(self, *args: tuple[type[BaseException], Exception, TracebackType] | tuple[None, None, None]) -> Self:

Les contextes asynchrones sont seulement disponibles en mode scoped, avec une session d’injection asynchrone.

⚠️ WIP: la méthode __aenter__ n’est pas encore supportée.

from typing import Self

from bolinette.core.injection import injectable, Injection

@injectable("singleton")
class SyncService:
  def __enter__(self) -> Self:
    print("enter sync")

  def __exit__(self, *args: tuple[type[BaseException], Exception, TracebackType] | tuple[None, None, None]) -> Self:
    print("exit sync")

@injectable("scoped")
class AsyncService:
  async def __aenter__(self) -> Self:
    print("enter async")

  async def __aexit__(self, *args: tuple[type[BaseException], Exception, TracebackType] | tuple[None, None, None]) -> Self:
    print("exit async")

async def cmd() -> None:
  with Injection() as inject:
    inject.require(SyncService)
    async with inject.get_async_scoped_session() as async_scope:
      async_scope.require(AsyncService)
# stdout:
# enter sync
# exit async
# exit sync

Environnement

Bolinette permet de charger des fichiers de configuration YAML en fonction de l’environnement activé. Différents fichiers sont chargés et les clés sont écrasées pour permettre des configurations communes à plusieurs environnements. L’ordre de chargement est le suivant:

• env.yaml
• env.[profil].yaml
• env.local.[profil].yaml
• les variables d’environnement qui commancent par "BLNT_"

Par défaut, ces fichiers se trouvent dans le sous-dossier env, relatif au dossier courant d’exécution.

# env/env.yaml
api:
  name: Test
  url: test.example
  params:
    - 24
    - 478
bd:
  url: bd.example:1234
  user: test
  password: test

# script
from bolinette.core.environment import Environment

def cmd(env: Environment) -> None:
  assert env.config["api"]["name"] == "Test"
  assert env.config["db"]["user"] == "test"

Il est également possible de mapper l’environnement à des instances de classes. Les variables doivent être rangées en sections qui seront affectées à une classe différente.

from bolinette.core.environment import environment

@environment("api")
@dataclass(init=False)  # Pour faire un taire un pyright réglé sur strict
class ApiOptions:
  name: str
  url: str
  params: list[str]

@environment("db")
@dataclass(init=False)
class DbOptions:
  url: str
  user: str
  password: str

def cmd(api_opt: ApiOptions, db_opt: DbOptions) -> None:
  assert api_opt.name == "Test"
  assert db_opt.user == "test"

Mapper

Le mapper permet de transformer des données d’une instance à une autre, de la même classe ou de classes différentes. Par défaut, le mapper fonctionne automatiquement, se basant sur les noms des attributs d’une instance. Il n’est obligatoire de déclarer qu’un mapping, mais le mapper requiert que la classe d’arrivée soit intégralement annotée pour créer des nouvelles instances.

from bolinette.core.mapping import Mapper

class Source:
  def __init__(self, value: int) -> None:
    self.value = value

class Destination:
  value: int

class DestWithInit:
  def __init__(self, value: int) -> None:
    self.value = value

def cmd(mapper: Mapper) -> None:
  src = Source(42)
  dst1 = mapper.map(Source, Destination, src)
  # Destination n'a pas d'init, le mapper peut l'instancier.
  assert dst1.value == 42

  dst2 = DestWithInit(68)
  assert dst2.value == 68
  dst2 = mapper.map(Source, DestWithInit, src, dst2)
  # DestWithInit demande des paramètres d'init, une instance doit être donnée au mapper.
  assert dst2.value == 42

Profils de mapping

Quand le mapping n’est pas trivial, il est possible de surcharger le comportement par défaut du mapper. Cela se fait en déclarant un profil qui va décrire comment procéder pour chaque transformation.

from bolinette.core.mapping import Mapper, Profile, profile, mapping

class Source:
  def __init__(self, value: int) -> None:
    self.value = value

class Destination:
  id: int
  name: str | None

@mapping()
class TestProfile(Profile):
  def __init__(self) -> None:
    super().__init__()
    self.register(Source, Destination).for_attr(
      lambda dest: dest.content, lambda opt: opt.map_from(lambda src: src.value)
    ).for_attr(
      lambda dest: dest.name, lambda opt: opt.ignore()
    ) # Seulement possible car name est optionel

def cmd(mapper: Mapper) -> None:
  src = Source(42)
  dst = mapper.map(Source, Destination, src)
  assert dst.id == 42
  asser dst.name is None

Commandes

Le point d’entrée principal du package core de Bolinette sont les commandes CLI. Ce sont des fonctions, possiblement asynchrones, qui sont appelées lorsque la commande mappée est demandée par le terminal.

Les commandes sont exécutées dans une scope, il est donc possible de demander des services singleton, transcient et scoped. Les arguments sont également décrits dans la signature avec des types annotés, avec leur nom, leur positionnement et une éventuelle valeur par défaut.

from typing import Literal, Annotated

from bolinette.core.command import command, Argument
from bolinette.core.injection import injectable

@injectable("scoped")
class Service:
  async def create_user(self, name: str, age: int, admin: bool) -> User:
    ...

  async def add_to_group(self, user: User, group: str) -> User:
    ...

  async def commit(self) -> bool:
    ...

@command("create user", "Cette commande permet de créer un utilisateur")
async def my_command(
  service: Service,
  name: Annotated[str, Argument()],
  age: Annotated[int, Argument()],
  group: Annotated[str | None, Argument("option", "g")],  # "g" déclare un raccourci à une lettre
  admin: Annotated[bool, Argument("switch")] = False,
  commit: Annotated[bool, Argument("switch", "c")] = False,
) -> None:
  assert isinstance(service, Service)
  user = await service.create_user(name, age, admin)
  if group is not None:
    await service.add_to_group(user, group)
  if commit:
    await service.commit()

Dans l’example, name et age sont des arguments positionnels. Ils n’ont pas de valeur par défaut, ils sont obligatoires.group est un argument nommé, soit via --group ou -g, nullable donc non requis. admin et commit sont des bascules booléennes. Si les arguments, --admin et --commit, ou le raccourci -c, sont présents, une valeur True sera affectée. La valeur par défaut False permet de rendre ces bascules non requises.

Voici un exemple d’utilisation de cette commande. Voir plus bas comment initialiser Bolinette et lancer le script.

python -m my_app create user Bob 42 -g boys --admin -c

Fonctions de démarrage

Ces fonctions sont exécutées au démarrage de Bolinette. Elles peuvent être asynchrones et demander n’importe quel service dans leur signature. Elles sont exécutées dans une session d’injection asynchrone et peuvent dépendre de services scoped.

from bolinette.core import startup

@startup()
async def run_at_launch(service: MyService) -> None:
  await service.do_stuff()

Voir plus bas comment déclencher le démarrage de Bolinette.

Utiliser Bolinette dans mon projet

Tout le chapitre précédent décrit les élément indépendants du package core. Mais ils forment un tout qui fonctionne en symbiose. Le point d’entrée standard de Bolinette sont les commandes CLI. Le package web, voir plus bas, ajoute la possibilité de créer une API web.

Il est d’abord nécessaire de créer l’application Bolinette et charger les différentes extensions.

# module my_app.app
from bolinette import data, web
from bolinette.core import Bolinette

def make_bolinette():
  blnt = Bolinette()
  blnt.use_extension(data)
  blnt.use_extension(web).use_websockets()
  return blnt

Chaque extension expose ses propres méthodes de paramétrage afin d’activer des options et personnaliser les fonctionnalités.

Ensuite, il faut créer le point d’entrée de l’application, en créant un fichier __main__.py dans votre module racine, ou en écrivant le script de démarrage dans un bloc if __name__ == "__main__":

# module my_app.__main__
import asyncio
import sys

from my_app.app import make_bolinette

blnt = make_bolinette()
asyncio.run(blnt.exec_args(sys.argv[1:]))

Ce code boilerplate permet de lancer le CLI à partir des arguments fournis à Python. exec_args(...) va s’occuper du mécanisme de démarrage et lancer la commande passée en paramètre, avec ses arguments.

Bolinette.startup(self) peut être appelée directement, pour utiliser Bolinette sans le CLI. L’injection est disponible sous Bolinette.injection. Vous pouvez ensuite ouvrir une session scopée et créer votre propre flow de code en profitant de tous les services de Bolinette.

# module my_app.__main__
import asyncio
import sys

from bolinette.core import Bolinette

from my_app.app import make_bolinette, do_wacky_stuff

async def run_my_app(blnt: Bolinette) -> None:
  await blnt.startup()
  async with blnt.injection.get_async_scoped_session() as session:
    await session.call(do_wacky_stuff)

blnt = make_bolinette()
asyncio.run(run_my_app(blnt))

⚠️ Bolinette fonctionne principalement avec des décorateurs. Veillez à bien importer tous vos fichiers afin qu’il soient lus par l’interpréteur. Si vous obtenez des erreurs d’injections sur un service inconnu, c’est peut-être que vous avez oublié de charger le fichier qui contient votre classe.

bolinette.data

Modèles

La base du package data sont les modèles, qui servent à décrire les données. Bolinette utilise l’ORM SQLAlchemy, version 2, sans le masquer, afin d’offrir toutes les fonctionnalités complexes que propose la bibliothèque.

Il est possible de gérer des modèles provenant de plusieurs sources de données, mais les requêtes entre ces modèles sont impossibles.

Voici comment déclarer un modèle de données:

from bolinette.data.relational import entity, get_base
from sqlalchemy.orm import Mapped, mapped_column

@entity()
class User(get_base("default")):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    username: Mapped[str] = mapped_column(unique=True)

Manipulation des données

La classe Repository permet de manipuler les modèles. Toutes ses méthodes sont asynchrones afin d’être utilisées avec la bibliothèque asyncio de Python. SQLAlchemy supporte des pilotes asynchrones sur les bases de données PostgreSQL et SQLite. Les autres pilotes synchrones sont aussi supportés par le Repository.

Dans l’exemple suivant, voici comment ajouter et récupérer des données en base. Les deux fonctions sont appelées depuis une session scopée. Les repositories sont seulement disponibles en mode scoped car ils dépendent d’une session en base ouverte au début de la scope. Le mécanisme de commit de la session est automatique en sortie de scope.

from bolinette.core.injection import Injection
from bolinette.data.relational import Repository, repository
from sqlalchemy import select

@repository(User)
class UserRepository(Repository[User]):
  pass

def create_users(user_repo: UserRepository) -> None:
  user_repo.add(User(username="Bob"))
  user_repo.add(User(username="Alice"))

async def get_user(username: str, user_repo: UserRepository) -> User:
  return await user_repo.first(select(User).where(User.username == username))

async def cmd(inject: Injection) -> None:
  async with inject.get_async_scoped_session() as scoped:
    scoped.call(create_users)
  async with inject.get_async_scoped_session() as scoped:
    bob = await scoped.call(get_user, "Bob")
    assert bob.username == "Bob"

La session SQLAlchemy est exposée dans la classe Repository. Il est possible de l’utiliser directement sans passer par les fonctions wrapper de Bolinette.

@repository(User)
class UserRepository(Repository[User]):
  async def print_all_users_manual(self) -> None:
    """Itération directement sur le curseur SQLAlchemy."""
    result = await self.session.execute(select(User))
    for user in result.scalars():
      print(f"<User {user.username}>")

  async def print_all_users_iter(self) -> None:
    """Itération sur un wrapper asynchrone."""
    async for user in self.iterate(select(User)):
      print(f"<User {user.username}>")

bolinette.web

Le package web est basé sur la bibliothèque aiohttp et expose ses classes telles que Request et Response afin de personnaliser au maximum le format des données renvoyées au client.

Controlleurs web

Le principe du package web est mapper des endpoints web à des fonctions Python. Ces fonctions sont des routes, de méthodes de classes appelées contrôleurs. Le contrôleur est instancié à l’invocation de la route. L’__init__ du contrôleur et la route peuvent demander des services dans leur signature, l’__init__ permetant de mutualiser les dépendances communes des différentes routes du contrôleur, comme un service, l’environnement, le mapper, etc.

from typing import Annotated

from bolinette.core.mapping import Mapper
from bolinette.web import controller, get, post, put, patch, delete, Payload

from my_app.entities import Item
from my_app.services import UserService
from my_app.controllers.responses import ItemResponse
from my_app.controllers.payloads import CreateItemPayload, UpdateItemPayload, PatchItemPayload

@controller("/api/store")
class StoreController:
  def __init__(self, mapper: Mapper, service: UserService) -> None:
    self.mapper = mapper
    self.service = service

  @get("")
  async def get_all(self) -> list[ItemResponse]:
    items = await self.service.get_all()
    return self.mapper.map(list[Item], list[ItemResponse], items)

  @post("")
  async def create(self, payload: Annotated[CreateItemPayload, Payload()]) -> ItemResponse:
    item = await self.service.create(payload)
    return self.mapper.map(Item, ItemResponse, item)

  @get("{id}")
  async def get_one(self, id: int) -> ItemResponse:
    item = await self.service.get_by_id(id)
    return self.mapper.map(Item, ItemResponse, item)

  @put("{id}")
  async def update(self, id: int, payload: Annotated[UpdateItemPayload, Payload()]) -> ItemResponse:
    item = await self.service.update(id, payload)
    return self.mapper.map(Item, ItemResponse, item)

  @patch("{id}")
  async def patch(self, id: int, payload: Annotated[PatchItemPayload, Payload()]) -> ItemResponse:
    item = await self.service.update(id, payload)
    return self.mapper.map(Item, ItemResponse, item)

  @delete("{id}")
  async def delete(self, id: int) -> None:
    await self.service.delete(id)

Voici un exemple typique d’opérations CRUD. Les variables de route, formattées entre accolades, peuvent être injectées dans la signature de la route et typées avec des builtins Python. Le body des requêtes POST, PUT et DELETE peuvent être mappés sur des classes DTO en annotant le type avec Payload().

Ici, le service est une classe métier. Il est possible de combiner les packages web et data afin de manipuler les objets d’une base de données. Pour cela, il est recommandé de créer des objets intermédiaires afin de contrôler quelles informations sortent de l’API et quelles informations sont demandées pour créer et modifier la base métier.

from typing import TypedDict

from bolinette.core.mapper import Profile, mapping

from my_app.entities import Item

@dataclass(init=False)  # permet d'éviter un warning sur les attributs non initialisés et un init vide est nécessaire pour le mapper
class ItemResponse:
  name: str
  price: float
  tax: float
  final_price: float

@dataclass(init=False)
class CreateItemPayload:
  name: str
  price: float
  tax: float

@dataclass(init=False)
class UpdateItemPayload:
  name: str
  price: float
  tax: float

class PatchItemPayload(TypedDict, total=False):
  name: str
  price: float
  tax: float

@mapping
class ItemProfile(Profile):
  def __init__(self) -> None:
    super().__init__()
    self.register(Item, ItemResponse).for_attr(
      lambda dest: dest.final_price, lambda opt: opt.map_from(lambda src: src.price * src.tax)
    )

Dans cet exemple, les classes pour la récupération, la création et l’édition sont des dataclasses, entièrement typées pour être compatibles avec le mapper.

La classe de patch est un TypedDict, qui est transformé au runtime en dict classique. Cela permet de gérer une requête PATCH REST classique. Si l’attribut est nul, il doit être passé à nul en base. Si l’attribut n’est pas présent, il doit être ignoré. La métadonnée total=False permet de dire au mapper de ne pas lever d’erreur lorsque l’attribut n’est pas présent dans le body JSON. Le même résultat est possible en décorant les types avec NotRequired, comme par exemple name: NotRequired[str].

Middlewares

Les middlewares s’interposent entre la réponse et la route afin de faire des traitements, avant ou après le traitement de la route du contrôleur.

Un middleware est une classe qui requiert deux méthodes:

class Middleware[**MdlwInitP](Protocol):
  def options(self, *args: MdlwInitP.args, **kwargs: MdlwInitP.kwargs) -> None:
    ...

  async def handle(self, next: Callable[[], Awaitable[web.Response]]) -> web.Response:
    ...

• options permet de définir des paramètres à renseigner dans le décorateur.
• handle est appelée avant l’exécution de la route. Il est attendu que le code appelle next afin de poursuivre la chaîne des middlewares et enfin la route. Tout code placé avant l’appel à next sera exécuté avant la route et tout code placé après sera exécuté après.
• Utilisez l’__init__ du middleware pour demander des services à l’injection. Les middlewares sont instanciés à chaque appel de route dans une session scopée asynchrone.

Voici un exemple complet de déclaration et utilisation de middlewares:

from aiohttp
from bolinette.core.mapping import Mapper
from bolinette.web import with_middleware, without_middleware, controller, get
from bolinette.web.exceptions import UnauthorizedError, ForbiddenError

from my_app.services import UserService, AuthService
from my_app.controllers.responses import UserResponse, UserPublicResponse, UserAdminResponse

class AuthMiddleware:
  def __init__(self, request: Request, user_service: UserService, auth_service: AuthService) -> None:
    self.request = request
    self.user_service = user_service
    self.auth_service = auth_service
    self.roles: list[str] | None

  def options(self, roles: list[str] | None = None) -> None:
    self.roles = roles

  async def handle(self, next: Callable[[], Awaitable[web.Response]]) -> web.Response:
    token: str | None = self.request.headers.get('Authorization')
    if not self.auth_service.verify_token(token):
      raise UnauthorizedError()
    if self.roles is not None:
      user = await self.user_service.get_from_token(token)
      if not any(r for r in user.roles if r.name in self.roles):
        raise ForbiddenErorr()
    return await next()

@controller("/api/user")
@with_middleware(AuthMiddleware)
class UserController:
  def __init__(self, user_service: UserService, mapper: Mapper) -> None:
    self.user_service = user_service
    self.mapper = mapper

  @get("")
  async def get_all_users(self) -> list[UserPublicResponse]:
    users = await self.user_service.get_all()
    return self.mapper.map(list[User], list[UserResponse], users)

  @get("public")
  @without_middleware(AuthMiddleware)
  async def get_all_users_public(self) -> list[UserPublicResponse]:
    users = await self.user_service.get_all()
    return self.mapper.map(list[User], list[UserPublicResponse], users)

  @get("admin")
  @with_middleware(AuthMiddleware, roles=['admin'])
  async def get_all_users_admin(self) -> list[UserPublicResponse]:
    users = await self.user_service.get_all()
    return self.mapper.map(list[User], list[UserAdminResponse], users)

Dans l’exemple précédant, le middleware est appliqué à la classe, ce qui l’applique à toutes les routes du contrôleur.without_middleware permet d’annuler l’exécution du middleware global pour une route spécifique. Il est possible de redéfinir un même with_middleware afin de modifier les arguments attendus par la méthode options du middleware pour une route spécifique.

Autant de with_middleware et without_middleware peuvent être empilés au-dessus d’un contrôleur ou d’une route que nécessaire. Un middleware peut être déclaré directement sur une route, sans qu’il soit déclaré sur le contrôleur. Les middlewares sont exécutés dans l’ordre de déclaration, ceux de la classe puis ceux de la méthode. Si un middleware est redéclaré sur une route, il garde l’ordre de passage qui a été déclaré sur la classe.

Les tests

Une batterie de tests unitaires poussés permettent d’ajouter et refactoriser sereinement les fonctionnalités les plus critiques du framework. Bolinette est testé à plus de 90% et de nouveaux tests significatifs sont ajoutés continuellement. Les tests utilisent la souplesse et la puissance de la bibliothèque pytest et ses extensions.

La suite

Si le framework est déjà fonctionnel, il reste beaucoup de travail pour réaliser ma vision. Voici les prochaines fonctionnalités déjà prévue pour 2024:

• package core:
  • améliorer le mapper, ajouter des cas d’utilisation
  • ajouter le support SMTP ainsi que les fournisseurs de services mails les plus populaires
• package data:
  • créer un système de validation d’entités afin d’éviter des erreurs de format relevés par la base de données
• package web:
  • ajouter le support des websockets
  • ajouter un mécanisme d’authentification normalisé
  • ajouter un swagger automatique
• nouveau package api:
  • faire le lien entre data et web afin de générer des opérations CRUD automatiques

Pour la petite histoire

J’ai créé une première version de Bolinette en 2019. C’était un projet à greffer sur Flask, un framework web. Il manquait à Flask des fonctionnalités qui me semblaient utiles pour faire une API web de manière efficace, mais Flask étant décrit comme un micro framework, leur absence n’est en rien un défaut du framework.

J’ai donc, en quelques mois de travail sur mon temps libre, ajouté un système de sérialisation automatique et paramétrable afin de pouvoir renvoyer des objets métier directement depuis les contrôleurs, sans avoir à manuellement construire un dictonnaire que le module json de Python sait manipuler.

Les fonctionnalités telles que la validation automatique des données, la création de route automatiques et l’ajout de middlewares se sont rajoutées au cours de l’année suivante.

J’ai ensuite réalisé quelques tests de performance afin de voir les temps de réponse d’une API très simple réalisée avec Bolinette. J’ai été horrifié par les 500ms nécessaires à la récupération de quelques entités de base de données. Aujourd’hui encore, je ne sais pas d’où venait le problème, mais j’ai pris la décision radicale de remplacer Flask par aiohttp.

aiohttp n’est pas un framework, mais une bibliothèque HTTP qui fournis un routeur et traite les requêtes et réponses HTTP. J’ai dû remplacer les fonctionnalités offertes par Flask par d’autres bibliothèques spécialisées. Mes efforts de refactorisation ont été couronnés par un gain de vitesse multiplié par 10!

L’injection de dépendances a été ajoutée quelques temps après, mais le système fonctionnait seulement sur les routes de contrôleurs et n’était absolument pas flexible. Le framework utilisait encore beaucoup des magic strings, où les entités, services et contrôleurs étaient identifiés avec un nom, très difficile à refactoriser. Bolinette était un projet monolithique, encore décrit comme framework web.

Devant le constat de la compléxité d’utilisation et la difficulté de déméler le code intriqué, j’ai pris une autre décision radicale: tout recommencer.

Le chantier à commencé en 2022. J’ai d’abord créé le package core avec en tête une séparation des facettes de Bolinette. Le framework n’était plus un framework web, mais une plateforme de services divers liés par un système d’inversion de contrôle flexible et intuitif.

Le reste a suivi et aujourd’hui, je suis très fier du travail accompli et je vois clairement ou je souhaite ammener le projet. Mon but ultime est d’accepter des contributions d’autres développeurs et que Bolinette soit utilisé.

See you in 2025 🚀