python3 Flask应用 使用 Flask-SQLAlchemy操作MySQL数据库

一、环境搭建

下载命令：

pip install flask flask-sqlalchemy pymysql

二、创建项目结构

yourProjectFolder/
|—— app.py
|—— config.py
|—— models.py
|__ mydb.py

三、基本使用

3.1 config.py 进行数据库连接配置

import os
basedir = os.path.abspath(os.path.dirname(__file__))
class Config(object):
    SECRET_KEY = os.environ.get('SECRET_KEY') or 'you-will-never-guess'
    SQLALCHEMY_DATABASE_URI = 'mysql+pymysql://用户名:密码@数据库地址/数据库名'
    SQLALCHEMY_TRACK_MODIFICATIONS = False  # 关闭修改跟踪，节省资源

3.2 models.py 定义数据库模型

from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()
class User(db.Model):
    __tablename__ = 'user'
    id = db.Column(db.BigInteger, primary_key=True, autoincrement=True, comment='主键id')
    username = db.Column(db.String(255), index=True, unique=True, nullable=False, comment='用户名（手机号）')
    password = db.Column(db.String(255), nullable=False, comment='密码')
    ip = db.Column(db.String(255), nullable=True, default=None, comment='注册时IP地址')
    ipCity = db.Column(db.String(255), nullable=True, default=None, comment='注册时的ip所在地')
    registerTime = db.Column(db.String(255), nullable=True, default=None, comment='注册时间')
    def __repr__(self):
        return '<User {}>'.format(self.username)

3.3 mydb.py 增删改查例子

from flask import Flask
from sqlalchemy.exc import SQLAlchemyError
from config import Config
from models import db, User
app = Flask(__name__)
app.config.from_object(Config)
# 初始化数据库
db.init_app(app)
# 新增例子
def insert(username, password, ip, ip_city, register_time):
    is_user = login_user_db(username)
    if is_user is not None:
        return 1
    else:
        with app.app_context():
            try:
                u = User(username=username, password=password, ip=ip, ipCity=ip_city, registerTime=register_time)
                db.session.add(u)
                db.session.commit()
                return 2
            except Exception as e:
                db.session.rollback()
                print(f"发生错误: {e}")
                return 3
# 查询例子
def select(username):
    with app.app_context():
        return User.query.filter_by(username=username).first()
        # 稍微复杂一点的查询
        # return User.query.filter(User.username.like('%john%')).first()
# 修改例子
def update(username,password):
    with app.app_context():
        try:
            user = User.query.filter_by(username=username).first()
            if user:
                user.password = str(password)
                db.session.commit()
                return True
            else:
                print(f"密码重置用户 {username} 没有.")
                return False
        except SQLAlchemyError as e:
            db.session.rollback()
            print(f"An error occurred: {e}")
            return False
# 删除例子
def delete(username):
    with app.app_context():
        try:
            # 查找用户 通过用户名
            user = User.query.filter_by(username=username)
            # .get()方法用的是主键
            # user = User.query.get(1) 
            if user:
                db.session.delete(user)  # 删除用户
                db.session.commit()  # 提交事务
                return True
            else:
                print(f"没有找到用户 {username}.")
                return False
        except SQLAlchemyError as e:
            db.session.rollback()  # 回滚事务
            print(f"An error occurred: {e}")
            return False

3.3 app.py 调用 增删该查例子

import mydb
mydb.update(username=username, password=password)
db_user = mydb.select(username)
db_user.id
db_user.username
db_user.password