在 Python 中实现存储库模式？

Question

主要出于好奇，我正在寻找一个 Python 框架或用于将持久性逻辑与域逻辑解耦的存储库模式的示例。

名称“存储库模式”出现在帖子“Untangle Domain and Persistence Logic with Curator”（Ruby）中，想法来自“领域驱动设计”一书的section 和Martin Fowler。模型类不包含持久性逻辑，而是应用程序声明存储库子类，其实例的作用类似于模型实例的内存集合。每个存储库以不同的方式保存模型，例如 SQL（各种模式约定）、Riak 或其他 noSQL 以及内存（用于缓存）。框架约定意味着存储库子类通常需要最少的代码：只需声明 SQLRepository 的“WidgetRepository”子类将提供一个集合，该集合将模型 Widget 持久保存到名为“widgets”的数据库表中，并将列与 Widget 属性匹配。

与其他模式的区别：

活动记录模式：例如，Django ORM。应用程序只定义了具有域逻辑的模型类和一些用于持久性的元数据。 ORM 将持久性逻辑添加到模型类。这将域和持久性混合在一个类中（根据帖子，这是不受欢迎的）。

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

所以在某种意义上，Repository Pattern 就像 Active Record 一样，将持久性逻辑移到了一个单独的类中。

Data Mapper Pattern：例如，SQLAlchemy 的 Classical Mappings。该应用程序为数据库表定义了额外的类，以及从模型到表的数据映射器。因此模型实例可以通过多种方式映射到表，例如支持遗留模式。不要认为 SQLAlchemy 为非 SQL 存储提供映射器。

Answer 1

I have written a python repository implementation using SqlAlchemy as the backend.我在找一个，但找不到，所以我决定自己做一个。

但我认为存储库有一个有用的特性，您只涉及过，特别是在 DDD 的上下文中，并且就 Active Record 模式而言，即存储库模式使模型的接口能够嵌入领域语言，这意味着您可以在不考虑或不了解实现的情况下谈论它。存储库模式有助于使模型的界面与领域专家实际使用概念的方式保持一致。

假设您的模型是Car。现在，Cars 大概可以drive()，他们大概可以steer()，等等。他们可能做不到的是save()。 save() 和保存的概念是属于与持久性有关的不同抽象的事物。

这似乎是一件小事，但让模型的界面与领域语言保持良好的一致性非常有用，因为这意味着您可以与客户进行简单而清晰的对话，而不必担心实现细节。

Answer 2

您可能想好好看看 James Dennis 的 DictShield project

“DictShield 是一个与数据库无关的建模系统。它提供了一种轻松建模、验证和重塑数据的方法。所有这些都不需要任何特定的数据库。”

Answer 3

想不通：

我定义了两个示例域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)