天天看點

Flask後端實踐 連載六 基于Flask與SQLAlchemy的單表接口基于Flask與SQLAlchemy的單表接口

基于Flask與SQLAlchemy的單表接口

tips:

  • 本文主要介紹基于Flask與SQLAlchemy的單表接口
  • 本文基于python3編寫
  • 本文适合有一定Flask項目的朋友閱讀
  • 代碼倉庫

項目場景

一日,項目經理A找到我,先是表揚最近項目重構的不錯,然後,提出一個單表接口想法。經過和他的仔細探讨,單表接口主要是實作一個資料表的增删改查,不會對其他的資料表造成幹擾。

技術細節

  • 實作http請求分發處理
  • 實作請求參數識别和處理
  • 實作資料庫操作

實作邏輯

Flask後端實踐 連載六 基于Flask與SQLAlchemy的單表接口基于Flask與SQLAlchemy的單表接口

Flask MethodView

  • Flask提供MethodView對每個 HTTP 方法執行不同的函數,即隻需要修改不同函數處理方法。詳見官方文檔
  • 官方例子
  1. 定義接口方法
    RL HTTP 方法 描述
    /users/ GET 獲得全部使用者的清單
    /users/ POST 建立一個新使用者
    /users/< id > GET 顯示某個使用者
    /users/< id > PUT 更新某個使用者
    /users/< id > DELETE 删除某個使用者
  2. 定義不同方法的處理邏輯
    from flask.views import MethodView
    
    class UserAPI(MethodView):
    
        def get(self, user_id):
            if user_id is None:
                # return a list of users
                pass
            else:
                # expose a single user
                pass
    
        def post(self):
            # create a new user
            pass
    
        def delete(self, user_id):
            # delete a single user
            pass
    
        def put(self, user_id):
            # update a single user
            pass
               
  3. 注冊到路由
    user_view = UserAPI.as_view('user_api')
    app.add_url_rule('/users/', defaults={'user_id': None},
                     view_func=user_view, methods=['GET',])
    app.add_url_rule('/users/', view_func=user_view, methods=['POST',])
    app.add_url_rule('/users/<int:user_id>', view_func=user_view,
                     methods=['GET', 'PUT', 'DELETE'])
               

編寫字段解析類

  1. 參數定義

    用法 operator_modelFiled ,即 下劃線 前部分為操作方法,後部分為model的鍵。是以所有的鍵都應與資料庫表字段對應。

    符号 含義 用法
    gt 大于 gt_id=1
    ge 大于等于 ge_id=1
    lt 小于 lt_id=1
    le 小于等于 le_id=1
    ne 不等于 ne_id=1
    eq 等于 eq_id=1
    ic 包含 ic_id=1
    ni 不包含 ni_id=1
    in 查詢多個相同字段的值 in_id=1|2
    by 排序(0:正序,1:倒序) by_id=1
  2. 實作代碼(base.py)
    class BaseParse(object):
        """
        識别要查詢的字段
        """
        __model__ = None
        __request__ = request
        by = frozenset(['by'])
        query = frozenset(['gt', 'ge', 'lt', 'le', 'ne', 'eq', 'ic', 'ni', 'in'])
    
        def __init__(self):
            self._operator_funcs = {
                'gt': self.__gt_model,
                'ge': self.__ge_model,
                'lt': self.__lt_model,
                'le': self.__le_model,
                'ne': self.__ne_model,
                'eq': self.__eq_model,
                'ic': self.__ic_model,
                'ni': self.__ni_model,
                'by': self.__by_model,
                'in': self.__in_model,
            }
    
        def _parse_page_size(self):
        	"""
        	擷取頁碼和擷取每頁資料量
        	:return: page 頁碼
                 page_size 每頁資料量
            """
        	default_page = current_app.config['DEFAULT_PAGE_INDEX']
        	default_size = current_app.config['DEFAULT_PAGE_SIZE']
            page = self.__request__.args.get("page",default_page)
            page_size = self.__request__.args.get("size",default_size )
            page = int(page) - 1
            page_size = int(page_size)
            return page, page_size
    
        def _parse_query_field(self):
            """
            解析查詢字段
            :return: query_field 查詢字段
                     by_field 排序字段
            """
            args = self.__request__.args
            query_field = list()
            by_field = list()
            for query_key, query_value in args.items():
                key_split = query_key.split('_', 1)
                if len(key_split)!= 2:
                	continue
                operator, key = key_split
                if not self._check_key(key=key):
                	continue
                if operator in self.query:
                	data = self._operator_funcs[operator](key=key, value=query_value)
                	query_field.append(data)
                elif operator in self.by:
                	data = self._operator_funcs[operator](key=key, value=query_value)
                    by_field.append(data)
            return query_field, by_field
    
        def _parse_create_field(self):
            """
            檢查字段是否為model的字段,并過濾無關字段.
            1.list(dict) => list(dict)
            2. dict => list(dict)
            :return:
            """
            obj = self.__request__.get_json(force=True)
            if isinstance(obj, list):
                create_field = list()
                for item in obj:
                    if isinstance(item, dict):
                        base_dict = self._parse_field(obj=item)
                        create_field.append(base_dict)
                return create_field
            elif isinstance(obj, dict):
                return [self._parse_field(obj=obj)]
            else:
                return list()
    
        def _parse_field(self, obj=None):
            """
            檢查字段模型中是否有,并删除主鍵值
            :param obj:
            :return:
            """
            obj = obj if obj is not None else self.__request__.get_json(force=True)
            field = dict()
            # 擷取model主鍵字段
            primary_key = map(lambda x: x.name, inspect(self.__model__).primary_key)
    
            for key, value in obj.items():
                if key in primary_key:
                    continue
                if self._check_key(key):
                    field[key] = value
            return field
    
        def _check_key(self, key):
            """
            檢查model是否存在key
            :param key:
            :return:
            """
            if hasattr(self.__model__, key):
                return True
            else:
                return False
    
        def __gt_model(self, key, value):
            """
            大于
            :param key:
            :param value:
            :return:
            """
            return getattr(self.__model__, key) > value
    
        def __ge_model(self, key, value):
            """
            大于等于
            :param key:
            :param value:
            :return:
            """
            return getattr(self.__model__, key) > value
    
        def __lt_model(self, key, value):
            """
            小于
            :param key:
            :param value:
            :return:
            """
            return getattr(self.__model__, key) < value
    
        def __le_model(self, key, value):
            """
            小于等于
            :param key:
            :param value:
            :return:
            """
            return getattr(self.__model__, key) <= value
    
        def __eq_model(self, key, value):
            """
            等于
            :param key:
            :param value:
            :return:
            """
            return getattr(self.__model__, key) == value
    
        def __ne_model(self, key, value):
            """
            不等于
            :param key:
            :param value:
            :return:
            """
            return getattr(self.__model__, key) != value
    
        def __ic_model(self, key, value):
            """
            包含
            :param key:
            :param value:
            :return:
            """
            return getattr(self.__model__, key).like('%{}%'.format(value))
    
        def __ni_model(self, key, value):
            """
            不包含
            :param key:
            :param value:
            :return:
            """
            return getattr(self.__model__, key).notlike('%{}%'.format(value))
    
        def __by_model(self, key, value):
            """
            :param key:
            :param value: 0:正序,1:倒序
            :return:
            """
            try:
                value = int(value)
            except ValueError as e:
                logger.error(e)
                return getattr(self.__model__, key).asc()
            else:
                if value == 1:
                    return getattr(self.__model__, key).asc()
                elif value == 0:
                    return getattr(self.__model__, key).desc()
                else:
                    return getattr(self.__model__, key).asc()
    
        def __in_model(self, key, value):
            """
            查詢多個相同字段的值
            :param key:
            :param value:
            :return:
            """
            value = value.split('|')
            return getattr(self.__model__, key).in_(value)
    
               

編寫資料庫查詢類

  1. 定義

    根據項目實際需求,單表接口主要涉及到資料表的增删改查,主要有建立資料、批量建立資料、删除、更新、主鍵查詢、多條件查詢、分頁、排序等資料庫操作方法。

  2. 實作代碼(base.py)
    class BaseQuery(object):
        """
        查詢方法
        """
        __model__ = None
    
        def _find(self, query):
        	"""
        	根據查詢參數擷取内容
        	"""
            return self.__model__.query.filter(*query).all()
    
        def _find_by_page(self, page, size, query, by):
        	"""
        	根據查詢參數,分頁,排序擷取内容
        	"""
            base = self.__model__.query.filter(*query).order_by(*by)
            cnt = base.count()
            data = base.slice(page * size, (page + 1) * size).all()
            return cnt, data
    
        def _get(self, key):
        	"""
        	根據主鍵ID擷取資料
        	"""
            return self.__model__.query.get(key)
    
        def _create(self, args):
        	"""
        	建立新的一條資料或批量建立
        	"""
            for base in args:
                model = self.__model__()
                for k, v in base.items():
                    setattr(model, k, v)
                db.session.add(model)
            try:
                db.session.commit()
                return True
            except Exception as e:
                logger.error(e)
                return False
    
        def _update(self, key, kwargs):
        	"""
        	更新資料
        	"""
            model = self._get(key)
            if model:
                for k, v in kwargs.items():
                    setattr(model, k, v)
                try:
                    db.session.add(model)
                    db.session.commit()
                    return True
                except Exception as e:
                    logger.error(e)
                    return False
            else:
                return False
    
        def _delete(self, key):
        	"""
        	删除資料
        	"""
            model = self._get(key)
            if model:
                try:
                    db.session.delete(model)
                    db.session.commit()
                    return True
                except Exception as e:
                    logger.error(e)
                    return False
            else:
                return False
    
        def parse_data(self, data):
        	"""
        	解析查詢的資料
        	"""
            if data:
                if isinstance(data, (list, tuple)):
                
                    data = list(map(lambda x: {p.key: getattr(x, p.key)
                                               for p in self.__model__.__mapper__.iterate_properties
                                               }, data))
                else:
                    data = {p.key: getattr(data, p.key) 
                    for p in self.__model__.__mapper__.iterate_properties}
            return data
               

編寫Service類

  1. 定義

    前面我們已經知道了Flask實作了不同http請求的分發(MethodView),是以隻需要自定義不同http請求處理邏輯。通過前面以及解決了參數解析或資料庫操作。那麼将BaseParse, BaseQuery, MethodView組合起來,就可以完成單表接口。

  2. 實作代碼(base.py)
    def view_route(f):
        """
        路由設定,統一傳回格式
        :param f:
        :return:
        """
        def decorator(*args, **kwargs):
            rv = f(*args, **kwargs)
            if isinstance(rv, (int, float)):
                res = ResMsg()
                res.update(data=rv)
                return jsonify(res.data)
            elif isinstance(rv, tuple):
                if len(rv) >= 3:
                    return jsonify(rv[0]), rv[1], rv[2]
                else:
                    return jsonify(rv[0]), rv[1]
            elif isinstance(rv, dict):
                return jsonify(rv)
            elif isinstance(rv, bytes):
                rv = rv.decode('utf-8')
                return jsonify(rv)
            else:
                return jsonify(rv)
        return decorator
        
    class Service(BaseParse, BaseQuery, MethodView):
        __model__ = None
        
        # 裝飾器控制資料傳回格式
        decorators = [view_route]
    
        def get(self, key=None):
            """
            擷取清單或單條資料
            :param key:
            :return:
            """
            res = ResMsg()
            if key is not None:
                data = self.parse_data(self._get(key=key))
                if data:
                    res.update(data=data)
                else:
                    res.update(code=ResponseCode.NO_RESOURCE_FOUND)
            else:
                query, by = self._parse_query_field()
                page, size = self._parse_page_size()
                cnt, data = self._find_by_page(page=page, size=size, query=query, by=by)
                data = self.parse_data(data)
                if data:
                    res.update(data=data)
                else:
                    res.update(code=ResponseCode.NO_RESOURCE_FOUND)
                res.add_field(name='total', value=cnt)
                res.add_field(name='page', value=page + 1)
                res.add_field(name='size', value=size)
                res.update(data=data)
            return res.data
    
        def post(self):
            """
            建立資料
            1.單條
            2.多條
            :return:
            """
            res = ResMsg()
            data = self._parse_create_field()
            if data:
                if not self._create(args=data):
                    res.update(code=ResponseCode.FAIL)
            else:
                res.update(code=ResponseCode.INVALID_PARAMETER)
            return res.data
    
        def put(self, key=None):
            """
            更新某個資料
            :return:
            """
            res = ResMsg()
            if key is None:
                res.update(code=ResponseCode.INVALID_PARAMETER)
            else:
                data = self._parse_field()
                if not self._update(key=key, kwargs=data):
                    res.update(code=ResponseCode.FAIL)
            return res.data
    
        def delete(self, key=None):
            """
            删除某個資料
            :return:
            """
            res = ResMsg()
            if key is None:
                res.update(code=ResponseCode.INVALID_PARAMETER)
            elif not self._delete(key=key):
                res.update(code=ResponseCode.FAIL)
            return res.data
               

單表接口使用

  1. 資料庫表(model.py)
    class Article(db.Model):
        """
        文章表
        """
        __tablename__ = 'article'
        id = db.Column(db.Integer, autoincrement=True, primary_key=True)
        title = db.Column(db.String(20), nullable=False)  # 文章标題
        body = db.Column(db.String(255), nullable=False)  # 文章内容
               
  2. 接口服務(service.py)
    from base import Service
    from models import *
    
    class ArticleAPI(Service):
        """
        文章單表接口
        """
        __model__ = Article
        service_name = 'article'
               
  3. 注冊服務(app.py)
    from flask import Flask
    from service import ArticleAPI
    app = Flask(__name__)
    article_view = ArticleAPI.as_view('article_api')
    app.add_url_rule('/article/', defaults={'key': None},
                     view_func=article_view , methods=['GET',])
    app.add_url_rule('/article/', view_func=article_view , methods=['POST',])
    app.add_url_rule('/article/<string:key>', view_func=article_view ,
                     methods=['GET', 'PUT', 'DELETE'])
    if __name__ =='__main__':
    	app.run()
               
  4. 測試
    • 單條建立 POST http://127.0.0.1:5000/article/

      發送資料:

      {"title":"測試1","body":"測試1"}

      傳回:

      {"code":0,"data":null,"lang":"zh_CN","msg":"成功"}

    • 批量建立 POST http://127.0.0.1:5000/article/

      發送資料:

      [{"title":"測試2","body":"測試2"},{"title":"測試3","body":"測試3"}]

      傳回:

      {"code":0,"data":null,"lang":"zh_CN","msg":"成功"}

    • 查詢 GET http://127.0.0.1:5000/article/?eq_title=測試1

      傳回:

      {"code":0,"data":[{"body":"測試1","id":1,"title":"測試1"}],"lang":"zh_CN","msg":"成功", "page":1,"size":10,"total":1}

    • 查詢 GET http://127.0.0.1:5000/article/1

      傳回:

      {"code":0, "data":{"body":"測試1", "id":1,"title":"測試1" },"lang":"zh_CN","msg":"成功"}

    • 更新 PUT http://127.0.0.1:5000/article/1

      發送資料:

      {"title":"測試111","body":"測試111"}

      傳回:

      {"code":0,"data":null,"lang":"zh_CN","msg":"成功"}

    • 删除 DELETE http://127.0.0.1:5000/article/1

      傳回:

      {"code":0,"data":null,"lang":"zh_CN","msg":"成功"}

總結

  • 文中涉及到了字元串操作解析方面、類的定義、SQLAlchemy使用,Flask MethodView使用
  • 通過類繼承的方法,可以快速擴充單表接口,大大的提高了開發效率。如果有其他實作需求,也可以通過重寫的方法完成定制。
  • 下一篇文章将介紹reids在Flask裡面的使用