flask中pymysql连接数据库

可以通过以下代码实现pymysql连接数据库：

import pymysql
from flask import Flask

app = Flask(__name__)

@app.route('/')
def index():
# 连接数据库
db = pymysql.connect(host='localhost', port=3306, user='root', password='123456', database='test')
# 创建游标对象
cursor = db.cursor()
# 执行SQL语句
cursor.execute('select * from user')
# 获取查询结果
data = cursor.fetchall()
# 关闭游标和数据库连接
cursor.close()
db.close()
# 返回查询结果
return str(data)

if __name__ == '__main__':
app.run()

以上代码中，首先导入pymysql和Flask模块。在Flask应用中定义一个路由'/test'，该路由返回从数据库中查询到的数据。

连接数据库的代码可以看到，需要指定数据库的主机、端口、用户名、密码和数据库名等参数。获取游标对象后，可以执行SQL语句并获取结果，最后关闭游标和数据库连接。

相关问题

改为使用mysql数据库

要将现有的Flask应用程序从SQLite迁移到MySQL，需要进行以下步骤：

1. **安装MySQL驱动**：首先，确保安装了MySQL的Python驱动程序 `mysql-connector-python` 或者 `pymysql`。可以使用以下命令安装：

   pip install mysql-connector-python

2. **修改数据库连接配置**：将原有的SQLite数据库连接代码更改为MySQL的连接代码。

3. **调整SQL语句**：虽然大多数SQL语句在SQLite和MySQL之间是兼容的，但某些语法可能会有所不同，需要进行相应的调整。

以下是修改后的代码示例：

### 修改后的代码

from flask import Flask, render_template, request, redirect, url_for, flash, session
import mysql.connector
import bcrypt
import logging

app = Flask(__name__)
app.secret_key = 'your_secret_key'

# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# 数据库连接
def connect_db():
    conn = mysql.connector.connect(
        host='localhost',
        user='your_username',
        password='your_password',
        database='pet_weight_management'
    )
    return conn

# 初始化数据库
def init_db():
    conn = connect_db()
    cursor = conn.cursor()
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS users (
            id INT AUTO_INCREMENT PRIMARY KEY,
            username VARCHAR(255) UNIQUE NOT NULL,
            password VARCHAR(255) NOT NULL
        )
    ''')
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS pets (
            id INT AUTO_INCREMENT PRIMARY KEY,
            user_id INT NOT NULL,
            pet_name VARCHAR(255) NOT NULL,
            FOREIGN KEY (user_id) REFERENCES users (id) ON DELETE CASCADE
        )
    ''')
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS pet_weights (
            id INT AUTO_INCREMENT PRIMARY KEY,
            pet_id INT NOT NULL,
            weight FLOAT NOT NULL,
            recorded_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            FOREIGN KEY (pet_id) REFERENCES pets (id) ON DELETE CASCADE
        )
    ''')
    conn.commit()
    conn.close()

@app.route('/')
def index():
    return render_template('index.html')

@app.route('/register', methods=['GET', 'POST'])
def register():
    if request.method == 'POST':
        username = request.form['username']
        password = request.form['password']
        conn = connect_db()
        cursor = conn.cursor()
        hashed_password = bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt())
        try:
            cursor.execute('INSERT INTO users (username, password) VALUES (%s, %s)', (username, hashed_password))
            conn.commit()
            flash("注册成功！", 'success')
            logging.info("用户 %s 注册成功！", username)
            return redirect(url_for('login'))
        except mysql.connector.IntegrityError:
            flash("用户名已存在！", 'warning')
            logging.warning("用户名 %s 已存在！", username)
        finally:
            conn.close()
    return render_template('register.html')

@app.route('/login', methods=['GET', 'POST'])
def login():
    if request.method == 'POST':
        username = request.form['username']
        password = request.form['password']
        conn = connect_db()
        cursor = conn.cursor()
        cursor.execute('SELECT id, password FROM users WHERE username = %s', (username,))
        result = cursor.fetchone()
        if result and bcrypt.checkpw(password.encode('utf-8'), result[1]):
            session['user_id'] = result[0]
            flash("登录成功！", 'success')
            logging.info("用户 %s 登录成功！", username)
            return redirect(url_for('dashboard'))
        else:
            flash("用户名或密码错误！", 'danger')
            logging.warning("用户名或密码错误！")
        conn.close()
    return render_template('login.html')

@app.route('/logout')
def logout():
    session.pop('user_id', None)
    flash("已退出登录。", 'info')
    logging.info("已退出登录。")
    return redirect(url_for('index'))

@app.route('/dashboard')
def dashboard():
    if 'user_id' not in session:
        return redirect(url_for('login'))
    user_id = session['user_id']
    conn = connect_db()
    cursor = conn.cursor()
    cursor.execute('SELECT id, pet_name FROM pets WHERE user_id = %s', (user_id,))
    pets = cursor.fetchall()
    conn.close()
    return render_template('dashboard.html', pets=pets)

@app.route('/add_pet', methods=['GET', 'POST'])
def add_pet():
    if 'user_id' not in session:
        return redirect(url_for('login'))
    if request.method == 'POST':
        user_id = session['user_id']
        pet_name = request.form['pet_name']
        weight = float(request.form['weight'])
        conn = connect_db()
        cursor = conn.cursor()
        cursor.execute('INSERT INTO pets (user_id, pet_name) VALUES (%s, %s)', (user_id, pet_name))
        pet_id = cursor.lastrowid
        cursor.execute('INSERT INTO pet_weights (pet_id, weight) VALUES (%s, %s)', (pet_id, weight))
        conn.commit()
        flash("宠物添加成功，初始体重为 %.2f kg" % weight, 'success')
        logging.info("宠物 %s 添加成功，初始体重为 %.2f kg", pet_name, weight)
        conn.close()
        return redirect(url_for('dashboard'))
    return render_template('add_pet.html')

@app.route('/view_pet/<int:pet_id>')
def view_pet(pet_id):
    if 'user_id' not in session:
        return redirect(url_for('login'))
    conn = connect_db()
    cursor = conn.cursor()
    cursor.execute('SELECT pet_name FROM pets WHERE id = %s', (pet_id,))
    pet = cursor.fetchone()
    if not pet:
        flash("宠物不存在！", 'warning')
        logging.warning("宠物不存在！")
        return redirect(url_for('dashboard'))
    cursor.execute('SELECT weight, recorded_at FROM pet_weights WHERE pet_id = %s ORDER BY recorded_at', (pet_id,))
    weights = cursor.fetchall()
    conn.close()
    return render_template('view_pet.html', pet=pet, weights=weights)

@app.route('/update_pet_weight/<int:pet_id>', methods=['GET', 'POST'])
def update_pet_weight(pet_id):
    if 'user_id' not in session:
        return redirect(url_for('login'))
    conn = connect_db()
    cursor = conn.cursor()
    # 获取当前宠物的所有体重记录
    cursor.execute('SELECT weight, recorded_at FROM pet_weights WHERE pet_id = %s ORDER BY recorded_at DESC', (pet_id,))
    weights = cursor.fetchall()
    if request.method == 'POST':
        weight = float(request.form['weight'])
        cursor.execute('INSERT INTO pet_weights (pet_id, weight) VALUES (%s, %s)', (pet_id, weight))
        conn.commit()
        flash("宠物体重更新成功！", 'success')
        logging.info("宠物体重更新成功！")
        conn.close()
        return redirect(url_for('view_pet', pet_id=pet_id))
    conn.close()
    return render_template('update_pet_weight.html', pet_id=pet_id, weights=weights)

@app.route('/delete_account', methods=['GET', 'POST'])
def delete_account():
    if 'user_id' not in session:
        return redirect(url_for('login'))
    if request.method == 'POST':
        user_id = session['user_id']
        conn = connect_db()
        cursor = conn.cursor()
        try:
            # 删除宠物的体重记录
            cursor.execute('DELETE FROM pet_weights WHERE pet_id IN (SELECT id FROM pets WHERE user_id = %s)', (user_id,))
            # 删除宠物信息
            cursor.execute('DELETE FROM pets WHERE user_id = %s', (user_id,))
            # 删除用户信息
            cursor.execute('DELETE FROM users WHERE id = %s', (user_id,))
            conn.commit()
            session.pop('user_id', None)
            flash("账号已注销。", 'success')
            logging.info("用户 %s 账号已注销。", user_id)
        except Exception as e:
            flash("注销账号失败: %s" % str(e), 'danger')
            logging.error("注销账号失败: %s", str(e))
        finally:
            conn.close()
        return redirect(url_for('index'))
    return render_template('delete_account.html')

if __name__ == '__main__':
    init_db()
    app.run(debug=True)

### 关键点总结
1. **数据库连接**：使用 `mysql.connector` 连接到MySQL数据库。
2. **SQL语句**：调整SQL语句以适应MySQL的语法（例如，使用 `%s` 作为占位符）。
3. **初始化数据库**：创建表时使用MySQL的语法。

通过以上步骤，你可以将现有的Flask应用从SQLite迁移到MySQL。

用python完成一个基于浏览器的试题管理系统，数据库建议使用Mysql。Python web 框架建议使用flask框架。题型包括：单选题、填空题、判断题、问答题、编程题。系统功能： 1. 试题录入和提交 2. 数据增删查改。 3. 数据导出

为了创建一个基于浏览器的试题管理系统，你可以按照以下步骤来构建：

1. **环境配置**:
- 安装必要的工具：首先确保安装了Python（版本3.x）、Flask Web框架以及MySQLdb（用于Python连接MySQL）库。

   pip install Flask mysql-connector-python

2. **项目结构**:
- 创建一个名为`exam_system`的文件夹，里面包含app.py主应用文件，models.py处理数据库模型，views.py编写视图函数，templates目录存放HTML模板，static目录存放静态资源。

3. **数据库模型（models.py）**:
- 使用SQLAlchemy创建一个基础的试题表结构，例如：

   from flask_sqlalchemy import SQLAlchemy

   db = SQLAlchemy()

   class Question(db.Model):
       id = db.Column(db.Integer, primary_key=True)
       type = db.Column(db.String(50))
       content = db.Column(db.Text)
       options = db.Column(db.JSON)  # 对于非选择题类型的选项
       answer = db.Column(db.Text)  # 对于有固定答案的题目

4. **Flask应用（app.py）**:
- 初始化Flask应用并配置数据库连接。

   from flask import Flask, render_template, request, redirect, url_for, flash
   from models import db, Question
   
   app = Flask(__name__)
   app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://username:password@localhost/exam_system'
   db.init_app(app)
   with app.app_context():
       db.create_all()

5. **视图函数（views.py）**:
- 实现主要的路由和功能。例如，添加试题：

   @app.route('/add_question', methods=['GET', 'POST'])
   def add_question():
       if request.method == 'POST':
           new_question = Question(content=request.form['content'], type=request.form['type'])
           # 根据题型添加相应字段
           db.session.add(new_question)
           db.session.commit()
           return redirect(url_for('index'))
       else:
           return render_template('add_question.html')

6. **HTML模板（templates）**:
- 创建对应的HTML模板，比如`add_question.html`，让用户输入试题信息。

7. **其他功能实现**:
- 通过类似的方式实现其他功能，如显示试题列表、编辑/删除试题、用户登录权限管理等。
- 数据导出可以提供CSV下载功能，使用`csv`模块将数据转换成CSV格式供下载。

8. **运行应用**:
- 最后，在命令行中运行`app.run()`启动服务器，并在浏览器上访问`http://localhost:5000`查看是否正常工作。