Implementación del patrón de repositorio en Python?

Question

Principalmente por curiosidad, estoy buscando un marco de Python o un ejemplo para el patrón de repositorio de la lógica de persistencia de desacoplamiento de la lógica del dominio.

El nombre "Patrón repositorio" aparece en el post "Untangle Domain and Persistence Logic with Curator" (Rubí), idea proviene de un section del libro "Diseño de dominio-Driven" y Martin Fowler. La clase de modelo no contiene lógica de persistencia, sino que la aplicación declara subclases de repositorio cuyas instancias actúan como colecciones en memoria de instancias de modelo. Cada repositorio persiste en el modelo de diferentes maneras, por ejemplo en SQL (varias convenciones de esquema), en Riak u otro noSQL y en la memoria (para el almacenamiento en caché). Las convenciones de framework significan que las subclases de repositorio generalmente requieren un código mínimo: la simple declaración de la subclase "WidgetRepository" de SQLRepository proporcionaría una colección que persistirá en el widget de modelo en la tabla de DB denominada "widgets" y las columnas de coincidencia con los atributos de widgets.

Diferencias con otros patrones:

activo Grabar patrón: por ejemplo, ORM de Django. La aplicación define solo la clase de modelo con lógica de dominio y algunos metadatos de persistencia. El ORM agrega lógica de persistencia a la clase de modelo. Esto mezcla el dominio y la persistencia en una clase (indeseable según la publicación).

Gracias a @marcin veo que cuando Active Record admite diversos backends y la función .save (using = "other_database"), eso le da al beneficio de varios recursos del Repository Pattern.

Por lo tanto, en cierto sentido, el Patrón de repositorio es como el Registro activo con la lógica de persistencia trasladada a una clase separada.

Data Mapper Pattern: por ejemplo, asignaciones clásicas de SQLAlchemy. La aplicación define clases adicionales para la (s) tabla (s) de base de datos, y el (los) mapeador (es) de datos del modelo a la (s) tabla (s). Por lo tanto, la instancia del modelo se puede asignar a las tablas de múltiples maneras, p. para admitir esquemas heredados. No crea que SQLAlchemy proporciona mapeadores a almacenamiento que no sea SQL.

Answer 1

Es posible que desee tener un buen vistazo a James Dennis' DictShield project

"DictShield es un sistema de modelado de base de datos agnóstica. Proporciona una forma de modelar, validar y formar de nuevo los datos con facilidad. Todo ello sin necesidad de ninguna base de datos concreta "

Answer 2

Fuera de mi cabeza:

defino dos dominios ejemplo, User y Animal, una clase de almacenamiento de base Store y dos clases de almacenamiento especializados UserStore y AnimalStore. El uso del administrador de contexto cierra la conexión db (para simplificar uso sqlite en este ejemplo):

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)