Реализация шаблона репозитория в Python?

Question

В основном из любопытства я ищу структуру или пример Python для шаблона репозитория логики переноса развязки логики домена.

Название "Repository Pattern" появляется в пост "Untangle Domain and Persistence Logic with Curator" (Ruby), идея исходит от section книги "Domain-Driven Design" и Martin Fowler. Класс модели не содержит логики постоянства, а приложение объявляет подклассы репозитория, экземпляры которых действуют как коллекции памяти экземпляров модели. Каждый репозиторий сохраняет модель по-разному, например, SQL (различные схемы), Riak или другой noSQL и в память (для кэширования). Соглашения о структуре означают, что подклассы репозитория обычно требуют минимального кода: просто объявление подкласса «WidgetRepository» SQLRepository предоставит коллекцию, которая сохраняет модельный виджет в таблице БД с именем «виджеты» и сопоставляет столбцы с атрибутами виджета.

Отличия от других моделей:

Active Record Pattern: например, Django ORM. Приложение определяет только класс модели с логикой домена и некоторыми метаданными для сохранения. ORM добавляет логику сохранения в класс модели. Это смешивает домен и постоянство в одном классе (нежелательно в соответствии с поста).

Благодаря @marcin, я вижу, что когда Active Record поддерживает различные функции backend и .save (using = "other_database"), это дает многогранное преимущество шаблона репозитория.

Так что в некотором смысле шаблон хранилища похож на активную запись с логикой сохранения, перемещенной в отдельный класс.

Data Mapper Pattern: например, классические сопоставления SQLAlchemy. Приложение определяет дополнительные классы для таблицы (ов) базы данных и сопоставления данных из модели в таблицу (таблицы). Таким образом, экземпляр модели может быть сопоставлен с таблицами несколькими способами, например. для поддержки устаревших схем. Не думайте, что SQLAlchemy предоставляет mappers для хранения, отличного от SQL.

Answer 1

Вы могли бы хотеть иметь хороший взгляд на Джеймса Dennis' DictShield project

«DictShield система моделирования баз данных агностиком. Это дает возможность моделировать, проверки и перекроить данные легко. Все, не требуя какой-либо конкретной базы данных «.

Answer 2

Из моей головы:

Я определяю два примера доменов, User и Animal, класс Store база хранения и два специализированных хранения классов UserStore и AnimalStore. Использование диспетчера контекста закрывает соединение db (для простоты я использую 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)