Flask project construction and configuration sharing

Flask project construction and configuration sharing

Author : Tom .Lee, GitHub ID: tomoncle , Web and cloud computing developer, Java, Golang, Python enthusiast.

summary

I haven’t touched the flask framework for a long time. I’ll write something recently, remember, share the introduction of flask with friends, and share the source code, see the bottom of the article

  • Expand flask to support banner, support config.properties configuration file import
  • Modular design, support database migration
  • Encapsulate sqlalchemy database operations
  • Automatically transfer to json
  • Configure the interceptor, automatic analysis of exceptions (web request returns an error page, curl request returns an error json)
  • Expand flask built-in functions and support environment variables
  • Integrated celery framework asynchronous processing
  • Support docker build
  • Flask jinja2 template example
  • swagger api document configuration
  • and so on

Module structure diagram

.

    .
    ├── banner.txt
    ├── bootstrap_app.py
    ├── bootstrap_celery.py
    ├── config.properties
    ├── config.py
    ├── Dockerfile
    ├── examples
    │ ├── extensions_flask_form.py
    │ ├── extensions_flask_SQLAlchemy.py
    │ ├── hello_world.py
    │ ├── index.py
    │ ├── __init__.py
    │ └── rest_api.py
    ├── flaskapp
    │ ├── common
    │ │ ├── error_view.py
    │ │ ├── exceptions.py
    │ │ ├── __init__.py
    │ │ ├── logger.py
    │ │ ├── response.py
    │ │ ├── tools.py
    │ │ └── utils.py
    │ ├── core
    │ │ ├── database.py
    │ │ ├── http_handler.py
    │ │ ├── http_interceptor.py
    │ │ └── __init__.py
    │ ├── extends
    │ │ ├── banner.py
    │ │ ├── functions.py
    │ │ └── __init__.py
    │ ├── __init__.py
    │ ├── models
    │ │ ├── base.py
    │ │ ├── clazz.py
    │ │ ├── __init__.py
    │ │ ├── school.py
    │ │ └── user.py
    │ ├── plugins
    │ │ ├── flask_celery.py
    │ │ └── __init__.py
    │ ├── services
    │ │ ├── base.py
    │ │ ├── __init__.py
    │ │ └── statement.py
    │ └── views
    │ ├── async_handler.py
    │ ├── error_handler.py
    │ ├── index_hander.py
    │ ├── __init__.py
    │ ├── rest_clazz_handler.py
    │ ├── rest_login_handler.py
    │ ├── rest_school_handler.py
    │ └── rest_user_handler.py
    ├── git-user-config.sh
    ├── README.md
    ├── requirements.txt
    ├── static
    │ ├── css
    │ │ └── layout.css
    │ ├── favicon.ico
    │ ├── images
    │ │ └── 0.jpg
    │ └── js
    │ └── app.js
    ├── stop-app.sh
    ├── templates
    │ ├── 404.html
    │ ├── examples
    │ │ ├── extensions_flask_form.html
    │ │ └── extensions_flask_sqlAlchemy.html
    │ ├── index.html
    │ └── layout.html
    └── test
        ├── config.properties
        ├── __init__.py
        ├── plugins
        │ ├── __init__.py
        │ └── test_celery_task.py
        ├── test_banner.py
        ├── test_celery.py
        ├── test_db.py
        ├── test_extend_func.py
        ├── test_lru.py
        ├── test_platform.py
        └── views
            └── test_school.py

Database package

class Database(object):
    """
    database interface
    """


class Transactional(Database):
    def __init__(self, **kwargs):
        """
        Transaction layer
        :param auto_commit: Whether to submit automatically
        """
        self._auto_commit = kwargs.get('auto_commit', True)
        self.model = kwargs.get('model_class')
        if not self.model:
            raise AssertionError('<class {}>: Required parameter model_class is not present.'
                                 .format(self.__class__.__name__))
        self.session = db.session
        # logger.info('init Transactional')

    def auto_commit(self):
        """
        Whether to commit the transaction automatically
        :return:
        """
        if self._auto_commit:
            self.session.commit()

    def _check_type(self, obj):
        if not isinstance(obj, self.model):
            raise AssertionError('obj must be <class {}> type.'
                                 .format(self.model.__class__.__name__))


class Persistence(Transactional):
    def __init__(self, **kwargs):
        super(Persistence, self).__init__(**kwargs)
        # logger.info('init Persistence')


class Modify(Transactional):
    def __init__(self, **kwargs):
        super(Modify, self).__init__(**kwargs)
        # logger.info('init Modify')


class Remove(Transactional):
    def __init__(self, **kwargs):
        super(Remove, self).__init__(**kwargs)
        # logger.info('init Remove')


class Query(Database):
    def __init__(self, **kwargs):
        # logger.info('init Query')
        self.model = kwargs.get('model_class', None)
        if not self.model:
            raise AssertionError('<class {}>: model_class is not found.'
                                 .format(self.__class__.__name__))


class Modify2(Database):
    @classmethod
    def _auto_commit(cls):
        db.session.commit()


class Query2(Database):
    def __init__(self):
        """The entity type needs to be passed in to use the class"""
        # logger.info('init Query2')

banner configuration

def _banner():
    banner_path = os.path.join(
        os.path.dirname(os.path.dirname(
            os.path.dirname(__file__))),'banner.txt')
    if os.path.exists(banner_path) and os.path.isfile(banner_path):
        with open(banner_path) as f:
            for line in f:
                print line.rstrip('\n')
    else:
        print banner_text

Interface browse

$ curl localhost:5000/paths

Error handling

  • Page request:
  • curl request:
$ curl localhost:5000/api/vi/students/err

Cascade query to json

def json(self, lazy=False, ignore=None, deep=1):
    """
    Convert json
    :param lazy: Whether to load lazy
    :param ignore: filter attributes
    :param deep: current depth
    :return:
    """
    ignore_filed = ['query','metadata'] + ignore if isinstance(ignore, list) else ['query','metadata', ignore]

    def _filter_filed(obj):
        return filter(lambda y: all(
            (
                y not in ignore_filed,
                not y.endswith('_'),
                not y.startswith('_'),
                not callable(getattr(obj, y))
            )), dir(obj))

    def _get_ignore_filed(base_obj, obj, _filed_list):
        _ignore_filed = []
        for _filed in _filed_list:
            _filed_obj = getattr(obj, _filed)
            if isinstance(_filed_obj, BaseQuery):
                _primary_entity ='%s'% _filed_obj.attr.target_mapper
                if _primary_entity.split('|')[1] == base_obj.__class__.__name__:
                    _ignore_filed.append(_filed)
            if isinstance(_filed_obj, type(base_obj)):
                _ignore_filed.append(_filed)
        return _ignore_filed

    __relationship__, res, filed_list = None, {}, _filter_filed(self)
    for filed in filed_list:
        filed_type = getattr(self, filed)
        if filed == __relationship__:
            continue
        if isinstance(filed_type, DictModel):
            _ignore = _get_ignore_filed(self, filed_type, _filter_filed(filed_type))
            relationship_model = filed_type.json(ignore=_ignore, deep=deep + 1)
            if not lazy:
                res[filed] = relationship_model
        elif isinstance(filed_type, (int, list)):
            res[filed] = filed_type
        elif isinstance(filed_type, BaseQuery):
            res[filed] = []
            if not lazy:
                for f in filed_type.all():
                    _ignore = _get_ignore_filed(self, f, _filter_filed(f))
                    res[filed].append(f.json(lazy=lazy, ignore=_ignore, deep=deep + 1))
        else:
            try:
                if isinstance(filed_type, unicode):
                    filed_type = filed_type.encode('UTF-8')
                res[filed] ='{}'.format(filed_type)
            except (UnicodeEncodeError, Exception), e:
                logger.error('{class_name}.{filed}: {e}'.format(
                    class_name=self.__class__.__name__, filed=filed, e=e))
                res[filed] = None
    return res

Expand flask start method start

from flaskapp import app

if __name__ == "__main__":
    app.start()
    # app.start(port=5258, debug=False)

Database update migration

$ python manager.py db init
$ python manager.py db migrate

Dockerfile build

$ ./docker-build.sh

Celery asynchronous processing

  • See project test directory test_celery.py
@celery.task()
def async_compute(a, b):
    from time import sleep
    sleep(10)
    return a + b


@cly.route('/compute')
@json_encoder
def task():
    result = async_compute.delay(1, 2)
    print result.wait()
    return'task id: {}'.format(result)

swagger configuration

  • See swaggerforapi.py in the project examples directory
  • Source address: https://github.com/tomoncle/flaskapp
Reference: https://cloud.tencent.com/developer/article/1167557 flask project construction and configuration sharing-Cloud + Community-Tencent Cloud