Python中使用Mysql


Mysql数据库的安装可查看这里✈️

Python中使用Mysql

在Python中使用Mysql数据库需要驱动，有多种驱动可以选择，如pymysql、mysqldb、mysqlclient等。本篇文章主要介绍通过pymysql库来实现。

  1. 安装pymysql
    1
    pip install pymysql
  2. 连接数据库
    1
    2
    3
    4
    5
    6
    7
    8
    9
    import pymysql

    # 连接数据库
    conn = pymysql.connect(
        host='localhost',
        user='root',
        password='123456',
        database='mydb'
    )
  3. 执行查询
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    # 创建游标
    cursor = conn.cursor()

    # 执行查询
    cursor.execute('SELECT * FROM users')

    # 获取结果
    results = cursor.fetchall()

    # 打印结果
    for row in results:
        print(row)

    # 关闭游标
    cursor.close()

    # 关闭连接
    conn.close()
  4. 执行更新
    1
    2
    3
    4
    5
    # 执行更新
    cursor.execute('UPDATE users SET age = %s WHERE id = %s', (25, 1))

    # 提交事务
    conn.commit()
  5. 执行删除
    1
    2
    3
    4
    5
    # 执行删除
    cursor.execute('DELETE FROM users WHERE id = %s', (1,))

    # 提交事务
    conn.commit()
  6. 执行插入
    1
    2
    3
    4
    5
    # 执行插入
    cursor.execute('INSERT INTO users (name, age) VALUES (%s, %s)', ('John Doe', 30))

    # 提交事务
    conn.commit()
  7. 联表查询
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    # 执行查询
    cursor.execute('SELECT users.*, orders.* FROM users JOIN orders ON users.id = orders.user_id')

    # 获取结果
    results = cursor.fetchall()

    # 打印结果
    for row in results:
        print(row)

    # 关闭游标
    cursor.close()

    # 关闭连接
    conn.close()

Flask-SQLAlchemy的使用

Flask-SQLAlchemy是Flask框架的一个关系型数据库的扩展包，提供了简化的数据库操作方式。Flask-SQLAlchemy内部使用了SQLAlchemy库，SQLAlchemy是一个功能强大的SQL数据库映射工具。

SQLAlchemy是由Python开发的一个SQL数据库映射工具，它提供了一种简单的方式来与数据库进行交互。该框架内部封装了ORM（对象关系映射）和原生数据库的操作，可以让开发人员在不用编写SQL语句的前提下，通过Python对象操作数据库及其内部的数据。SQLAlchemy支持多种数据库，包括MySQL、PostgreSQL、SQLite等。

SQLAlchemy一般会让其内部的Dialect与数据库API进行交互，并根据不同的配置文件调用不同的数据库API，比如Pymysql。Dialect是SQLAlchemy与数据库进行通信的桥梁，它负责将SQLAlchemy的语句转换为数据库的原生语句。

在Python中一个类对应数据库中的一个表，类的属性对应表的字段，类的实例对应表的一条记录。

下面是Flask-SQLAlchemy的使用方法：

  1. 安装Flask-SQLAlchemy

    1
    pip install flask-sqlalchemy

  2. 配置数据库连接

    1
    2
    3
    4
    5
    6
    7
    from flask import Flask
    from flask_sqlalchemy import SQLAlchemy

    app = Flask(__name__)
    # mysql+pymysql://用户名:密码@主机名:端口号/数据库名
    app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://root:123456@localhost:3306/mydb'
    db = SQLAlchemy(app)

  3. 定义模型
    在Flask-SQLAlchemy中定义模型时，需要继承db.Model类，通常保存在models.py文件中，模型类的名称通常与数据库中的表名相同。

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    from flask_sqlalchemy import SQLAlchemy
    db = SQLAlchemy()

    # 定义用户模型，对应数据库中的users表，继承db.Model
    class User(db.Model):
        id = db.Column(db.Integer, primary_key=True)
        name = db.Column(db.String(50), nullable=False)
        age = db.Column(db.Integer, nullable=False)
        email = db.Column(db.String(120), nullable=False)
        password = db.Column(db.String(120), nullable=False)
        role = db.Column(db.String(20), nullable=False)
        created_at = db.Column(db.DateTime, default=db.func.now())
        updated_at = db.Column(db.DateTime, default=db.func.now(), onupdate=db.func.now())
        deleted_at = db.Column(db.DateTime, nullable=True)
        # 定义表名，如果没有__tablename__，则默认使用类名的小写形式，如果有多个单词，则使用下划线连接
        __tablename__ = 'users'

    db.Column类构造方法的参数说明：
    语法格式：

    1
    db.Column(db.Integer, primary_key=True)
  • db.Integer：整数类型
  • primary_key=True：主键
  • nullable=False：非空
  • default=db.func.now()：默认值，当前时间
  • onupdate=db.func.now()：更新时间，每次更新时自动更新为当前时间
  • unique=True：唯一，确保字段的值在表中是唯一的
  • index=True：索引，加速查询
  • db.String(50)：字符串类型，长度为50
  • db.DateTime：日期时间类型
  • db.Boolean：布尔类型，True或False
  • db.Float：浮点数类型
  • db.Text：文本类型，适用于长文本字段
  • db.JSON：JSON类型，适用于存储JSON数据
  • db.Enum：枚举类型，适用于有限的选项列表
  • db.ForeignKey('other_table.id')：外键，引用其他表的主键，通常用于建立表之间的关系。

  1. 创建数据库表

    1
    2
    3
    4
    5
    6
    7
    # 自动根据模型创建表，app.app_context()是Flask应用上下文，作用是将当前应用上下文推送到栈顶，确保在使用SQLAlchemy时可以访问到当前应用的配置。
    with app.app_context():
        db.create_all()

    # 删除所有表
    with app.app_context():
        db.drop_all()

  2. 数据库操作

    1
    2
    3
    4
    5
    6
    7
    8
    9
    # 创建用户
    user = User(name='John Doe', age=30, email='[email protected]', password='123456', role='user')
    db.session.add(user)
    db.session.commit()

    # 查询用户
    users = User.query.all()
    for user in users:
        print(user.name)

  3. 更新用户

    1
    2
    3
    4
    # 更新用户
    user = User.query.get(1)
    user.age = 35
    db.session.commit()

  4. 删除用户

    1
    2
    3
    4
    # 删除用户
    user = User.query.get(1)
    db.session.delete(user)
    db.session.commit()

  5. 数据库迁移

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    # 初始化迁移目录
    flask db init

    # 生成迁移脚本
    flask db migrate -m 'Add users table'

    # 应用迁移
    flask db upgrade

    # 回滚到上一个迁移版本
    flask db downgrade

  6. 完整示例代码

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78
    79
    80
    81
    82
    83
    84
    85
    86
    87
    88
    89
    90
    91
    92
    93
    94
    95
    96
    97
    98
    99
    100
    101
    102
    103
    104
    105
    106
    107
    108
    109
    110
    111
    112
    113
    114
    115
    116
    117
    118
    119
    120
    121
    122
    from _md5 import md5
    from enum import unique

    from flask import Flask, request, jsonify
    from flask_sqlalchemy import SQLAlchemy
    from urllib.parse import quote_plus
    from sqlalchemy import text


    app = Flask(__name__)

    USERNAME = 'root'
    PASSWORD = 'Ct310827@..'
    # 密码中有特殊符号可能会导致编码问题，因此使用 quote_plus 进行编码
    ENCODED_PASSWORD = quote_plus(PASSWORD)
    SERVER = 'localhost'
    DB = 'hj-db'

    print(f"USERNAME: {USERNAME}")
    print(f"PASSWORD: {ENCODED_PASSWORD}")
    print(f"SERVER: {SERVER}")
    print(f"DB: {DB}")

    URI = 'mysql+pymysql://{}:{}@{}/{}'.format(USERNAME, ENCODED_PASSWORD, SERVER, DB)
    print(f"Generated URI: {URI}")

    app.config['SQLALCHEMY_DATABASE_URI'] = URI
    app.config['SQLALCHEMY_TRACK_MODIFICATION'] = False
    db = SQLAlchemy(app)

    # with app.app_context():
    #     with db.engine.connect() as conn:
    #         # SQLAlchemy 2.0 及更高版本中，直接传入字符串形式的 SQL 语句不再被支持，需要使用 text 函数将其转换为可执行的 SQL 表达式对象。
    #         rs = conn.execute(text('select 1'))
    #         print(rs.fetchone())

    class User(db.Model):
        __tablename__ = 'user'
        id = db.Column(db.Integer, primary_key=True, autoincrement=True)
        username = db.Column(db.String(20), unique=True, nullable=False)
        password = db.Column(db.String(100), nullable=False)
        def to_dict(self):
            """将 User 对象转换为字典"""
            return {
                'id': self.id,
                'username': self.username,
                # 注意：实际应用中通常不返回密码
                'password': self.password
            }

        def __repr__(self):
            return '<User %r>' % self.username


    # 自动根据模型创建表
    with app.app_context():
        db.create_all()


    @app.route('/add', methods=['POST'])
    def add_user():
        username = request.form.get('username')
        password = request.form.get('password')
        # 密码md5加密
        if password:
            password = md5(password.encode('utf-8')).hexdigest()
        user = User(username=username, password=password)
        db.session.add(user)
        db.session.commit()
        if user:
            # 将 User 对象转换为字典，再使用 jsonify 转换为 JSON 响应
            return jsonify(user.to_dict())
        else:
            return jsonify({"message": "User not found"}), 404


    @app.route('/get/<int:user_id>', methods=['GET'])
    def get_user_by_id(user_id):
        # 返回json
        user = User.query.get(user_id)
        if user:
            # 将 User 对象转换为字典，再使用 jsonify 转换为 JSON 响应
            return jsonify(user.to_dict())
        else:
            return jsonify({"message": "User not found"}), 404


    def get_user_instance(user_id):
        return User.query.get(user_id)


    # 修改用户信息
    @app.route('/update', methods=['PUT'])
    def update_user():
        user_id = request.form.get('id')
        username = request.form.get('username')
        password = request.form.get('password')
        user = User.query.get(user_id)
        if not user:
            return "User not found", 404
        if username:
            user.username = username
        if password:
            password = md5(password.encode('utf-8')).hexdigest()
            user.password = password
        db.session.commit()
        if user:
            # 将 User 对象转换为字典，再使用 jsonify 转换为 JSON 响应
            return jsonify(user.to_dict())
        else:
            return jsonify({"message": "User not found"}), 404

    @app.route('/delete/<int:user_id>', methods=['DELETE'])
    def delete_user(user_id):
        user = get_user_instance(user_id)
        db.session.delete(user)
        db.session.commit()
        return jsonify({"message": "User deleted"}), 200


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