在Python中實現存儲庫模式？

Question

主要是出於好奇，我正在尋找一個Python框架或Repository Pattern的例子，用於從域邏輯中解耦持久性邏輯。

「存儲庫模式」名稱出現在「Untangle Domain and Persistence Logic with Curator」（Ruby）文章中，想法來自「Domain-Driven Design」書籍的section和Martin Fowler。模型類不包含持久性邏輯，而是應用程序聲明其實例像模型實例的內存中集合一樣操作的存儲庫子類。每個存儲庫都以不同的方式保存模型，例如SQL（各種模式約定），Riak或其他noSQL以及內存（用於緩存）。框架約定意味着存儲庫子類通常只需要最少的代碼：只是聲明SQLRepository的「WidgetRepository」子類將提供一個集合，該集合將模型Widget保存到名爲「widgets」的數據庫表中，並將列與Widget屬性進行匹配。從其他模式

差異：

活動記錄模式：例如，Django的ORM。該應用程序只定義了具有域邏輯的模型類和用於持久性的一些元數據。 ORM將持久性邏輯添加到模型類。這在一個類中混合了域和持久性（根據帖子不合需要）。

感謝@marcin，我發現當Active Record支持不同的後端和.save（使用=「other_database」）函數時，它提供了Repository Pattern的多後端優勢。

因此，從某種意義上說，Repository Pattern就像Active Record一樣，將持久性邏輯移動到一個單獨的類中。

數據映射器模式：例如SQLAlchemy的古典映射。該應用程序爲數據庫表和數據映射器定義了額外的類，從模型到表格。因此模型實例可以以多種方式映射到表格，例如以支持傳統模式。不要以爲SQLAlchemy將映射器提供給非SQL存儲。

Answer 1

你可能想在一個很好看詹姆斯·丹尼斯DictShield project

「DictShield是一個數據庫無關的建模系統，它提供了一種容易建模，驗證和重塑數據。所有這些都不需要任何特定的數據庫「。

Answer 2

從我腦袋：

我定義了兩個例子域，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)