SQLAlchemy+Flask-RESTful使用(一)

前言

開新坑啦.最近打算自己開一個資源聚合網站.就用Flask.

當然也使用了 Flask-RESTful和SQLAlchemy啦

寫的過程中遇到過很多坑/覺得比較有意義的就寫在這里.

變更記錄

# 19.3.15 增加 SQLAlchemy查詢數據里中文亂碼的問題

# 19.2.18 增加 Flask-RESTful中序列化組件的使用方法

# 19.4.11 增加 序列化組件寫在def中的書寫方式

正文

前幾章也講過了基本的結合,這里主要講講常見問題 or 使用技巧

中文亂碼(問號)

最先遇到的問題是亂碼問題,數據庫里的數據是這樣的

經過flask傳回后是這樣的

解決方法是檢查連接數據庫時是否指定了數據庫數據格式

指定數據庫格式寫在訪問連接后類似GET請求傳參

# 用戶名:密碼@訪問地址:端口/數據庫?編碼
engine = create_engine('mysql+mysqldb://root:***@***:***/website?charset=utf8mb4')

序列化

說到ORM,就不得不說序列化.因為我們在使用ORM的時候,獲取的obj是所有字段的集合.那么問題來了,如果 用戶表 中包含了用戶的密碼字段,我們將所有字段全部返回明顯是不合適的/正確做法是視開發邏輯來選擇傳遞給前端哪些字段

序列化的方法有很多.適合自己的才是墜吼的.本人一直討厭捧一踩一的人.

一種方法是依次取再放入字典中然后return字典(序列化),這樣可以保證我們每次都重新指定了返回的字段.但是這樣對后期的維護多有不便,比如如果POST和GET用戶要求都返回一樣的字段.是不是要在GET和POST里寫上兩段同樣的代碼?如果需求改變是不是要同時改變兩段代碼?

另一種方法是利用Flask-RESTful(以下省略為'官方')自帶的序列化組件(EN官方文檔)

https://flask-restful.readthedocs.io/en/latest/fields.html

如果我們使用 官方 帶的序列化時,我們需要寫一個序列化函數.序列化函數可以有多個參數,例如

    resource_fields = {
        # 設置序列化,k為序列化后的字段名/v為序列化的類型(如果重命名需要設置attribute/不寫attribute代表與k同名的字段)
        'id': fields.Integer, # integer代表數字
        'name': fields.String,  # String代表str
        'money': fields.Integer(default=0),  # 設置如果money無值默認為0
        'status': fields.Integer,
        'lv': fields.Integer(attribute='fk_vip_on_vip_lv'),  # 設置該字段重命名為lv
        'info': fields.String(attribute='VipInfo.info')  # 該attr代表跨表到VipInfo的info字段(代表attr可以寫表達式來跨表查詢),注意這里的VipInfo是model中uselist所在行的字段名
    }

編寫完序列化我們怎樣將obj與組件結合呢?文檔給出兩個方法

方法1:裝飾器

class Vip(Resource):
    # VIP信息(單)
    resource_fields = {
        # 設置序列化,k為序列化后的字段名/v為序列化的類型(如果重命名需要設置attribute/不寫attribute代表與k同名的字段)
        'id': fields.Integer, # integer代表數字
        'name': fields.String,  # String代表str
        'money': fields.Integer(default=0),  # 設置如果money無值默認為0
        'status': fields.Integer,
        'lv': fields.Integer(attribute='fk_vip_on_vip_lv'),  # 設置該字段重命名為lv
        'info': fields.String(attribute='VipInfo.info')  # 該attr代表跨表到VipInfo的info字段(代表attr可以寫表達式來跨表查詢),注意這里的VipInfo是model中uselist所在行的字段名

  } @marshal_with(resource_fields) # 將序列化組件結合/如obj為空也會返回, envelope指定返回的json里的序列化字典的名字(如不填則直接在最大的json中) # 寫裝飾器會導致return任意值都經過渲染/如規避使用一下方式 def get(self, vip_id): # 獲取vip基本信息 from app.website.models import DBSession, Vip, VipInfo # 必須在函數內引入 session = DBSession() obj = session.query(Vip).join(VipInfo).filter(Vip.id==vip_id).first() return obj

使用裝飾器會將所有請求在傳回時強制序列化,這也帶來了一個問題點,如果正常訪問,一切看起來如此正常

但是當我們訪問一個不存在的id時

這樣的接口邏輯就可能讓前后端對接出現問題.並且增加前端判斷的工作量.大家都是打工的,何必苦苦為難呢?

如果我們先判斷是否為空然后再返回呢?我們修改一下代碼

class Vip(Resource):
    # VIP信息(單)
    resource_fields = {
        # 設置序列化,k為序列化后的字段名/v為序列化的類型(如果重命名需要設置attribute/不寫attribute代表與k同名的字段)
        'id': fields.Integer, # integer代表數字
        'name': fields.String,  # String代表str
        'money': fields.Integer(default=0),  # 設置如果money無值默認為0
        'status': fields.Integer,
        'lv': fields.Integer(attribute='fk_vip_on_vip_lv'),  # 設置該字段重命名為lv
        'info': fields.String(attribute='VipInfo.info')  # 該attr代表跨表到VipInfo的info字段(代表attr可以寫表達式來跨表查詢),注意這里的VipInfo是model中uselist所在行的字段名

  } @marshal_with(resource_fields) # 將序列化組件結合/如obj為空也會返回, envelope指定返回的json里的序列化字典的名字(如不填則直接在最大的json中) # 寫裝飾器會導致return任意值都經過渲染/如規避使用一下方式 def get(self, vip_id): # 獲取vip基本信息 from app.website.models import DBSession, Vip, VipInfo # 必須在函數內引入 session = DBSession() obj = session.query(Vip).join(VipInfo).filter(Vip.id==vip_id).first() if obj: return obj else: return None

結果:

為什么我直接寫了無obj返回 None 還是不起作用呢?因為使用裝飾器的方式會攔截所有返回強制序列化,所以找不到序列化字段所以都是默認值.

如果我們想更為靈活的,比如obj存在時再使用序列化,為空時傳送我們定制的錯誤信息就需要使用方法2

方法2:手動使用序列化組件

Flask-RESTful官方也提供了自定義的方式,這種方式更為靈活.

class Vip(Resource):
    # VIP信息(單)
    vip_resource_fields = {
        # 設置序列化,k為序列化后的字段名/v為序列化的類型(如果重命名需要設置attribute/不寫attribute代表與k同名的字段)
        'id': fields.Integer, # integer代表數字
        'name': fields.String,  # String代表str
        'money': fields.Integer(default=0),  # 設置如果money無值默認為0
        'status': fields.Integer,
        'lv': fields.Integer(attribute='fk_vip_on_vip_lv'),  # 設置該字段重命名為lv
        'info': fields.String(attribute='VipInfo.info')  # 該attr代表跨表到VipInfo的info字段(代表attr可以寫表達式來跨表查詢),注意這里的VipInfo是model中uselist所在行的字段名

  } # @marshal_with(resource_fields) # 將序列化組件結合/如obj為空也會返回, envelope指定返回的json里的序列化字典的名字(如不填則直接在最大的json中) # 寫裝飾器會導致return任意值都經過渲染/如規避使用一下方式 def get(self, vip_id, resource_fields=vip_resource_fields): # 必須映引入resource # 獲取vip基本信息 from app.website.models import DBSession, Vip, VipInfo # 必須在函數內引入 session = DBSession() obj = session.query(Vip).join(VipInfo).filter(Vip.id==vip_id).first() if obj: dic = marshal(obj, resource_fields) # return時再指定序列化,不指示則不序列化 return info_tool.get_success_dic(dic) else: return info_tool.get_error_dic(1001, 'not user') return obj

# 我們也可以將圖上的 vip_resource_fields 卸載def中(比如寫在 get 中),這樣我們就不用在get的參數中傳 resource_fields ,其他不變即可

# info_tool模塊是本人寫的自用模塊,功能有將 含有各種無法正常轉json的字典轉為正常的字典/定制封裝返回值 等等.

# 此處的 get_success_dic 功能是在外層包一個dic,有一個code值作為前后端交流,信息在data里,例如

# get_error_success 同上

info_tool模塊如下(19.3.18貼出,后續優化恕不通知)

# -*- coding=utf-8 -*-
# 將pymysql得到的dic轉換為可以正常轉json的dic
import time, datetime, decimal


def kv_to_safe(k, v):
    # 判斷v的數據類型,轉換成可json的類型
    if type(v) == datetime.datetime:
        # 如果是datetime類型則轉換為時間戳
        v = time.mktime(v.timetuple())
    if type(v) == decimal.Decimal:
        # 如果是decimal類型則轉換為flaot
        v = float(v)
    if type(v) == bytes:
        # bytes類型轉str(utf8)
        v = str(v, encoding='utf8')
    return k, v


def dict_to_safe(inf, code):
    # dict轉可轉json的dict
    if inf:
        if type(inf) == dict:
            info = {}
            for k,v in inf.items():
                k, v = kv_to_safe(k, v)
                info[k] = v
        elif type(inf) == list:
            info = []
            for i in inf:
                dic = {}
                for k,v in i.items():
                    k, v = kv_to_safe(k, v)
                    dic [k] = v
                info.append(dic)
    else:
        info = None
    dic = {
        'code': code,
        'data': info
    }
    return dic


def oneobj_to_safe(model):
    # 單一對象轉dic
    dic = {}
    if model == None:
        return None
    for col in model._sa_class_manager.mapper.mapped_table.columns:
        dic[col.name] = getattr(model, col.name)
    return dic


def allobj_to_safe(model_list):
    # 多個對象轉dic
    dic_list = []
    if model_list == None:
        return None
    for model in model_list:
        dic_list.append(oneobj_to_safe(model))
    return dic_list


def get_error_dic(code, message):
    # 生成錯誤json
    dic = {
        'code': code,
        'data': {
            'message': message
        }
    }
    return dic


def get_success_dic(dic):
    # 生成正確json,code默認200
    dic = {
        'code': 200,
        'data': dic
    }
    return dic