Toy Repository

base_repository.py

from operator import attrgetter
from hashlib import md5 # because it's fast

class BaseRepository(object):
    
    def find(self, pk):
        raise NotImplementedError()
    
    def find_by(self, **keys):
        raise NotImplementedError()
    
    def filter(self, *filters):
        raise NotImplementedError()
    
    def persist(self, entity):
        raise NotImplementedError()
    
    def delete(self, entity):
        raise NotImplementedError()
    
    def create(self, **kw):
        raise NotImplementedError()
        
class InMemoryRepository(BaseRepository):
    def __init__(self, entity_class):
        self.entity_class = entity_class
        self.entities = {}
    
    def find(self, pk):
        return self.entities.get(pk, None)
    
    def find_by(self, **keys):
        matches = [(attrgetter(k), v) for k,v in keys.items()]
        return [entity for entity in self.entities.values() if all(g(entity) == v for g, v in matches)]
    
    def filter(self, *filters):
        entities = self.entities.values()
        for filt in filters:
            entities = filter(filt, entities)
        return list(entities)
    
    def persist(self, entity):
        self.entities[str(entity)] = entity
    
    def delete(self, entity):
        self.entities.pop(str(entity))
    
    def create(self, **kw):
        return self.entity_class(**kw)
    
    def update(self, pk, entity):
        if pk not in self.entities:
            raise Exception("Entity does not exist yet!")
        self.entities[pk] = entity
        
    def all(self):
        return list(self.entities.values())
        
class User(object):
    def __init__(self, username, email, password):
        self.username = username
        self.email = email
        self.password = md5(password.encode()).hexdigest()
    
    def __str__(self):
        return '<User {0}>'.format(self.username)
    
    __repr__ = __str__
    
UserRepository = InMemoryRepository(User)

u = UserRepository.create(username='justanr', email='[email protected]', password='awesome')
UserRepository.persist(u)

UserRepository.find_by(username='justanr') # [<User justanr>]
UserRepository.find('<User justanr>') # [<User justanr>]
UserRepository.filter(lambda u: 'service' in u.email) # [<User justanr>]

u.email = '[email protected]'
UserRepository.update('<User justanr>', u)
UserRepository.find_by(username='justanr')[0].email # [email protected]

u2 = UserRepository.create(username='fred', email='[email protected]', password='fred') # doesn't persist it 

UserRepository.delete(u)
UserRepository.all() # []

fsqla_repo.py

from myapp.domain.repository import BaseRepository

class User(db.Model):
    id = Column(db.Integer, primary_key=True)
    username = Column(db.Unicode(32), unique=True)
    email = Column(db.UnicodeText, unique=True)
    _password = Column(db.String(32))
    
    password = property(fset=lambda s, v: setattr(s, md5(v).hexdigest()))
    
class FSQLAlchemyRepository(BaseRepository):
    def __init__(self, model, session):
        self.model = model
        self.session = session
    
    def persist(self, entity):
        self.session.add(entity)
        self.session.commit()
    
    def delete(self, entity):
        self.session.delete(entity)
        self.session.commit()
    
    def find(self, pk):
        return self.model.query.get(pk)
    
    def find_by(self, **keys):
        return self.model.query.filter_by(**keys).all()
    
    def filter(self, *filts):
        return self.model.query.filter(*filts).all()
    
    def create(self, **kw):
        return self.model(**kw)