使用 SQLAlchemy 库来实现对 MySQL 数据库的增删改查

在 Flask 中使用 SQLAlchemy 库来实现对 MySQL 数据库的增删改查

from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://username:password@localhost/dbname'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

db = SQLAlchemy(app)

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(50))
    email = db.Column(db.String(50), unique=True)
    password = db.Column(db.String(50))

    def __init__(self, name, email, password):
        self.name = name
        self.email = email
        self.password = password

@app.route('/users', methods=['POST'])
def create_user():
    data = request.get_json()
    name = data['name']
    email = data['email']
    password = data['password']
    new_user = User(name, email, password)
    db.session.add(new_user)
    db.session.commit()
    return jsonify({'message': 'User created successfully!'})

@app.route('/users', methods=['GET'])
def get_users():
    users = User.query.all()
    output = []
    for user in users:
        user_data = {}
        user_data['id'] = user.id
        user_data['name'] = user.name
        user_data['email'] = user.email
        user_data['password'] = user.password
        output.append(user_data)
    return jsonify({'users': output})

@app.route('/users/<id>', methods=['GET'])
def get_user(id):
    user = User.query.get(id)
    if not user:
        return jsonify({'message': 'User not found!'})
    user_data = {}
    user_data['id'] = user.id
    user_data['name'] = user.name
    user_data['email'] = user.email
    user_data['password'] = user.password
    return jsonify({'user': user_data})

@app.route('/users/<id>', methods=['PUT'])
def update_user(id):
    user = User.query.get(id)
    if not user:
        return jsonify({'message': 'User not found!'})
    data = request.get_json()
    user.name = data['name']
    user.email = data['email']
    user.password = data['password']
    db.session.commit()
    return jsonify({'message': 'User updated successfully!'})

@app.route('/users/<id>', methods=['DELETE'])
def delete_user(id):
    user = User.query.get(id)
    if not user:
        return jsonify({'message': 'User not found!'})
    db.session.delete(user)
    db.session.commit()
    return jsonify({'message': 'User deleted successfully!'})

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

在上面的示例代码中，我们定义了一个名为 User 的数据模型，然后在 Flask 中创建了五个路由，分别用于创建用户、获取所有用户、获取单个用户、更新用户和删除用户。在每个路由中，我们都使用 SQLAlchemy 对数据库进行操作。需要注意的是，在使用 SQLAlchemy 时，我们需要在 Flask 应用中先定义数据库连接信息和数据模型。

使用 MySQL cursor 也可以直接执行 SQL 命令来实现对 MySQL 数据库的增删改查。以下是一个简单的示例代码，可以使用 MySQL cursor 来实现对 MySQL 数据库的增删改查：

import mysql.connector

# 连接 MySQL 数据库
mydb = mysql.connector.connect(
    host="localhost",
    user="username",
    password="password",
    database="dbname"
)

# 创建 MySQL cursor
mycursor = mydb.cursor()

# 插入数据
sql = "INSERT INTO users (name, email, password) VALUES (%s, %s, %s)"
val = ("John", "[email protected]", "password123")
mycursor.execute(sql, val)
mydb.commit()
print(mycursor.rowcount, "record inserted.")

# 查询数据
mycursor.execute("SELECT * FROM users")
myresult = mycursor.fetchall()
for x in myresult:
    print(x)

# 更新数据
sql = "UPDATE users SET name = %s WHERE email = %s"
val = ("Mike", "[email protected]")
mycursor.execute(sql, val)
mydb.commit()
print(mycursor.rowcount, "record(s) updated")

# 删除数据
sql = "DELETE FROM users WHERE email = %s"
val = ("[email protected]",)
mycursor.execute(sql, val)
mydb.commit()
print(mycursor.rowcount, "record(s) deleted")

在上面的示例代码中，我们首先连接到 MySQL 数据库，然后创建了一个 MySQL cursor 对象。接下来，我们可以使用 execute() 函数来执行 SQL 命令，例如插入数据、查询数据、更新数据和删除数据等。需要注意的是，在执行 SQL 命令后，我们需要使用 commit() 函数来提交事务的更改。