Forem: Przemysław Górecki

Implementing the Repository pattern in Python using SqlAlchemy as a backend

Przemysław Górecki — Wed, 16 Nov 2022 13:23:05 +0000

In the previous post, we implemented a really simple file-based repository that used the pickle module. As we didn't use a database, we didn't have to deal with creating tables and columns, and all the domain models were directly serialized from a memory to a byte stream (and vice versa). Such an approach was ok to introduce the concept of a repository, but in a production system we will need some kind of a database to store the application state.

As a side note, last time I wrote that repositories are used in loading and saving entities. In fact, repositories work with aggregates (which are composites of entities) but for the sake of simplicity, I'm sticking with entities.

This time we are going to use SQL Alchemy as a persistence mechanism for the entities. We will keep the original interface, but all the implementation details will be encapsulated in the SqlAlchemyListingRepository. Let's start with a revised interface to a ListingRepository, which additionally supports adding, removing, and persisting Listing domain entities:

# 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)

In Domain-Driven Design, the goal of a repository is to encapsulate the logic required to access domain objects. Therefore, the interface to a repository is a part of a domain layer, while the actual implementation of this interface belongs to the infrastructure layer. As a consequence, the domain code stays clean - it is isolated from any technical concerns, and it is database agnostic.

A bit of theory...

Here are some concepts I'd like to introduce and discuss before we jump right into the implementation of the SQL Alchemy repository:

1. ORM Model ≠ Domain Object

Entities are Python classes - this is not a surprise. They contain attributes: primitive types, value objects, enums, references to other entities, and collections of all the above. In memory, this forms a graph-like structure of nested fields. Just as a reminder, this is how our Listing entity looks like (at this point we do not care about the logic):

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

On the other hand, databases organize data in a flat, tabular fashion. Python classes and relational data models do not work very well together, and ORMs were introduced to overcome this problem. However, because of the mapping between model attributes and table columns, ORM models are still tightly coupled to a database and a specific ORM implementation:

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))

This raises 2 problems:

ORM Models shouldn't be used in the domain layer. ORM Models are harder to test when compared to Domain Objects (they need a database to exist). Also, we want the business layer to remain clean, without any dependencies on underlying storage mechanisms. ListingModel is tightly coupled to SQL Alchemy, and having such a dependency in a business layer is a no-go. However, it's still fine to use SQLA Models in some cases, i.e. if we do basic CRUD operations that do not require business logic.
There is no easy way to have a multi-field value object (i.e. min_price Money composed of amount and currency) to multiple columns in the model. Theoretically, we could add some getters and setters for Value Objects:

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

...but this contradicts the idea of value objects being immutable (you can still change an individual attribute of a model that logically belongs to a value object).

To overcome these problems, we need a mechanism to transfer data between a business layer - a data mapper.

2. Data Mapper

We already know that we don't want to use Models in the domain layer. Models are SQL Alchemy-specific, as they are composed of Columns, ForeignKeys, relationships, etc. We do not want any of the infrastructure details to leak into the domain. We could still use Models for querying data (as it's okay to query the database through other channels than repos), but this is beyond the scope of this article.

What we need to do is to:

map models to entities when reading data from a repo (i.e. get_by_id(...) call returns a domain entity),
map entities to models when any data changes are about to be saved.

This can be implemented in 2 ways:

We can use SQL Alchemy imperative mapping to map data from SQL tables to pure Python classes. Cosmic Python follows this approach. Here, all the attributes of a model are automatically translated from/to a domain object, via mapper_registry.map_imperatively(). However, this technique is limited to models consisting only of primitive types. If our domain model has multi-attribute value objects, then we need a more sophisticated approach.
We can implement our own data mapper for translating to and from our domain model. To do the mapping we will need 2 functions: XYZ_model_to_entity and XYZ_entity_to_model. The repository will use the data mapper to do all the translations. This is the approach we are going to take in the implementation of SqlAlchemyListingRepository.

3. Identity Map

One of the essential elements in the repository pattern is the Identity Map, which is used to improve performance by providing a context-specific, in-memory cache to prevent duplicate retrieval of the same object data from the database during a single transaction. Here is another definition:

An Identity Map keeps a record of all objects that have been read from the database in a single business transaction. Whenever you want an object, you check the Identity Map first to see if you already have it.
Martin Fowler

This is pretty straightforward: when we query the repository for an entity, we will check the identity map first. If the entity is already present in the map, the repository will simply return a reference to a cached object. Otherwise, we will read the model from a database, transform it with a data mapper, store it in the identity map and return a reference. When saving an entity, we will use a data mapper to transform it into a model instance and then persist it using the built-in SQL Alchemy session mechanism.

The repository pattern implementation

Now that we know what are the building blocks of a repository, let's look at all the pieces that we need to implement the SQL Alchemy Listing Repository.

Domain objects

Here is the data part of our domain model. For clarity reasons, all the logic was removed from domain objects.

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)

Here are the interesting things to notice. A Listing is composed of some primitive types (i.e. id, name), a single value object (min_price), and a list of value objects (bids). For some undisclosed reason, let's assume that we intentionally use a list so that we can preserve the order of bids (otherwise we could use an unordered set).

Data model

The above domain objects are reflected as database models as follows:

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")

As you can see, we do not have a separate table for Money - these are stored as primitive fields in the corresponding models (i. e. price__amount, price__currency and min_price__amount, min_price__currency). However, we are storing bids in a separate BidModel as there is a 1 to many relationship between ListingModel and BidModel. Also, bids are ordered by idx field to preserve the order of the elements in Listing.bids list, and there is a cascade option set in such a way to keep the Listing.bids and the table rows in sync when Bid value objects are removed from the list. Also, take note that is using a composite primary key composed of listing_id and idx as it is enough to identify a bid at a database level.

Data mapper

The data mapper logic is nothing fancy - just a rather tedious mapping from one type to the other. One small interesting bit to notice is packing/unpacking value objects.

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)]

Listing Repository

Finally, let's look at the implementation of a repository:

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)

There are a couple of things worth noting here:

we are passing a SQL AlchemySession and the Identity Map instances via the __init__. There should be one session and one identity map per business transaction.
when add()ing a new entity to a repository, we store the entity in the identity map and the corresponding model in the SQL Alchemy session.
when entity state is being changed
to persist the changes made to the entity, use persist() or persist_all() methods. When an entity is persisted, its state is translated back to the model using listing_entity_to_model data mapper.
persist()ing the changes and commit()ing the session is not a responsibility of a repository - and this is intended. Coordinating the writes to the database should be handled by a Unit Of Work.
this repository implementation is not ideal in terms of memory consumption (in fact we use 2 identity maps here: one which is used by a repo, and the other one which is a part of SQLA session) but I consider it to be good enough for our purposes.

The use case

This is how we could publish a listing via publish_listing_use_case function:

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()

As we can see, there is no explicit saving of the entity state in the database by the repository at this point. All the heavy lifting is handled by execute_publish_listing_via_unit_of_work function which is responsible for creating a session, instantiating the repository, calling a use case function, and saving all the results. In general, this is a responsibility of a Unit of Work which will be introduced later.

Conclusion & Final remarks

As we can see, the actual implementation of a repository pattern is a lot of work. We need to define a data model for our domain object, configure the mapping between these two, then implement the repository itself. We also need some logic to keep a model instance and entity in sync, so that if the entity changed it will be saved automatically in a database. Another layer of abstraction certainly means much more work. So is it all worth it? When should we use it? What are the benefits?

First of all, we should use the repository pattern if our intention is to change the state of the entity by executing the business logic. It would be an overkill if all we want to do is to query the data (i.e. searching, sorting, filtering) and display it on the screen.

And about the benefits: we get a separation of concerns - the business layer does not need to know the data source nor track the changes. Repositories are interchangeable - the code is easier to test and more maintainable in the longer perspective.

This article was first published in DDD in Python.

Implementing Value Objects in Python

Przemysław Górecki — Wed, 01 Dec 2021 09:18:13 +0000

A Value Object is one of the fundamental building blocks of Domain-Driven Design. It is a small object (in terms of memory), which consists of one or more attributes, and which represents a conceptual whole. Value Object is usually a part of Entity.

Some examples of value objects are: Email (consisting of a single email attribute), Money (consisting of amount and currency), DateRange (consisting of a start date, and an end date), GPSCoordinates (made of latitude and longitude), or Address (consisting of a street, zip code, city, state, etc.). Apart from the attributes, all of the above can (and should) include some kind of validation logic too.

As you can see from the examples above, value objects do not have an identity - they are simply a collection of attributes that are related to each other.

Here are the most important properties of a value object:

Its state is immutable. Once created, the state of a value object cannot be changed.
It is distinguishable only by the state of its attributes. Two instances with the same attribute values are considered to be equal (this is also known as structural equality).
It should encapsulate business logic that prevents us from constructing a value object with an invalid state (i.e. start date < end date for a date range).
All methods of a value object should be pure, i.e. calling a method does not trigger any side effects or change the state of a value object. However, returning a new instance that reflects the changes is fine.
It should be easy to unit-test a value object and it should be easy to reason about its logic.

To recognize a value object in your domain model, mentally replace it with a tuple with some validation logic and a few extra methods that are easy to test.

Let's implement a DateRange value object using Python dataclasses module:

from dataclasses import dataclass
from datetime import date

class BusinessRuleValidationException(Exception):
  """A base class for all business rule validation exceptions"""

class ValueObject:
  """A base class for all value objects"""

@dataclass(frozen=True)
class DateRange(ValueObject):
  """Our first value object"""
  start_date: date
  end_date: date

  def __post_init__(self):
    """Here we check if a value object has a valid state."""
    if not self.start_date < self.end_date
      raise BusinessRuleValidationException("end date date should be greater than start date")

  def days(self):
    """Returns the number of days between the start date and the end date"""
    delta = self.end_date - self.start_date + timedelta(days=1)
    return delta.days

  def extend(self, days):
    """Extend the end date by a specified number of days"""
    new_end_date = self.end_date + timedelta(days=days)
    return DateRange(self.start_date, new_end_date)

Re 1: To guarantee immutability of a DateRange, we are using @dataclass(frozen=True) decorator.

Re 2: Equality is guaranteed by a dataclass itself, which compares the class instance as if it were a tuple of its fields.

Re 3: We validate the state of an instance in __post_init__ method using simple logic to check invariants. It prevents us from creating an invalid date range.

Re 4: Our value object has only 2 methods: days and extend. Both of them are pure (they are side effects free). Note that extend returns a new instance of DateRage instead of modifying the end_date attribute.

Re 5: Thanks to its simple behavior, unit testing DateRage is also relatively straightforward:

import unittest

class DateRangeTestCase(unittest.TestCase):
    def test_equality(self):
        range1 = DateRange(start_date=date(2020,1,1), end_date=date(2021,1,1))
        range2 = DateRange(start_date=date(2020,1,1), end_date=date(2021,1,1))
        self.assertEqual(range1, range2)

    def test_days(self):
        range = DateRange(start_date=date(2020,1,1), end_date=date(2020,1,1))
        self.assertEqual(range.days(), 1)

    def test_days(self):
        range1 = DateRange(start_date=date(2020,1,1), end_date=date(2020,1,1))
        range2 = range1.extend(days=1)
        self.assertEqual(
            range2, 
            DateRange(start_date=date(2020,1,1), end_date=date(2021,1,2))
        )

    def test_cannot_create_invalid_date_range(self):
        with self.assertRaises(BusinessRuleValidationException):
            DateRange(start_date=date(2021,1,1), end_date=date(2020,1,1))

Using value objects in your code will also help you in fighting with primitive obsession. Why is it important? Let me give you an example to illustrate the problem. Let's say that you decide to cut corners and use string to represent emails. There is a high chance that you will need to validate those emails as well, and most likely you will need to do it in multiple places (i.e. user inputs, form data, serializers, business logic, etc.). Having a simple Email value object will help you to stay DRY in the long run.

This article was first published in DDD in Python.

Where to place the business logic in Domain-Driven Design (Python example)

Przemysław Górecki — Sat, 06 Nov 2021 13:49:28 +0000

In Domain-Driven Design we have a clear sepatartion between the Domain, Application, and Infrastructure. If we are talking about the business rules, then then the most obvious choice is the Domain layer, but sometimes it could be OK to place the logic in the Application layer. Let's see what options do we have here:

Option 1 - put your business logic in an application service layer

This is the simplest scenario. If your domain model is very simple (i.e. CRUD based), then putting the business logic directly in a transaction script or a request handler might be ok. However, I don't like this idea if we have a non-trivial business logic.

Option 2 - put your business logic in domain models.

This is a really reasonable approach. To most natural place to put the business logic concerning an aggregate is the aggregate itself.

Option 3 - put your business logic in a domain services.

Sometimes, the business logic goes beyond a single aggregate, or spans multiple aggregates. In this case we need to take into consideration both the aggregate and it's surrounding, and this responsibility can be passes to the domain service.

Since option 1 is pretty straightforward, let's get deeper into options 2 and 3.

Where should I put a businsess logic concerning a single aggregate?

We want to change the state of the system by executing actions (commands) on the Aggregates. Those actions can be exposed via methods of a public API of the Aggregate. Those methods are accessible from an Aggregate Root.

Any references from outside the aggregate should only go to the aggregate root. The root can thus ensure the integrity of the aggregate as a whole.Any references from outside the aggregate should only go to the aggregate root. The root can thus ensure the integrity of the aggregate as a whole.
-- Martin Fowler

So an Aggegate Root acts as a facade of the aggregate, and it is responsible of enforcing the aggregate consistency rules. For each public method, the invariants are checked using check_rule method and the business logic is applied to an aggretage. Let's consider an Auction domain as an example. Our use case here is to allow users (sellers) to publish their items for sale (listings) on an auction website similar to Ebay. The business rules are as following:

Sellers cannot list items for free
Only listing drafts can be published - a listing that is already published cannot be published again.

class AggregateRoot:
    def check_rule(...):
        ...

class Listing(Entity):
    """A class representing an item for sale"""
    status: ListingStatus = ListingStatus.Draft

    def publish():
        self.status = ListingStatus.Published

    def unpublish():
        self.status = ListingStatus.Draft

class Seller(AggregateRoot):
    """A class representing a seller willing to list a listing for sale"""
    id: UUID
    published_listings_count: int = 0

    def publish_listing(self, listing: Listing):
        """This method is a part of a public Seller API"""
        # validate listing
        self.check_rule(ListingPriceMustBeGreaterThanZero(listing))
        self.check_rule(ListingMustBeDraft(listing))
        # do some business logic
        listing.publish()
        self.published_listings_count += 1
        # check aggregate invariants
        self.check_rule(SellerCanHaveUpToThreeListingsInTheCatalog(self.published_listings_count))

Where should I put logic concerning two or more aggregates?

Sometimes the business logic spans multiple aggregates. A common scenario here is to ensure uniqueness. Let's, we have a User entity in our system. The business are as following:

User can change it's username, but no more that once a month (30 days)
Username of a Usermust by unique within a system.

The first rule can be easily check within an aggregate:

class User(AggregateRoot):
    id: UUID
    username: str
    username_changed_at: date

    def change_username(username: str):
        self.check_rule(UsernameCanBeChangedAtMostOnceAMonth(self.username_changed_at))
        self.username = username
        username_changed_at = date.today()

But how we can guarantee the uniqueness of a new username? I think the only plausible choice here is to use the domain service. We could create a UsernameUniquenessChecker domain service to handle the job:

class UsernameUniquenessChecker:
    def __init__(self, user_repository):
        self.user_repository = user_repository

    def is_unique(username: str) -> bool:
        if self.user_repository.find_user_by_username(username):
            # user with this username exists, so it's not unique
            return False
        return True

class User(AggregateRoot):
    id: UUID
    username: str
    username_changed_at: date

    def change_username(username: str, username_uniqueness_checker: UsernameUniquenessChecker):
        self.check_rule(UsernameCanBeChangedAtMostOnceAMonth(self.username_changed_at))
        if not username_uniqueness_checker.is_unique(username):
            raise BusinessRuleValidationException("Username must be unique")
        self.username = username
        username_changed_at = date.today()

How to track request logs using correlation id

Przemysław Górecki — Mon, 06 Sep 2021 06:29:11 +0000

Some time ago I stumbled across an interesting problem. In a Python web app, I wanted to add a correlation_id to each log produced by the request handler. This would allow me to easily capture all the logs that happen within a request, even if requests are being processed concurrently:

[480f4c31-3ceb-45be-afda-5676e59cc391 info] RegisterUserCommand handling started
[480f4c31-3ceb-45be-afda-5676e59cc391 info] Processing RegisterUserCommand command with params {'name': 'Alice', ...}
[249ae775-845d-456c-ab71-48776fcad938 info] RegisterUserCommand handling started
[249ae775-845d-456c-ab71-48776fcad938 info] Processing RegisterUserCommand command with params {'name': 'Bob', ...}
[249ae775-845d-456c-ab71-48776fcad938 info] RegisterUserCommand handling completed
[480f4c31-3ceb-45be-afda-5676e59cc391 warn] RegisterUserCommand handling failed

As you can see, without correlation_id it is impossible to tell from the logs which command has failed.

So how can we deal with this problem? This sounds easy at the beginning. We could simply instantiate a logger when route handling starts, easy peasy:

import uuid

class Logger:
  """Not a real logger, but it is enough for the purpose of this post"""
  def __init__(self, correlation_id):
    self.correlation_id = correlation_id

  def info(self, *args, **kwargs):
    print(f'[{self.correlation_id} info]', *args, **kwargs)

@app.post("/register")
def register(payload):
    logger = Logger(uuid.uuid4())
    logger.info("Registering user", payload)
    ...
    return {"result": "OK"}

The first problem with the above code is that you need to add logger = Logger(uuid.uuid4()) to every route handler. We can solve this problem by injecting the logger into the handler function. How you do it depends on a framework, but let’s use a create_logger factory function for now.

def create_logger(): 
  logger = Logger(uuid.uuid4())

@app.post("/register")
def register(payload, logger = create_logger()):
    logger.info("Registering user", payload)
    return {"result": "OK"}

As a result, with each request, a new logger instance (containing a unique id) will be created. Let’s move on and add more dependencies (DatabaseEngine, UserService) to our handler:

class DatabaseEngine:
  def __init__(self, logger:Logger):
    self.logger = logger   
    ...
  def connect(self):
    ...
  def execute(self, query):
    self.logger.info('Executing DB query...')
    ...

class UserService:
  def __init__(self, logger:Logger, db_engine:DatabaseEngine):
    self.logger = logger
    self.db = db_engine
  def register(self, payload):
    self.logger.info('RegisterUserCommand handling started for', payload)
    self.db.execute(...)
    time.sleep(1)
    self.logger.info('RegisterUserCommand handling completed for', payload)

def create_user_service():
  logger = Logger(uuid.uuid4())
  db_engine = DatabaseEngine(logger)
  db_engine.connect()
  user_service = UserService(logger=logger, db_engine=db_engine)
  return user_service

@app.post("/register")
def register(payload, service:UserService = create_user_service()):
    logger = service.logger # ugh!
    logger.info("Registering user", payload)
    result = service.register(payload)
    return {"result": result}

Here is the second problem. Since a logger is a dependency for both UserService and DatabaseEngine, we must create a logger first. Moreover, since we instantiate a logger with every request, every other object that depends on a logger must be instantiated with every request as well. So having a global db_engine object is not an option unless we give up on a per-request logger in favor of a global logger, which is something we don’t want to do. Instantiating a whole dependency graph for the application on every request just because of the per-request logger seems like overkill in terms of performance and resource management. If you think that “there must be a better way” you are right. Let’s explore our options.

Global logger

Why do we need a per-request logger anyway? The official Python logging cookbook says:

While it might be tempting to create Logger instances on a per-connection basis, this is not a good idea because these instances are not garbage collected [source]

If we make correlation_id a global variable, then a logger could be a global object (or singleton if you prefer), referencing this global variable. As a result, both db_engine and servive could use the global logger, which means that the entire dependency graph could be constructed at once, on the application startup.

correlation_id = None

class Logger:
  def info(self, *args, **kwargs):
    print(f'[{correlation_id} info]', *args, **kwargs)

# application startup
logger = Logger()
db_engine = DatabaseEngine(logger)
db_engine.connect()

def create_user_service():
  user_service = UserService(logger=logger, db_engine=db_engine)
  return user_service

@app.post("/register")
def register(payload, service:UserService = create_user_service()):
    global correlation_id
    correlation_id = uuid.uuid4() # we need to generate new uuid for each request
    logger.info("Registering user", payload)
    result = service.register(payload)
    return {"result": result}

This approach is naive and it will work only if the requests are processed sequentially, one after another. No multithreading, no concurrency is possible here. If Thread-2 changes correlation_id, while Thread-1 is still processing its own request, logs will be corrupted. Let's simulate this behavior:

threads = [
           threading.Thread(target=register, args=(payload,)) 
           for payload in ['Alice', 'Bob']
          ]

for thread in threads:
  thread.start()

for thread in threads:
  thread.join()

As expected, the logs are completely messed up:

[e2931dc9-ca83-4d79-9cfc-23aba03d264a info] Registering user Alice
[e2931dc9-ca83-4d79-9cfc-23aba03d264a info] RegisterUserCommand handling started for Alice
[605e6790-387a-4fbe-ba08-874b57f8e0d6 info] Registering user Bob
[605e6790-387a-4fbe-ba08-874b57f8e0d6 info] RegisterUserCommand handling started for Bob
[605e6790-387a-4fbe-ba08-874b57f8e0d6 info] Executing DB query...
[605e6790-387a-4fbe-ba08-874b57f8e0d6 info] Executing DB query...
[605e6790-387a-4fbe-ba08-874b57f8e0d6 info] RegisterUserCommand handling completed for Bob
[605e6790-387a-4fbe-ba08-874b57f8e0d6 info] RegisterUserCommand handling completed for Alice

Thread context to the rescue

If we will be processing incoming requests using threads, we could use Python's threading.local() to store thread-specific information, correlation_id in our case.

class ContextLogger:
  """
  This time we have a logger which is using threading.local()
  """
  def __init__(self, thread_context):
    self.thread_context = thread_context

  def info(self, *args, **kwargs):
    print(f'[{self.thread_context.correlation_id} info]', *args, **kwargs)


request_context = threading.local()
logger = ContextLogger(request_context)

...

@app.post("/register")
def register(payload, service:UserService = create_user_service()):
    request_context.correlation_id = uuid.uuid4() # we store new uuid per thread
    ...

Context Variables

We can use a conceptually similar approach when dealing with coroutines. If our framework supports concurrent requests running in a single thread via async/await calls, we need to use ContextVars instead of threading.local. This is due to the event loop running in a single thread. Here is the updated version of the logger:

class ConcurrentContextLogger:
  def __init__(self, correlation_id: ContextVar):
    self.correlation_id = correlation_id

  def info(self, *args, **kwargs):
    print(f'[{self.correlation_id.get()} info]', *args, **kwargs)

correlation_id: ContextVar[uuid.UUID] = ContextVar('correlation_id', default=uuid.UUID('00000000-0000-0000-0000-000000000000'))
logger = ConcurrentContextLogger(correlation_id)

...

@app.post("/register")
async def register(payload, service:UserService = create_user_service()):
    correlation_id.set(uuid.uuid4()) # we store new uuid per coroutine
    ...

Bonus

So far, we have used a dummy logger to mimic the functionality of a real logger. This is how we can add a correlation_id to the logging.logger instance:

class ContextFilter(logging.Filter):
    def filter(self, record):
        record.correlation_id = correlation_id.get()
        return True  


logging.basicConfig(
  level=logging.DEBUG,
  format='%(asctime)-15s %(name)-5s %(levelname)-8s %(correlation_id)s %(message)s'
)

context_filter = ContextFilter()
logger = logging.getLogger('app')
logger.addFilter(context_filter)

logger.info('Hello World!')

Demo

If you want to play with the code, I've prepared a repl for you: https://replit.com/@pgorecki/request-logger?v=1

Have fun logging!

This article was first published in DDD in Python.

Looking for a component-based Django template rendering library

Przemysław Górecki — Thu, 01 Apr 2021 11:56:32 +0000

One of the greatest features of React is the possibility of defining components - independent and reusable bits of HTML code. I really miss it in Django templates. I imagine this could work like so:

{% component Link %}
  <a href="{{href}}">{{children}}</a>
{% endcomponent %}

{% component Card %}
  <div class="card">
    {%if title}
      <div class="card-title">{{title}}</div>
    {%endif}
    <div class="card-content">
      {{children}}
    </div>
  </div>
{% endcomponent %}

{% usecomponents %}
<body>
  <Card title="Foo">
    I am a very simple card. I am good at containing small bits of information. I am convenient because I require little markup to use effectively.
    <Link href="#">Click Me</Link>
  </Card>
</body>
{% endusecomponents %}

How do you like the idea? Are you aware of any libraries that solves this problem?

Introducing django-cancan: authorization library for Django

Przemysław Górecki — Fri, 06 Nov 2020 11:48:10 +0000

I'd like to share with you my Django authorization library (https://github.com/pgorecki/django-cancan) I wrote some time ago and which I'm happily using for couple of my projects now.

My main motivation was to create a rule-based permission system which can be used both for checking per-object access rights and for generating access-based querysets. As a result, you do not need to write your own querysets for retrieving accessible objects, django-cancan will handle it automatically. For example, cancan can generate a queryset of articles which can be modified by the current user, based on a the user's role or permissions.

Here is an actual usage example from one my latest projects, with chat functionality: the exchange of messages between customers and consultants.

The data model is pretty simple:

class Thread(models.Model):
    customer = models.ForeignKey(User, ...)
    consultant = models.ForeignKey(User, ...)

class Message(models.Model):
    thread = models.ForeignKey(Thread, ...)
    content = models.TextField()

First, you define a function that grants actions based on user role.

def define_access_rules(user, rules):
    # user == request.user
    if not user.is_authenticated:
        return

    rules.allow("create", Thread)
    rules.allow("create", Message)
    rules.allow("view", Thread, customer=user)
    rules.allow("view", Message, thread__customer=user)
    rules.allow("view", Message, thread__consultant=user)

    if user.is_staff:
        rules.allow("view", Thread, consultant=user)
        rules.allow("view", Message)

    if user.is_superuser:
        rules.allow("view", Thread)
        rules.allow("view", Message)

In this example, client (logged-in user) is allowed to create conversation threads, view all threads he started, including all messages within those threads. On top of that, consultant (is_staff=True) can list threads he is assigned to, and superusers have unlimited access to both models.

Note that rules.allow use the same syntax as Model.objects.filter.

Then the access control is implemented by overriding the get_queryset method to return a list of Threads accessible by request.user and by adding has_permission method to check if user can perform a view action.

class ThreadList(ListView):
    model = Thread

    def get_queryset(self):
        ability = request.ability
        return ability.queryset_for("view", Thread)

class ThreadDetail(PermissionRequiredMixin, DetailView):
    model = Thread

    def test_func(self):
        ability = request.ability
        thread = self.get_object()
        return ability.can("view", thread)

Check out project README for more examples, including template-level checks, DRF integrations, etc.

I hope you will enjoy django-cancan as much as I do :)