基于Flask與SQLAlchemy的單表接口

tips:

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

項目場景

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

技術細節

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

實作邏輯

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

Flask MethodView

Flask提供MethodView對每個 HTTP 方法執行不同的函數，即隻需要修改不同函數處理方法。詳見官方文檔
官方例子

定義接口方法

RL	HTTP 方法	描述
/users/	GET	獲得全部使用者的清單
/users/	POST	建立一個新使用者
/users/< id >	GET	顯示某個使用者
/users/< id >	PUT	更新某個使用者
/users/< id >	DELETE	删除某個使用者

定義不同方法的處理邏輯

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

注冊到路由

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'])

編寫字段解析類

參數定義

用法 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

實作代碼(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)

編寫資料庫查詢類

定義

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

實作代碼(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類

定義

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

實作代碼(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

單表接口使用

資料庫表(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)  # 文章内容

接口服務(service.py)

from base import Service
from models import *

class ArticleAPI(Service):
    """
    文章單表接口
    """
    __model__ = Article
    service_name = 'article'

注冊服務(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()

測試
- 單條建立 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裡面的使用

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

基于Flask與SQLAlchemy的單表接口

項目場景

技術細節

實作邏輯

Flask MethodView

編寫字段解析類

編寫資料庫查詢類

編寫Service類

單表接口使用

總結

繼續閱讀

無法解析的外部符号 wmain，該符号在函數 "void cdecl mainCRTStartupHelper(struct HINSTANCE *,unsigned short con......

TestLink導出用例轉換工具(XML2Excel)

YAML簡介和PyYAML安全操作YAML支援的類型YAML的優點：yaml的基本文法python操作

Small tricks

libsvm for python 安裝

學習軟體測試基礎測試第七天

Zeppelin 配置通路 REST APIApache Zeppelin Configuration REST API

【Torch】最簡潔logging使用指南

27. Remove Element(清單)題目代碼

配置網頁内容的通路3案例3：配置網頁内容通路

Cloud Studio初體驗

使用 ctypes 進行 Python 和 C 的混合程式設計

【python】【資料處理】畫多元資料分布圖

【python】netconf協定對接管理裝置

「Python 網絡自動化」NETCONF —— Python 使用 NETCONF 管理配置 H3C 網絡裝置

在python中建立excel并寫入

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

基于Flask與SQLAlchemy的單表接口

項目場景

技術細節

實作邏輯

Flask MethodView

編寫字段解析類

編寫資料庫查詢類

編寫Service類

單表接口使用

總結

繼續閱讀

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