mp_user_service.py

from .app import App
from .db import get_dbsession
from .models import User


@App.method(App.dbsession)
def app_dbsession(app):
    return get_dbsession(app.settings.sqlalchemy.__dict__)

  
class UserService(object):
    def __init__(self, dbsession):
        self.dbsession = dbsession

    def all(self):
        return self.dbsession.query(User).all()

    def by_id(self, id):
        return self.dbsession.query(User)\
            .filter(User.id == id)\
            .first()

    def by_username(self, username):
        return self.dbsession.query(User)\
            .filter(User.username == username)\
            .first()

    def by_email(self, email):
        return self.dbsession.query(User)\
            .filter(User.email == email)\
            .first()

    def add(self, username, email):
        user = User(
            username=username,
            email=email
            )
        self.dbsession.add(user)
        self.dbsession.flush()
        return user

    def delete(self, user):
        self.dbsession.delete(user)


@App.method(App.service, name='user')
def user_service(app, name):
    dbsession = app.dbsession()
    return UserService(dbsession=dbsession)