pyramid services

pyramid_services.py

from pyramid.interfaces import IRequest
from zope.interface.registry import Components

def includeme(config):
    config.add_request_method(find_service)
    config.add_request_method(lambda _: Components(), 'service_cache',
                              reify=True)

    config.add_directive('register_service', register_service)
    config.add_directive('register_service_factory', register_service_factory)

def register_service(config, service, type_or_iface, name=''):
    service = config.maybe_dotted(service)
    def service_factory(request):
        return service
    config.register_service_factory(service_factory, type_or_iface, name=name)

def register_service_factory(config, service_factory, type_or_iface, name=''):
    service_factory = config.maybe_dotted(service_factory)
    type_or_iface = config.maybe_dotted(type_or_iface)
    def register():
        registry = config.registry
        registry.registerAdapter(
            service_factory, (IRequest,), type_or_iface, name=name)
    config.action(('service factory', type_or_iface), register)

def find_service(request, type_or_iface, name=''):
    cache = request.service_cache
    svc = cache.queryUtility(type_or_iface, name=name)
    if svc is None:
        registry = request.registry
        svc = registry.queryAdapter(request, type_or_iface, name=name)
        cache.registerUtility(svc, type_or_iface, name=name)
    return svc

services.py

from zope.interface import Interface

class ILoginService(Interface):
    pass

def includeme(config):
    config.include('pyramid_services')
    
    dbmaker = sessionmaker()
    dbmaker.configure(bind=engine)
    zope.sqlalchemy.register(dbmaker)
    
    config.add_request_method('db', lambda: dbmaker(), reify=True)
    
    def factory(request):
        return LoginService(request.db)
    config.register_service_factory(factory, ILoginService)
    
class LoginService(object):
    def __init__(self, db):
        self.db = db
        
    def find_user_by_name(self, name):
        return self.db.query(model.User).filter_by(name=name).first()

views.py

from myapp import services as S

def my_view(request):
    login_svc = request.find_service(S.ILoginService)
    user = login_svc.find_user_by_name(request.POST['name'])
    if user:
        return user.id