Flask + peewee app skeleton

readme.md

Folder structure

app
  +- main.py
  +- compnents.py
  +- todo
        +- __init__.py
manage.py

• Each module (component) has to provide an init function:

def init(app, api, model_target):
    components.register_controllers(api, components.Controller.__subclasses__())
    components.register_models(model_target, components.BaseModel.__subclasses__())

Some issues:

• Model registration / creation is a bit messy; it just aggregates up a models[] array
• Management for models also loads and registers api endpoints as well, since it's totally unnecesarry

components.py

import logging
import inspect
import os
import sys
import inspect

import logging
import json

from datetime import datetime

from peewee import *
from playhouse.shortcuts import *
from playhouse.pool import *

from flask_mongoengine import MongoEngine

from flask_restful import Resource


# fix import paths for internal imports
cmd_folder = os.path.realpath(os.path.abspath(
    os.path.split(inspect.getfile(inspect.currentframe()))[0]))
if cmd_folder not in sys.path:
    sys.path.insert(0, cmd_folder)


# -- Controller

BASE_PATH = "/api"


class Controller(Resource):
    path = ""


class MyJsonEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, datetime):
            return int(obj.strftime('%s'))
        return json.JSONEncoder.default(self, obj)

# --- Database


_DB = Proxy()


class BaseModel(Model):
    class Meta:
        database = _DB

    created = TimestampField(null=False, default=None)
    edited = TimestampField(null=True, default=None, index=True)
    is_deleted = BooleanField(null=False, default=False)
    # is_archived = BooleanField(null=False, default=False)

    def __init__(self):
        super()
        self.modified()

    @classmethod
    def get(cls, id):
        sq = cls.select().where(cls.id == id, cls.is_deleted == False)
        if sq.count():
            return sq.get()
        return None

    @classmethod
    def getAll(cls,):
        sq = cls.select().where(cls.is_deleted == False)
        return sq

    def modified(self):
        if not self.created:
            self.created = datetime.now()
        self.edited = datetime.now()

    def delete(self):
        self.modified()
        self.is_deleted = True
        self.save()


# --- Register class tools


logger = logging.getLogger('peewee')
logger.setLevel(logging.WARN)
logger.addHandler(logging.StreamHandler())


def register_controllers(api, controllers):
    for clazz in controllers:
        path = BASE_PATH + clazz.path
        logging.info("Register endpoint {} {}".format(path, clazz.__name__))
        api.add_resource(clazz, path)
    pass


def register_models(target, models):
    for model in models:
        target.append(model)
    pass

def registered_models():
    return []

def database_init(app):
    logging.debug("ConnectDB: " + app.config["DATABASE"])
    if app.config["DATABASE"] == "postgresql":
        database = PooledPostgresqlExtDatabase(
            app.config["DATABASE_NAME"], max_connections=16, stale_timeout=300, **app.config["DATABASE_AUTH"])
    elif app.config["DATABASE"] == "sqlite":
        database = SqliteDatabase(app.config["DATABASE_PATH"], pragmas={
                                  'journal_mode': 'wal', 'cache_size': -1024 * 64})
    else:
        raise RuntimeError("No database set or invalid")

    _DB.initialize(database)
    pass


def database_connect(testing=False):
    _DB.connect()


def database_create_tables(tables):
    _DB.create_tables(tables, safe=True)

main.py

# coding=utf-8
import logging
import os
import sys
import inspect

logging.basicConfig(
    format='%(asctime)s %(levelname)-7s %(module)s.%(funcName)s - %(message)s')
logging.getLogger().setLevel(logging.DEBUG)
logging.disable(logging.NOTSET)
logging.info('Loading %s, app version = %s', __name__,
             os.getenv('CURRENT_VERSION_ID'))


# fix import paths for internal imports
cmd_folder = os.path.realpath(os.path.abspath(
    os.path.split(inspect.getfile(inspect.currentframe()))[0]))
if cmd_folder not in sys.path:
    sys.path.insert(0, cmd_folder)


PRODUCTION = __name__ != "__main__"
DEBUG = not PRODUCTION


from components import MyJsonEncoder


class MyConfig(object):
    RESTFUL_JSON = {'cls': MyJsonEncoder}

    @staticmethod
    def init_app(app):
        import settings
        app.config.from_object(settings)
        config = "settings.production" if PRODUCTION else "settings.local"
        import importlib
        try:
            cfg = importlib.import_module(config)
            logging.debug("Loaded %s" % config)
            app.config.from_object(cfg)
        except ImportError:
            logging.warning("Local settings module not found: %s", config)
        # app.config['RESTFUL_JSON']['cls'] = app.json_encoder = MyJsonEncoder
        # app.secret_key = app.con


from flask import Flask
from flask_restful import Resource, Api


app = Flask(__name__)
app.config.from_object(MyConfig)
MyConfig.init_app(app)
api = Api(app)


# ---

import components
import tasks

components.database_init(app)

models = []
tasks.init(app, api, models)


# --- start dev server

if app.debug and __name__ != '__main__':
    from werkzeug.debug import DebuggedApplication
    app.wsgi_app = DebuggedApplication(app.wsgi_app, True)

if __name__ == '__main__':
    components.database_connect()

    logging.debug("PRODUCTION: %s" % PRODUCTION)
    logging.debug("app.debug: %s" % app.debug)
    logging.debug("app.testing: %s" % app.testing)

    app.run(host="0.0.0.0", port=5000)

manage.py

# coding=utf-8
import logging

from app.main import app
from flask_script import Manager, Command

class InitDatabase(Command):
    """Initialize database"""
    def run(self):
        from app import components
        from app.main import models

        components.database_init(app)
        components.database_connect()
        components.database_create_tables(models)
        pass
    pass

manager = Manager(app)
manager.add_command('initdb', InitDatabase())

# from flask_security import script
# manager.add_command('create-user', script.CreateUserCommand())
# manager.add_command('create-role', script.CreateRoleCommand())
# manager.add_command('add-role', script.AddRoleCommand())
# manager.add_command('remove-role', script.RemoveRoleCommand())
# manager.add_command('activate-user', script.ActivateUserCommand())
# manager.add_command('deactivate-user', script.DeactivateUserCommand())


if __name__ == "__main__":
    manager.run()

requirements.txt

Flask>=0.10.1
flask-restful>=0.3.6
flask_script
peewee