目標(biāo)：

使用flask框架+pymongo練習(xí)mysql

需求：

1. 添加人員

• 姓名必填
• 郵箱必填
• 姓名不能重復(fù)
• 郵箱不能重復(fù)

2. 使用post請求，form-data請求格式

?tip:flask有封裝好的操作數(shù)據(jù)庫的flask-sqlalchemy，目標(biāo)還是以mysql為主。
我在網(wǎng)上找了一個簡單封裝pymysql的例子做了一些改動：

1. 使用configparser庫把配置文件配置在文件中
2. 將例子中ExecNonQuery方法中的execute方法改成了批量的executemany方法

https://www.jb51.net/article/178054.htm

??我們先看看pymysql封裝的類
db_mysql:

import pymysql
import configparser


class MYSQL:

    def __init__(self):
        config = configparser.ConfigParser()
        filename = 'config.ini'
        config.read(filename, encoding='utf-8')
        self.host = config.get('data','host')
        self.user = config.get('data','user')
        self.pwd = config.get('data','pwd')
        self.db = config.get('data','db')

    def __GetConnect(self):
        if not self.db:
            raise (NameError,"未設(shè)置數(shù)據(jù)庫")
        self.conn = pymysql.connect(host= self.host,user = self.user,password = self.pwd,database =self.db,charset = "utf8")
        cur = self.conn.cursor()
        if not cur:
            raise (NameError, "連接數(shù)據(jù)庫失敗")
        else:
            return cur

    def ExecQuery(self, sql):
        """
        執(zhí)行查詢語句
        返回的是一個包含tuple的list，list的元素是記錄行，tuple的元素是每行記錄的字段

        調(diào)用示例：
            ms = MYSQL(host="localhost",user="sa",pwd="123456",db="PythonWeiboStatistics")
            resList = ms.ExecQuery("SELECT id,NickName FROM WeiBoUser")
            for (id,NickName) in resList:
              print str(id),NickName
        """
        cur = self.__GetConnect()
        cur.execute(sql)
        resList = cur.fetchall()

        # 查詢完畢后必須關(guān)閉連接
        self.conn.close()
        return resList

    def ExecNonQuery(self, sql,values):
        """
        執(zhí)行非查詢語句

        調(diào)用示例：
          cur = self.__GetConnect()
          cur.execute(sql)
          self.conn.commit()
          self.conn.close()
        """
        cur = self.__GetConnect()
        #傳入要用list
        cur.executemany(sql,values)
        self.conn.commit()
        self.conn.close()

if __name__ == '__main__':
    m=MYSQL()
    mysql = "insert into staff""(staff_name,staff_dept,staff_age,staff_email,staff_sex)values(%s,%s,%s,%s,%s)"
    values =[('小王', '測試部', '15', 'ww.', '1')]
    m.ExecNonQuery(mysql,values)

??看一下配置文件
config.ini

[data]
host=*****
user=***
pwd=****
db=****

??添加人員接口
py_flask

from flask import Flask
import db_mysql
from flask import request
import json


app = Flask(__name__)


@app.route('/add_staff', methods=["POST","GET"])
def add_staff():
    data = request.form
    staff=db_mysql.MYSQL()
    mysql = "insert into staff(staff_name,staff_dept,staff_age,staff_email,staff_sex)values(%s,%s,%s,%s,%s)"
    staff_name = data['staff_name']
    staff_dept = data['staff_dept']
    staff_age = data['staff_age']
    staff_email = data['staff_email']
    print(staff_name)
    print(staff_email)
    if staff_name is '':
        return {'code': '1001', 'result': '姓名不能為空'}
    if staff_email is '':
        return {'code': '1001', 'result': 'email不能為空'}
    staff_sex = data['staff_sex']
    #查詢姓名重復(fù)
    mes=staff.ExecQuery(sql= "select*from staff where staff_name = '"+staff_name+"'" )
    #郵箱重復(fù)
    mes1=staff.ExecQuery(sql= "select*from staff where staff_email = '"+staff_email+"'" )
    values = [(staff_name, staff_dept, staff_age, staff_email, staff_sex)]
    if mes is not ():
        return {'code': '1002', 'result': '人員重復(fù)'}
    if mes1 is not():
        return {'code': '1003', 'result': 'email重復(fù)'}
    else:
        staff.ExecNonQuery(sql=mysql, values=values)
    return {'code': '0', 'result': 'succsed'}

if __name__ == '__main__':

    app.run (host='0.0.0.0', port=5001, debug=True)

下面我們用postman試試

image.png

??下一篇為了更方便體驗我們可以用docker+dockerfile快速構(gòu)建一個環(huán)境。實現(xiàn)快速搭建開發(fā)環(huán)境。