Implémenter le modèle de référentiel en Python en utilisant SQLAlchemy comme backend | de Przemysław Górecki | novembre 2022

Conception pilotée par domaine en Python

Cet article fait partie d’une série plus longue sur l’implémentation des modèles tactiques de conception pilotée par le domaine en Python. L’objectif ici est de créer une application Web qui permet aux vendeurs de répertorier les articles à vendre et aux acheteurs d’enchérir sur les articles qu’ils sont prêts à acheter (alias eBay clone) – plus d’informations sur le projet peuvent être trouvées ici.

Dans le article précédentnous avons implémenté un simple référentiel basé sur des fichiers qui utilisait le pickle module. Comme nous n’utilisions pas de base de données, nous n’avions pas besoin de créer des tables et des colonnes, et tous les modèles de domaine étaient directement sérialisés d’une mémoire vers un flux d’octets (et vice versa). Une telle approche était acceptable pour introduire le concept de référentiel, mais dans un système de production, nous aurons besoin d’une base de données pour stocker l’état de l’application.

En passant, j’ai écrit pour la dernière fois que les référentiels sont utilisés pour charger et enregistrer des entités. Les référentiels fonctionnent avec des agrégats (qui sont des composites d’entités), mais par souci de simplicité, je m’en tiens aux entités.

Cette fois, nous utiliserons Alchimie SQL comme mécanisme de persistance pour les entités. Nous conserverons l’interface d’origine, mais tous les détails d’implémentation seront encapsulés dans le SqlAlchemyListingRepository. Commençons par une interface révisée vers un ListingRepositoryqui prend également en charge l’ajout, la suppression et la persistance Listing entités de domaine :

# somewhere in the domain layer
class ListingRepository(metaclass=abc.ABCMeta):
"""An interface to listing repository"""
@abc.abstractmethod
def add(self, entity: Listing):
"""Adds new entity to a repository"""
raise NotImplementedError()
@abc.abstractmethod
def remove(self, entity: Listing):
"""Removes existing entity from a repository"""
raise NotImplementedError()
@abc.abstractmethod
def get_by_id(id: ListingId) -> Listing:
"""Retrieves entity by its identity"""
raise NotImplementedError()
def __getitem__(self, index) -> Listing:
return self.get_by_id(index)

Dans la conception axée sur le domaine, l’objectif d’un référentiel est d’encapsuler la logique requise pour accéder aux objets du domaine. Par conséquent, l’interface avec un référentiel fait partie d’une couche de domaine, tandis que la mise en œuvre réelle de cette interface appartient à la couche d’infrastructure. Par conséquent, le code de domaine reste propre – il est isolé de tout problème technique et indépendant de la base de données.

Voici quelques concepts que j’aimerais présenter et discuter avant de passer directement à l’implémentation du référentiel SQL Alchemy :

1. Modèle ORM ≠ objet de domaine

Les entités sont Cours Python – Ce n’est pas une surprise. Ils contiennent des attributs : des types primitifs, des objets de valeur, des énumérations, des références à d’autres entités et des collections de tout ce qui précède. En mémoire, cela forme une structure de type graphique de champs imbriqués. Pour rappel, voici comment notre Listing l’entité ressemble (à ce stade, nous ne nous soucions pas de la logique):

@dataclass
class Listing(Entity):
id: int
name: str
min_price: Money
....

D’autre part, les bases de données organisent les données à plat et sous forme de tableau. Les classes Python et les modèles de données relationnelles ne fonctionnent pas ensemble, et les ORM ont été introduits pour surmonter ce problème. Cependant, à cause de la cartographie entre les attributs de modèle et les colonnes de table, les modèles ORM sont toujours étroitement couplés à une base de données et à une implémentation ORM spécifique :

class ListingModel(ModelBase):
__tablename__ = "listing"
id = Column(Integer, primary_key=True)
name = Column(String)
min_price__amount = Column(Integer)
min_price__currency = Column(String(3))

Cela pose deux problèmes :

1. Les modèles ORM ne doivent pas être utilisés dans la couche domaine. Les modèles ORM sont plus difficiles à tester que les objets de domaine (ils ont besoin d’une base de données pour exister). De plus, nous voulons que la couche métier reste propre sans aucune dépendance vis-à-vis des mécanismes de stockage sous-jacents. ListingModel est étroitement couplé à SQL Alchemy, et avoir une telle dépendance dans une couche métier est interdit. Cependant, il est toujours possible d’utiliser des modèles SQLA dans certains cas, c’est-à-dire si nous effectuons des opérations CRUD de base qui ne nécessitent pas de logique métier.

2. Il n’y a pas de moyen facile d’avoir un objet de valeur multichamp (c’est-à-dire min_price Money composé de amount et currency) à plusieurs colonnes dans le modèle. Théoriquement, nous pourrions ajouter quelques getters et setters pour les objets de valeur :

class ListingModel(ModelBase):
...
min_price__amount = Column(Integer)
min_price__currency = Column(String(3))
def get_min_price(self) -> Money:
return Money(self.min_price__amount, self.min_price__currency)
def set_min_price(self, value: Money):
self.min_price__amount = value.amount
self.min_price__currency = value.currency

… mais cela contredit l’idée que les objets de valeur sont immuables (vous pouvez toujours modifier un attribut individuel d’un modèle qui appartient logiquement à un objet de valeur).

Pour surmonter ces problèmes, nous avons besoin d’un mécanisme de transfert de données entre une couche métier – un mappeur de données.

2. Mappeur de données

Nous savons déjà que nous ne voulons pas utiliser de modèles dans la couche domaine. Les modèles sont spécifiques à SQL Alchemy, composés de colonnes, de clés étrangères, de relations, etc. Nous ne voulons pas que les détails de l’infrastructure soient divulgués dans le domaine. Nous pourrions toujours utiliser des modèles pour interroger les données (car il est acceptable d’interroger la base de données via des canaux autres que les dépôts), mais cela dépasse le cadre de cet article.

Ce que nous devons faire, c’est :
– mapper des modèles sur des entités lors de la lecture de données à partir d’un référentiel (c’est-à-dire get_by_id(…) call renvoie une entité de domaine),
– mapper des entités sur des modèles lorsque des modifications de données sont sur le point d’être enregistrées.

Cela peut être mis en œuvre de deux manières :

1. Nous pouvons utiliser SQL Alchemy cartographie impérative pour mapper les données des tables SQL vers des classes Python pures. Python cosmique suit cette approche. Ici, tous les attributs d’un modèle sont automatiquement traduits depuis/vers un objet du domaine, via mapper_registry.map_imperatively(). Cependant, cette technique est limitée aux modèles constitués uniquement de types primitifs. Nous avons besoin d’une approche plus sophistiquée si notre modèle de domaine a des objets de valeur multi-attributs.

2. Nous pouvons implémenter notre propre mappeur de données pour la traduction vers et depuis notre modèle de domaine. Pour faire le mapping, nous aurons besoin de deux fonctions : XYZ_model_to_entity et XYZ_entity_to_model. Le référentiel utilisera le mappeur de données pour effectuer toutes les traductions. C’est l’approche que nous allons adopter dans la mise en œuvre de SqlAlchemyListingRepository.

3. Carte d’identité

L’un des éléments essentiels du modèle de référentiel est le Carte d’identité, qui est utilisé pour améliorer les performances en fournissant un cache en mémoire spécifique au contexte pour empêcher la récupération en double des mêmes données d’objet à partir de la base de données au cours d’une seule transaction. Voici une autre définition :

Une carte d’identité conserve un enregistrement de tous les objets qui ont été lus à partir de la base de données dans une seule transaction commerciale. Chaque fois que vous voulez un objet, vous vérifiez d’abord la carte d’identité pour voir si vous l’avez déjà.
— Martin Fowler

C’est assez simple : lorsque nous interrogeons le référentiel pour une entité, nous vérifions d’abord la carte d’identité. Si l’entité est déjà présente dans la carte, le référentiel renverra une référence à un objet mis en cache. Sinon, nous allons lire le modèle à partir d’une base de données, le transformer avec un mappeur de données, le stocker dans la carte d’identité et renvoyer une référence. Lors de l’enregistrement d’une entité, nous utiliserons un mappeur de données pour la transformer en une instance de modèle, puis la conserverons à l’aide du mécanisme de session SQL Alchemy intégré.

Maintenant que nous savons quels sont les éléments constitutifs d’un référentiel, examinons tous les éléments dont nous avons besoin pour implémenter le référentiel de listes d’alchimie SQL.

Objets de domaine

Voici la partie données de notre modèle de domaine. Pour des raisons de clarté, toute la logique a été retirée des objets du domaine.

from uuid import UUID
from dataclasses import dataclass
# some type aliases
ListingId = UUID
BidderId = UUID
@dataclass
class Money:
"""A value object that represents money"""
amount: int
currency: str
@dataclass
class Bid:
"""A value object that represents a bid placed on a listing by a buyer"""
bidder_id: BidderId
price: Money
@dataclass
class Listing(Entity):
"""An entity that represents a listing with an ask price and all the bids already placed on this item"""
id: int
name: str
min_price: Money
bids: List[Bid] = field(default_factory=list)

Voici les choses intéressantes à remarquer. UN Listing est composé de certains types primitifs (c’est-à-dire id, name), un objet à valeur unique (min_price) et une liste d’objets de valeur (bids). Pour une raison non divulguée, supposons que nous utilisons intentionnellement une liste afin de pouvoir préserver l’ordre des enchères (sinon, nous pourrions utiliser une liste non ordonnée set).

Modèle de données

Les objets de domaine ci-dessus sont reflétés en tant que modèles de base de données comme suit :

import uuid
from infrastructure.sqlalchemy_common import Base, Column, UUID, String, Integer, ForeignKey, relationship
UniqueIdentifer = UUID(as_uuid=True)
class BidModel(Base):
""" Stores Bid value object"""
__tablename__ = "bid"
# composite primary key
listing_id = Column(UniqueIdentifier,
ForeignKey("listing.id"),
primary_key=True)
# since bids are stored in an ordered collection (list), an index column is required
idx = Column(Integer, primary_key=True)
bidder_id = Column(UniqueIdentifier)
price__amount = Column(Integer)
price__currency = Column(String(3))
# parent relationship
listing = relationship("ListingModel", back_populates="bids")
class ListingModel(Base):
__tablename__ = "listing"
id = Column(UniqueIdentifier, primary_key=True, default=uuid.uuid4)
name = Column(String(30))
min_price__amount = Column(Integer)
min_price__currency = Column(String(3))
bids = relationship("BidModel",
order_by="BidModel.idx.asc()",
cascade="save-update, merge, delete, delete-orphan")

Comme vous pouvez le constater, nous n’avons pas de tableau distinct; ceux-ci sont stockés en tant que champs primitifs dans les modèles correspondants (c’est-à-dire, price__amount, price__currency et min_price__amount, min_price__currency). Cependant, nous stockons bids dans un espace séparé BidModel car il existe une relation 1 à plusieurs entre ListingModel et BidModel.

De plus, les offres sont classées par idx champ pour préserver l’ordre des éléments dans Listing.bids liste, et il y a un cascade option définie pour conserver Listing.bids et les lignes du tableau sont synchronisées lorsque Bid les objets de valeur sont supprimés de la liste. Notez également qu’il utilise une clé primaire composite composée de listing_id et idx car il suffit d’identifier une offre au niveau de la base de données.

Mappeur de données

La logique du mappeur de données n’a rien d’extraordinaire – juste un mappage plutôt fastidieux d’un type à un autre. Un petit élément intéressant à noter est l’emballage/déballage des objets de valeur.

from infrastructure.models import ListingModel, BidModel
from domain.entities import Listing
from domain.value_objects import Money, Bid
def listing_model_to_entity(instance: ListingModel) -> Listing:
def map_bid_to_value_object(bid: BidModel) -> Bid:
return Bid(
bidder_id=bid.bidder_id, 
price=Money(amount=bid.price__amount, currency=bid.price__currency)
)
return Listing(
id=instance.id,
name=instance.name,
min_price=Money(amount=instance.min_price__amount, currency=instance.min_price__currency),
bids=[map_bid_to_value_object(bid) for bid in instance.bids]
)
def listing_entity_to_model(listing: Listing, existing=None) -> ListingModel:
def map_bid_to_model(idx: int, bid: Bid) -> BidModel:
return BidModel(bidder_id=bid.bidder_id, price__amount=bid.price.amount, price__currency=bid.price.currency, idx=idx)
return ListingModel(
id=listing.id,
name=listing.name,
min_price__amount=listing.min_price.amount,
min_price__currency=listing.min_price.currency,
bids=[map_bid_to_model(idx, bid) for idx, bid in enumerate(listing.bids)]

Référentiel d’inscription

Enfin, regardons l’implémentation d’un référentiel :

from sqlalchemy.orm import Session
from domain.repositories import ListingRepository
from domain.entities import Listing
from infrastructure.models import ListingModel
from infrastructure.data_mappers import listing_model_to_entity, listing_entity_to_model
# a sentinel value for keeping track of entities removed from the repository
REMOVED = object()
class SqlAlchemyListingRepository(ListingRepository):
"""SqlAlchemy implementation of ListingRepository"""
def __init__(self, session: Session, identity_map=None):
self.session = session
self._identity_map = identity_map or dict()
def add(self, entity: Listing):
self._identity_map[entity.id] = entity
instance = listing_entity_to_model(entity)
self.session.add(instance)
def remove(self, entity: Listing):
self._check_not_removed(entity)
self._identity_map[entity.id] = REMOVED
listing_model = self.session.query(ListingModel).get(entity.id)
self.session.delete(listing_model)
def get_by_id(self, id):
instance = self.session.query(ListingModel).get(id)
return self._get_entity(instance, listing_model_to_entity)
def get_by_name(self, name):
instance = self.session.query(ListingModel).filter_by(name=name).one()
return self._get_entity(instance, listing_model_to_entity)
def _get_entity(self, instance, mapper_func):
if instance is None:
return None
entity = listing_model_to_entity(instance)
self._check_not_removed(entity)
if entity.id in self._identity_map:
return self._identity_map[entity.id]
self._identity_map[entity.id] = entity
return entity
def __getitem__(self, key):
return self.get_by_id(key)
def _check_not_removed(self, entity):
assert self._identity_map.get(entity.id, None) is not REMOVED, f"Entity {entity.id} already removed"
def persist(self, entity: Listing):
self._check_not_removed(entity)
assert entity.id in self._identity_map, "Cannon persist entity which is unknown to the repo. Did you forget to call repo.add() for this entity?"
instance = listing_entity_to_model(entity)
merged = self.session.merge(instance)
self.session.add(merged)
def persist_all(self):
for entity in self._identity_map:
if entity is not REMOVED:
self.persist(entity)

Il y a quelques choses à noter ici :

• Nous passons une alchimie SQL Session et les instances Identity Map via le __init__. Il devrait y avoir une session et une carte d’identité par transaction commerciale.
• Lorsque add()Lors de la création d’une nouvelle entité dans un référentiel, nous stockons l’entité dans la carte d’identité et le modèle correspondant dans la session SQL Alchemy.
• Lorsque l’état de l’entité est modifié
  – pour conserver les modifications apportées à l’entité, utilisez persist() ou persist_all() méthodes. Lorsqu’une entité est persistante, son état est retraduit dans le modèle à l’aide de listing_entity_to_model mappeur de données.
  – persist()les changements et commit()La gestion de la session n’est pas la responsabilité d’un référentiel – et c’est intentionnel. La coordination des écritures dans la base de données doit être gérée par une unité de travail.
  – cette implémentation de référentiel n’est pas idéale en termes de consommation de mémoire (en fait, nous utilisons ici deux cartes d’identité : l’une qui est utilisée par un référentiel et l’autre qui fait partie de la session SQLA), mais je la considère comme suffisante pour nos besoins.

C’est ainsi que nous pourrions publier une annonce via publish_listing_use_case fonction:

engine = create_engine("sqlite+pysqlite:///:memory:",
echo=True,
future=True)
Base.metadata.create_all(engine)
def publish_listing_use_case(listing_id: ListingId, repository: ListingRepository):
listing = repository.get_by_id(listing_id)
listing.publish()
def execute_publish_listing_via_unit_of_work():
identity_map = []
with Session(engine) as session:
repository = SqlAlchemyListingRepository(session, identity_map)
publish_listing_use_case(listing_id=..., repository=repository)
repository.persist_all()
execute_publish_listing_via_unit_of_work()

Comme nous pouvons le voir, il n’y a pas de sauvegarde explicite de l’état de l’entité dans la base de données par le référentiel. Tous les gros travaux sont assurés par execute_publish_listing_via_unit_of_work fonction responsable de la création d’une session, de l’instanciation du référentiel, de l’appel d’une fonction de cas d’utilisation et de la sauvegarde de tous les résultats. En général, il s’agit de la responsabilité d’un Unité de travail.

Comme nous pouvons le constater, la mise en œuvre d’un modèle de référentiel représente beaucoup de travail. Nous devons définir un modèle de données pour notre objet de domaine, configurer le mappage entre ces deux, puis implémenter le référentiel. Nous avons également besoin d’une certaine logique pour maintenir une instance de modèle et une entité synchronisées afin que si l’entité change, elle sera automatiquement enregistrée dans une base de données. Une autre couche d’abstraction signifie certainement beaucoup plus de travail. Alors, est-ce que tout ça en vaut la peine ? Quand doit-on l’utiliser ? Quels sont les bénéfices?

Tout d’abord, nous devons utiliser le modèle de référentiel si nous avons l’intention de modifier l’état de l’entité en exécutant la logique métier. Ce sera exagéré si nous voulons seulement interroger les données (c’est-à-dire rechercher, trier, filtrer) et les afficher à l’écran.

Et à propos des avantages : nous obtenons une séparation des préoccupations : la couche métier n’a pas besoin de connaître la source de données ni de suivre les modifications. Les référentiels sont interchangeables – le code est plus facile à tester et plus maintenable à plus long terme.

***

Cet article a été publié pour la première fois dans DDD en Python