- 要在python中使用MySQL数据库，首先需要安装pymysql，如果需要使用MySQL连接池，还需要安装dbutils
pip install pymysqlpip install dbutils
- 在项目目录中的utils或config目录中（没有可以新建一个）新建一个db.py文件（文件名可以自己随便起）
# 1. 先引入pymysql
import pymysql
# 2. 引入数据库工具
from dbutils.pooled_db import PooledDB# 3. 数据库配置
db_config = {'host': '192.168.1.119','port': 3306,'user': 'root','passwd': 'root','db': 'refresh_orders','charset': 'utf8'}# 4. 创建连接池
try:mysql_pool = PooledDB(creator=pymysql, # 使用链接数据库的模块maxconnections=6, # 连接池允许的最大连接数，0和None表示不限制连接数mincached=2, # 初始化时，链接池中至少创建的空闲的链接，0表示不创建maxcached=5, # 链接池中最多闲置的链接，0和None不限制maxshared=3, # 链接池中最多共享的链接数量，0和None表示blocking=True, # 连接池中如果没有可用连接后，是否阻塞等待；True表示阻塞，False表示不等待然后报错setsession=[], # 开启会话ping=0, # ping MySQL服务端，检查是否服务可用**db_config # 数据库连接参数)
except Exception as e:print(e)# 5. 定义方法（函数）
def get_conn():# 获取连接return mysql_pool.connection()def fetch_one(sql, args=None):# 获取单条数据conn = get_conn()cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)cursor.execute(sql, args)result = cursor.fetchone()close_conn(conn, cursor)return resultdef fetch_all(sql, args=None):# 获取单条数据conn = get_conn()cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)cursor.execute(sql, args)result = cursor.fetchall()close_conn(conn, cursor)return resultdef create(sql, args=None):# 创建数据conn = get_conn()cursor = conn.cursor()try:cursor.execute(sql, args) # 执行sql语句conn.commit() # 提交return cursor.lastrowid # 返回插入数据的idexcept Exception as e:print(e)finally:close_conn(conn, cursor)"""
关闭连接
"""
def close_conn(conn,cursor):if cursor:cursor.close()if conn:conn.close()
- 在项目中其它地方需要使用的地方先引入db,然后调用相应的函数（方法）
from flask import Blueprint, request,session,jsonify,render_template,redirect
from utils import dbfrom flask import Blueprint, request,session,jsonify,render_template,redirect
from utils import db
from utils import cacheod = Blueprint('order',__name__)ADMIN = 0
USER = 1@od.route('/order/list')
def list():# 从cookie中获取用户信息user = session.get('user_info')data_list = []if user['role'] == ADMIN:# data_list = db.fetch_all("select * from orders")# 连表查询 left joindata_list = db.fetch_all("select o.id, o.url, o.count, o.status, u.real_name from orders o left join users u on o.user_id=u.id")else:# 连表查询 left joindata_list = db.fetch_all("select o.id, o.url, o.count, o.status, u.real_name from orders o left join users u on o.user_id=u.id where o.user_id=%s", [user['id']])status_map = {0: {'text':'待处理', 'class':'info'}, 1: {"text":'处理中...', 'class':'warning'}, 2: {"text":'已完成', 'class':'success'}, 3: {"text":'失败', 'class':'danger'}}return render_template('order_list.html', order_list=data_list, status_map=status_map)@od.route('/order/create', methods=['GET','POST'])
def create():if request.method == 'POST':url = request.form.get('url')count = request.form.get('count')if not url or not count: return render_template('order_create.html', error='数据不完整, 请填写完整的订单信息')user = session.get('user_info')sql = "insert into orders(url,count,status,user_id) values(%s,%s,0,%s)"order_id = db.create(sql, [url, count, user['id']])print(order_id)# todo: 写入redis队列cache.push_task_queue(order_id)return redirect('/order/list')return render_template('order_create.html')
 一快速环境搭建)
