在Python中实现存储库模式？

Question

主要是出于好奇,我正在寻找一个Python框架或示例,用于将持久性逻辑与域逻辑分离的Repository Pattern.

命名为"库模式"出现在帖子" Untangle的域名和持久性逻辑与馆长 "(红宝石),主意来自一个部分的"领域驱动设计"一书和马丁·福勒.模型类不包含持久性逻辑,而是应用程序声明存储库子类,其实例的作用类似于模型实例的内存中集合.每个存储库以不同的方式持久化模型,例如SQL(各种模式约定),Riak或其他noSQL以及内存(用于缓存).框架约定意味着存储库子类通常需要最少的代码:只是声明SQLRepository的"WidgetRepository"子类将提供一个集合,该集合将模型Widget持久保存到名为"widgets"的DB表,并将列与Widget属性匹配.

与其他模式的差异:

活动记录模式:例如,Django ORM.应用程序仅定义具有域逻辑的模型类和用于持久性的一些元数据.ORM将持久性逻辑添加到模型类.这将域和持久性混合在一个类中(根据帖子不合需要).

感谢@marcin我看到当Active Record支持多种后端和.save(using ="other_database")函数时,它提供了存储库模式的多后端优势.

所以在某种意义上,Repository Pattern就像Active Record一样,持久性逻辑被移动到一个单独的类中.

数据映射器模式:例如,SQLAlchemy的Classical Mappings.该应用程序定义了数据库表的附加类,以及从模型到表的数据映射器.因此,模型实例可以以多种方式映射到表,例如,以支持遗留模式.不要以为SQLAlchemy为非SQL存储提供映射器.

Answer 1

意料之外:

我定义两个示例结构域,User以及Animal,一个基础存储类Store和两个专门存储类UserStore和AnimalStore.使用上下文管理器关闭数据库连接(为简单起见,我在本例中使用sqlite):

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)

Answer 2

我写了一篇关于使用 SQLAlchemy 作为后端实现存储库模式的文章。它遵循数据映射器方法。