flask sqlalchemy 操作数据相关代码和实现方法

admin · 发表于 2024-7-28 17:00:09

1.介绍
flask用于orm操作表，一般使用flask- sqlalchemy 操作简单
一般使用flask-sqlalchemy 2.5.1
https://flask-sqlalchemy.palletsprojects.com/en/2.x/

2.安装
pip instal flask-sqlalchemy==2.5.1
pip install Flask-Migrate==2.6.0  # 迁移数据库使用

3.数据库
1.配置连接数据库
代表驱动
MySQL-Python  # 可以写mysqldb
mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>

pymysql
mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]

MySQL-Connector
mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname

需要安装不同的扩展注意mysqlclient的版本
可以放到配置类里面

方法一
SQLALCHEMY_DATABASE_URI = 'mysql://user:密码@localhost:3306/库名?charset=utf8mb4'

方法二
DIALECT = 'mysql'
DRIVER = 'pymysql'
USERNAME = 'root'
PASSWORD = '****'
HOST = '127.0.0.1'
PORT = '3306'
DATABASE = '数据库名'

SQLALCHEMY_DATABASE_URI = "{}+{}://{}:{}@{}:{}/{}?charset=utf8".format(DIALECT,DRIVER,USERNAME,PASSWORD,HOST,PORT,DATABASE)

2.连接数据库
1 实例化SQLAlchemy对象  可放入其他文件
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()

2 注册db

# 初始化迁移数据库
migrate = Migrate()

def create_app() -> Flask:
app = Flask(__name__)
app.config.from_object(Config)
# 绑定数据库
db.init_app(app)

# 迁移数据库
migrate.init_app(app, db)
return app

注意命令可能无法迁移

1. 命令行一直报
# Error: Could not locate a Flask application. Use the 'flask --app' option, 'FLASK_APP' environment variable, or a 'wsgi.py' or 'app.py' file in the current directory.

这是在虚拟环境没有配置

使用 .env 文件来设置数据
# 下载插件
pip install python-dotenv

# 创建app
def create_app() -> Flask:
app = Flask(__name__)
# 要在最开始加载后面才能获取到
load_dotenv(verbose=True)

# 注册配置文件生成/测试
# env = str(os.environ.get("FLASK_ENV"))
env = str(os.getenv("FLASK_ENV"))
app.config.from_object(getattr(conf, env))

# 绑定数据库
db.init_app(app)

# 迁移数据库
migrate.init_app(app, db)
return app
#-------------------------------------
# .env 文件

FLASK_APP=main.py # 自己的启动文件默认是app.py
FLASK_ENV=development
FLASK_DEBUG=True

这样就可以输入命令了

报错误 either 'sqlalchemy_database_uri' or 'sqlalchemy_binds' must be set.
没有把模型放到 view文件里面只想要把模型导入就好了，可以不使用
from user.model import User

3 迁移数据库
# 初始化一个迁移脚本的环境，只需要执行一次。
flask db init
# 将模型生成迁移文件，只要模型更改了，就需要执行一遍这个命令
flask db migrate -m'迁移数据库'
# 将迁移文件真正的映射到数据库中。每次运行了migrate命令后，就记得要运行这个命令。
flask db upgrade
#-------------------------------------
flask db heads 查看当前最新的迁移id
flask db stamp heads # 更新
flask db current  # 查看当前迁移id
python manage.py db history  #查看历史
python manage.py db downgrade 652a55840d  #回退版本

4 数据库其他信息
# 查询时会显示原始SQL语句
SQLALCHEMY_ECHO = True
# 动态追踪修改设置，如未设置只会提示警告
SQLALCHEMY_TRACK_MODIFICATIONS = False
# 数据库连接池的大小
SQLALCHEMY_POOL_SIZE = 10
# 指定数据库连接池的超时时间
SQLALCHEMY_POOL_TIMEOUT = 10
# 控制在连接池达到最大值后可以创建的连接数。当这些额外的连接回收到连接池后将会被断开和抛弃。
SQLALCHEMY_MAX_OVERFLOW = 2

名字       备注
SQLALCHEMY_DATABASE_URI       用于连接的数据库 URI 。‘mysql+pymysql://root:密码@localhost:3306/库名?charset=utf8mb4’
SQLALCHEMY_BINDS       一个映射 binds 到连接 URI 的字典。更多 binds 的信息见用 Binds 操作多个数据库。
SQLALCHEMY_ECHO       如果设置为Ture， SQLAlchemy 会记录所有发给 stderr 的语句，这对调试有用。(打印sql语句)
SQLALCHEMY_RECORD_QUERIES       可以用于显式地禁用或启用查询记录。查询记录在调试或测试模式自动启用。更多信息见get_debug_queries()。
SQLALCHEMY_NATIVE_UNICODE       可以用于显式禁用原生 unicode 支持。当使用不合适的指定无编码的数据库默认值时，这对于一些数据库适配器是必须的（比如 Ubuntu 上某些版本的 PostgreSQL ）。
SQLALCHEMY_POOL_SIZE       数据库连接池的大小。默认是引擎默认值（通常是 5 ）
SQLALCHEMY_POOL_TIMEOUT       设定连接池的连接超时时间。默认是 10 。
SQLALCHEMY_POOL_RECYCLE       多少秒后自动回收连接。这对 MySQL 是必要的，它默认移除闲置多于 8 小时的连接。注意如果使用了 MySQL ， Flask-SQLALchemy 自动设定这个值为 2 小时。
4 创建表
from core import db       #这里需要引入实例化的SQLAlchemy对象

class User(db.Model):
"""用户表"""
__tablename__ = 'user'  # 创建后表名小写
# __abstract__ = True  # 抽象表不创建这个表可以继承

id = db.Column(db.Integer, primary_key=True, comment='id主键') # 必须要有
username = db.Column(db.String(32), unique=True, comment='用户名')
password = db.Column(db.String(64), nullable=False, comment='密码')
gender = db.Column(db.Boolean, default=0, comment='0 男 1 女')
avatar = db.Column(db.String(125), default='http://s5p.hd-bkt.clouddn.com/default.jpg', comment='头像')
addr = db.Column(db.String(256), comment='地址')
email = db.Column(db.String(125), comment='邮箱')
mobile = db.Column(db.String(11), comment='电话')
login_type = db.Column(db.Integer, comment='1 电话 2 邮箱 3 用户名')

def __repr__(self): # 查询的返回和str一样
      return f'<{self.__class__.__name__} {self.username}>'

1.字段解析
类型名       python中类型       说明
Integer       int       普通整数，一般是32位
SmallInteger       int       取值范围小的整数，一般是16位
BigInteger       int或long       不限制精度的整数
Float       float       浮点数
Numeric       decimal.Decimal       普通整数，一般是32位
String       str       变长字符串
Text       str       变长字符串，对较长或不限长度的字符串做了优化
Unicode       unicode       变长Unicode字符串
UnicodeText       unicode       变长Unicode字符串，对较长或不限长度的字符串做了优化
Boolean       bool       布尔值
Date/DateTime       datetime.date       时间
Time       datetime.datetime       日期和时间
LargeBinary       str       二进制文件
2.表属性
选项名       说明
primary_key       如果为True，代表表的主键
unique       如果为True，代表这列不允许出现重复的值
index       如果为True，为这列创建索引，提高查询效率
nullable       如果为True，允许有空值，如果为False，不允许有空值
default       为这列定义默认值
5.查询
注意只要对数据库操作就需要 db.session.commit()

查询方法
方法       说明
all()       以列表形式返回查询的所有结果
first()       返回查询的第一个结果，如果未查到，返回None
first_or_404()       返回查询的第一个结果，如果未查到，返回404
get()       返回指定主键对应的行，如不存在，返回None
get_or_404()       返回指定主键对应的行，如不存在，返回404
count()       返回查询结果的数量
paginate()       返回一个Paginate对象，它包含指定范围内的结果
1.普通查询
# 查询数据条数
User.query.count()

# 查单条获取不到none
user = User.query.first()
user = User.query.filter(User.username=="小红").first()
user = db.session.query(User).filter(User.username=="小红").first()
'''
db.session:是一种更明确的查询语法，适用于更复杂的查询场景，或者在不直接访问模型的情况下执行查询操作
'''

# 查所有
user = User.query.all()
user = User.query.filter(User.username=="小红").all()
user = db.session.query(User).filter(User.username=="小红").all()

# 只获取单条
# 结果为一个时正常，多了就报错
# one_or_none() 获取不到报错
user=User.filter(User.username=='小红').one()

# 获取不到报错
# get里面需要参数是id
user = User.query.get(2)

2.条件查询
# 条件查询
  user = User.query.filter(User.username=="小红",User.password=="123").first()
# 不等于
  user = User.query.filter(User.id!=2).all()
# 大小于
  user = User.query.filter(User.id>=2).all()
  user = User.query.filter(User.id<=2).all()

#----------------------------------------
# 结尾为红
User.query.filter(User.name.endswith("红")).all()
User.query.filter(User.name.like("%红")).all()
# 开头为小
User.query.filter(User.name.startswith("小")).all()
User.query.filter(User.name.like("小%")).all()
# 包含n mysql语法
User.query.filter(User.name.contains("n")).all()
User.query.filter(User.name.like("%n%")).all()  # 模糊查询

#-----------------------------------------
# 成员属于  in_
User.query.filter(User.username.in_(['aaa','bbb']).all()
#成员不属于  notin_
User.query.filter(User.username.notin_(['aaa','bbb']).all()

#----------------------------------------
from sqlalchemy import and_, or_, not_
# 和
User.query.filter(User.name.startswith('a'), User.email.startswith('q')).all()
User.query.filter(and_(User.name.startswith('a'), User.email.startswith('q'))).all()
# 或者
User.filter(or_(User.username=='qaz',User.password=='qwe123')).all()
# 非
User.filter(not_(User.username=='qaz',User.password=='qwe123')).all()

#是否空判断
User.filter(User.username==None).all() )
User.filter(User.username.is_(None)).all() )
User.filter(User.username.isnot(None)).all() )

#------------------------------------------
# limit 限制返回数据条数
# 开始条数是0
User.filter(User.gender!=1).limit(2).all())
# 从第几条开始获取偏移
User.query.offset(0).all()
# 从第2条数据到第3条包前不报后
User.filter(User.username!='qq').slice(2,3).all())

# 排序 order_by 默认正序
User.filter(User.username!='aa').order_by(User.username).all()
User.filter(User.username!='aa').order_by(User.id.asc()).all() 正序
User.filter(User.username!='aa').order_by(User.id.desc()).all() 倒叙

# 注意：上面的可以随意组合
# 其余后续补充

3.filter和filter_by
filter_by和filter都是过滤条件
filter_by 不支持比较运算符,只能用=

# 返回的都是对象
rows1 = User.query.filter_by(username='小红').all()
rows2 = User.query..filter(User.username=='小红').all()

# rows3 返回的是User的username属性，不是对象
rows3 = db.session.query(User.username).filter(User.username=='小红').all()
# 返回的是select语句
rows4 = db.session.query(User.username).filter(User.username=='小红')

6.增加
# 单数据增加
  user = User(username="张三",password ="123",email="123@qq.com")
  db.session.add(user)
  db.session.commit()

#-----------------------------------
# 多数据增加
# 后面两种可能会有颜色警告
方法一：使用字典列表一次性插入多条数据
@bl_user.route('/index')
def index():
data_list = [
      {'username': 'value1', 'password': 'value2'},
      {'username': 'value3', 'password': 'value4'},
]
db.session.bulk_insert_mappings(User, data_list)
# db.session.bulk_insert_mappings(模型, 数据)
db.session.commit()
return '增加成功'

方法二：将多个对象一次性保存到数据库
@bl_user.route('/index')
def index():
data_list = [
      User(username='value1', password='value2'),
      User(username='value3', password='value4'),
]
db.session.bulk_save_objects(data_list)
db.session.commit()
return '增加成功'

方法三：
@bl_user.route('/index')
data_list = [
      {'username': 'value1', 'password': 'value2'},
      {'username': 'value3', 'password': 'value4'},
]

data_objects = [User(**data) for data in data_list]
db.session.add_all(data_objects)
db.session.commit()
return '增加成功'

7.修改
方法一：先查询后修改可能会丢失数据
user = User.query.filter(User.username=="小红").first()
user.email = "abc@163.com"
db.session.commit()

方法二：直接修改推荐
User.query.filter(User.username == 'qqqq').update({'email': 'ABC@qq.com'})
db.session.commit()

#--------------------------
# 批量修改列表套字典
# 注意必须要有id 确定修改那一条
user = User.query.filter(User.id == 1).first()
data = [{'id': user.id, 'username': 'zzzzz', 'password': '12345678'},
         {'id': 2, 'username': 'aaa', 'password': 'zxcv'}]
db.session.bulk_update_mappings(User, data)
db.session.commit()

8.删除
方法一：先查询后删除推荐
user = User.query.filter(User.username=="小红").first()
db.session.delete(user)
db.session.commit()

方法二：直接删除可以删除多个
user = User.query.filter(User.id == 2).delete()
db.session.commit()

# 假设有多个xxx 那么就可以把xxx全部删除
user = User.query.filter(User.username== "xxx").delete()
db.session.commit()
cv'}]
db.session.bulk_update_mappings(User, data)
db.session.commit()

8.删除
方法一：先查询后删除推荐
user = User.query.filter(User.username=="小红").first()
db.session.delete(user)
db.session.commit()

方法二：直接删除可以删除多个
user = User.query.filter(User.id == 2).delete()
db.session.commit()

# 假设有多个xxx 那么就可以把xxx全部删除
user = User.query.filter(User.username== "xxx").delete()
db.session.commit()

admin · 发表于 2024-7-30 22:52:09

A Simple Example
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///example.sqlite"

class Base(DeclarativeBase):
pass

db = SQLAlchemy(app, model_class=Base)

class User(db.Model):
id: Mapped[int] = mapped_column(db.Integer, primary_key=True)
username: Mapped[str] = mapped_column(db.String, unique=True, nullable=False)

with app.app_context():
db.create_all()

db.session.add(User(username="example"))
db.session.commit()

users = db.session.execute(db.select(User)).scalars()

		自动登录	找回密码
密码			注册