快捷導(dǎo)航

flask連接并操作數(shù)據(jù)庫的實現(xiàn)示例

更新時間：2024年12月31日 09:08:27 作者：奈何不吃魚

本文詳細(xì)介紹了在Flask框架中連接并操作數(shù)據(jù)庫的方法,文中通過示例代碼介紹的非常詳細(xì),對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧

前言

在Flask框架中，數(shù)據(jù)庫的操作是一個核心功能，它允許開發(fā)者與后端數(shù)據(jù)庫進行交互，執(zhí)行數(shù)據(jù)的增刪改查操作。本文將詳細(xì)介紹如何在Flask中連接并操作數(shù)據(jù)庫，包括使用Flask-SQLAlchemy擴展和直接使用MySQL的Python驅(qū)動。

常見的有三種操作方法：使用 SQLAlchemy、使用 Flask-SQLAlchemy 連接 MySQL和直接使用 MySQL 的 Python 驅(qū)動，本篇先講述如何使用SQLAlchemy進行連接，后續(xù)會對其他方式進行講解。

一、安裝必要的庫

首先，安裝Flask-SQLAlchemy和MySQL的Python驅(qū)動?？梢酝ㄟ^以下命令安裝：

pip install -U SQLAlchemy
pip install pymysql  # Python3 使用pymysql
pip install mysqlclient

二、配置數(shù)據(jù)庫連接

在Flask應(yīng)用中，需要配置數(shù)據(jù)庫的地址（一般情況下，不建議直接在項目頭部配置，此處只為演示，建議寫在config.py中進行變量配置），這里的username等信息需要換成實際信息：

from flask import Flask
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)

三、定義模型

在Flask-SQLAlchemy中，模型是數(shù)據(jù)庫表的Python類，每個模型類代表數(shù)據(jù)庫中的一張表。例如：

class User(db.Model):
    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)

    def __repr__(self):
        return f'<User {self.username}>'

這里定義了一個用戶模型，對應(yīng)數(shù)據(jù)庫中的用戶信息表，屬性包括：id、用戶名和email，對應(yīng)的mysql表如下：

CREATE TABLE `users` (
  `id` INT NOT NULL AUTO_INCREMENT,
  `username` VARCHAR(80) NOT NULL,
  `email` VARCHAR(120) NOT NULL,
  PRIMARY KEY (`id`),
  UNIQUE INDEX `username_unique` (`username` ASC),
  UNIQUE INDEX `email_unique` (`email` ASC)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;

四、操作數(shù)據(jù)庫

配置好以上內(nèi)容后，我們就可以通過SQLAlchemy 提供的 ORM 功能操作數(shù)據(jù)庫了，我們簡單的用CURD操作進行測試 ~

1.添加用戶

# 添加用戶路由
@app.route('/add_user', methods=['POST'])
def add_user():
    data = request.get_json()
    username = data.get('username')
    email = data.get('email')

    if not username or not email:
        return '缺少用戶名或郵箱地址', 400

    # 檢查 email 是否已存在
    existing_user = User.query.filter_by(email=email).first()
    if existing_user:
        return '郵箱已存在，無法添加用戶', 400

    new_user = User(username=username, email=email)
    db.session.add(new_user)
    try:
        db.session.commit()
    except IntegrityError:
        db.session.rollback()
        return '插入失敗，可能是用戶名或郵箱重復(fù)', 400
    except Exception as e:
        db.session.rollback()
        return f'數(shù)據(jù)庫錯誤: {str(e)}', 500

    return '用戶添加成功'

2.刪除用戶

# 刪除用戶路由
@app.route('/delete_user/<int:user_id>', methods=['DELETE'])
def delete_user(user_id):
    user = User.query.get(user_id)
    if not user:
        return '用戶不存在', 404

    db.session.delete(user)
    db.session.commit()
    return '用戶刪除成功'

3.更新用戶信息

# 更新用戶路由
@app.route('/update_user/<int:user_id>', methods=['PUT'])
def update_user(user_id):
    user = User.query.get(user_id)
    if not user:
        return '用戶不存在', 404

    data = request.get_json()
    new_username = data.get('username')
    new_email = data.get('email')

    # 更新用戶名和郵箱
    if new_username:
        user.username = new_username
    if new_email:
        existing_user = User.query.filter_by(email=new_email).first()
        if existing_user and existing_user.id != user_id:
            return '郵箱已存在，無法更新用戶信息', 400
        user.email = new_email

    try:
        db.session.commit()
    except IntegrityError:
        db.session.rollback()
        return '更新失敗，可能是用戶名或郵箱重復(fù)', 400
    except Exception as e:
        db.session.rollback()
        return f'數(shù)據(jù)庫錯誤: {str(e)}', 500

    return '用戶更新成功'

4查詢所有用戶

# 獲取所有用戶路由
@app.route('/get_all', methods=['GET'])
def get_users():
    users = User.query.all()
    return '<br>'.join([f'{user.username} ({user.email})' for user in users])