Implementation of Repository Pattern in Python?

I have been working on a repository pattern library. It is still under development but as of the time of writing, it currently supports In-memory lists (useful for unit tests), SQL databases (via SQLAlchemy), MongoDB and HTTP(s) REST APIs. It's actually quite handy for things where simple CRUD operations are enough. It is also quite useful for plugging in APIs.

Installation

pip install redbird

Some initiation examples

Repository in temporal memory

from redbird.repos import MemoryRepo
repo = MemoryRepo()

Repository for SQL

from sqlalchemy import create_engine
from redbird.repos import SQLRepo

engine = create_engine("sqlite://")
repo = SQLRepo.from_engine(engine=engine, table="my_table")

Repository for MongoDB

from redbird.repos import MongoRepo

repo = MongoRepo.from_uri(uri="mongodb://localhost:27017", database="my_db", collection="my_collection")

Repository for HTTP(s) REST API

from redbird.repos import RESTRepo

repo = RESTRepo(url="https://example.com/items", headers={"Authorization": "Bearer 123456"})

You may also specify a Pydantic model to have more structure and better documentation. Also, an ID field is supported. See more from https://red-bird.readthedocs.io/en/latest/repos/index.html

Usage

# Insert items
repo.add({"id": "a", "color": "red", "car_type": "van"})
repo.add({"id": "b", "color": "red", "car_type": "truck"})

# Read
list(repo) # Get all cars in the repo

repo.filter_by(color="red").all() # Get all red cars
repo.filter_by(color="red").first() # Get first red car
repo.filter_by(color="red").last() # Get last red car

# Update
repo.filter_by(car_type="van").update(color="red")

# Delete
repo.filter_by(car_type="van").delete()

The read operations return a dict by default but will be Pydantic instance if model was specified in the repository.

Note that it is still under development (so significant changes may come) but it is already quite useful and fully functional.

Links:

• Documentations
• Source code
• Releases

You might want to have a good look at James Dennis' DictShield project

"DictShield is a database-agnostic modeling system. It provides a way to model, validate and reshape data easily. All without requiring any particular database."

Out of my head:

I define two example domains, User and Animal, an base storage class Store and two specialized Storage classes UserStore and AnimalStore. Use of context manager closes db connection (for simplicity I use sqlite in this example):

import sqlite3

def get_connection():
    return sqlite3.connect('test.sqlite')

class StoreException(Exception):
    def __init__(self, message, *errors):
        Exception.__init__(self, message)
        self.errors = errors


# domains

class User():
    def __init__(self, name):
        self.name = name


class Animal():
    def __init__(self, name):
        self.name = name


# base store class
class Store():
    def __init__(self):
        try:
            self.conn = get_connection()
        except Exception as e:
            raise StoreException(*e.args, **e.kwargs)
        self._complete = False

    def __enter__(self):
        return self

    def __exit__(self, type_, value, traceback):
        # can test for type and handle different situations
        self.close()

    def complete(self):
        self._complete = True

    def close(self):
        if self.conn:
            try:
                if self._complete:
                    self.conn.commit()
                else:
                    self.conn.rollback()
            except Exception as e:
                raise StoreException(*e.args)
            finally:
                try:
                    self.conn.close()
                except Exception as e:
                    raise StoreException(*e.args)


# store for User obects
class UserStore(Store):

    def add_user(self, user):
        try:
            c = self.conn.cursor()
            # this needs an appropriate table
            c.execute('INSERT INTO user (name) VALUES(?)', (user.name,))
        except Exception as e:
            raise StoreException('error storing user')


# store for Animal obects
class AnimalStore(Store):

    def add_animal(self, animal):
        try:
            c = self.conn.cursor()
            # this needs an appropriate table
            c.execute('INSERT INTO animal (name) VALUES(?)', (animal.name,))
        except Exception as e:
            raise StoreException('error storing animal')

# do something
try:
    with UserStore() as user_store:
        user_store.add_user(User('John'))
        user_store.complete()

    with AnimalStore() as animal_store:
        animal_store.add_animal(Animal('Dog'))
        animal_store.add_animal(Animal('Pig'))
        animal_store.add_animal(Animal('Cat'))
        animal_store.add_animal(Animal('Wolf'))
        animal_store.complete()
except StoreException as e:
    # exception handling here
    print(e)

I have written a python repository implementation using SqlAlchemy as the backend. I was looking for one, and couldn't find one, so I decided to make my own.

But I think there is one useful feature of a repository which you only touched on, particularly in the context of DDD, and with respect to the Active Record pattern, which is that the Repository pattern enables the model's interface to be couched in domain language, which means you can talk about it without thinking or knowing about the implementation. The Repository pattern helps keep the model's interface aligned with how the concepts are actually used by domain experts.

Let's say your model is a Car. Now, Cars can probably drive(), they can probably steer(), and so on. What they probably can't do, is save(). save(), and the concept of saving, are things that belong in a different abstraction to do with persistence.

It may seem like a small thing, but it can be really useful to keep the model's interface well aligned with the domain language, because it means you can have easy and clear conversations with clients, without worrying about the implementation details.