Implementation of Repository Pattern in Python?

Question

Mainly out of curiosity, I'm looking for a Python framework or example for the Repository Pattern of decoupling persistence logic from domain logic.

The name "Repository Pattern" appears in the post "Untangle Domain and Persistence Logic with Curator" (Ruby), idea comes from a section of the "Domain-Driven Design" book and Martin Fowler. The model class contains no persistence logic, rather the app declares repository subclasses whose instances act like in-memory collections of model instances. Each repository persists the model in different ways, for example to SQL (various schema conventions), to Riak or other noSQL and to memory (for caching). Framework conventions mean repository subclasses typically require minimal code: just declaring "WidgetRepository" subclass of SQLRepository would provide a collection that persists the model Widget to the DB table named "widgets" and match columns to Widget attributes.

Differences from other patterns:

Active Record Pattern: for example, Django ORM. The application defines just the model class with domain logic and some metadata for persistence. The ORM adds persistence logic to the model class. This mixes domain and persistence in one class (undesirable according to the post).

Thanks to @marcin I see that when Active Record supports diverse backends and .save(using="other_database") function, that gives the multi-backend benefit of the Repository Pattern.

So in a sense Repository Pattern is just like Active Record with the persistence logic moved to a separate class.

Data Mapper Pattern: for example, SQLAlchemy's Classical Mappings. The app defines additional classes for database table(s), and data mapper(s) from model to table(s). Thus model instance can be mapped to tables in multiple ways e.g. to support legacy schemas. Don't think SQLAlchemy provides mappers to non-SQL storage.

Answer 1

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)