目标：

使用flask框架+pymongo练习mysql

需求：

1. 添加人员

• 姓名必填
• 邮箱必填
• 姓名不能重复
• 邮箱不能重复

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

❤tip:flask有封装好的操作数据库的flask-sqlalchemy，目标还是以mysql为主。
我在网上找了一个简单封装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,"未设置数据库")
        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, "连接数据库失败")
        else:
            return cur

    def ExecQuery(self, sql):
        """
        执行查询语句
        返回的是一个包含tuple的list，list的元素是记录行，tuple的元素是每行记录的字段

        调用示例：
            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()

        # 查询完毕后必须关闭连接
        self.conn.close()
        return resList

    def ExecNonQuery(self, sql,values):
        """
        执行非查询语句

        调用示例：
          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']
    #查询姓名重复
    mes=staff.ExecQuery(sql= "select*from staff where staff_name = '"+staff_name+"'" )
    #邮箱重复
    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': '人员重复'}
    if mes1 is not():
        return {'code': '1003', 'result': 'email重复'}
    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试试

🐷下一篇为了更方便体验我们可以用docker+dockerfile快速构建一个环境。实现快速搭建开发环境。