pythonflask框架配置mysql数据库以及SQLAlchemy对数据库的操作
mysql下载后的初次使用
  本⼈是⼀个没有做笔记习惯的低级程序员，但是我还是喜欢编程，最近发现⾃⼰的记忆⼒严重的下降，原来⾃⼰学过的东西过段差不多都忘记了，现在才想起⽤博客记录⼀下，⽅便以后⾃⼰查看，同时也是为了能更好的提⾼⾃⼰的能⼒。最近在学习flask框架和mysql，所以再此总结⼀下。
⼀ flask框架mysql⽂件
  通过看别的⼤佬的项⽬最常见的配置mysql就是
  1)创建⼀个config.py⽂件
class BaseConfig(object):
# 数据库的配置
DIALCT = "mysql"
DRITVER = "pymysql"
HOST = '127.0.0.1'
PORT = "3306"
USERNAME = "root"
PASSWORD = "123456"
DBNAME = 'test_auto'
SQLALCHEMY_DATABASE_URI = f"{DIALCT}+{DRITVER}://{USERNAME}:{PASSWORD}@{HOST}:{PORT}/{DBNAME}?charset=utf8"
SQLALCHEMY_TRACK_MODIFICATIONS = True
2)导⼊config⽂件通过app,config添加配置⽂件
from flask import Flask
from config import BaseConfig
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
#添加配置⽂件
#初始化扩展，传⼊app 创建db
db = SQLAlchemy(app)
3 )config.py配置⽂件的 SQLALCHEMY_DATABASE_URI  以及SQLALCHEMY_TRACK_MODIFICATIONS注意不要写错这些变量可以在\Python37\Lib\site-packages\flask_sqlalchemy\__init__.py⽬录下查看SQLAlchemy的默认配置⽂件和\Python37\Lib\site-packages\flask\app.py⽬录下查看config的默认配置⽂件
  fig的默认配置⽂件
default_config = ImmutableDict(
{
"ENV": None,
"DEBUG": None,
"TESTING": False,
"PROPAGATE_EXCEPTIONS": None,
"PRESERVE_CONTEXT_ON_EXCEPTION": None,
"SECRET_KEY": None,
"PERMANENT_SESSION_LIFETIME": timedelta(days=31),
"USE_X_SENDFILE": False,
"SERVER_NAME": None,
"APPLICATION_ROOT": "/",
"SESSION_COOKIE_NAME": "session",
"SESSION_COOKIE_DOMAIN": None,
"SESSION_COOKIE_PATH": None,
"SESSION_COOKIE_HTTPONLY": True,
"SESSION_COOKIE_SECURE": False,
"SESSION_COOKIE_SAMESITE": None,
"SESSION_REFRESH_EACH_REQUEST": True,
"MAX_CONTENT_LENGTH": None,
"SEND_FILE_MAX_AGE_DEFAULT": None,
"TRAP_BAD_REQUEST_ERRORS": None,
"TRAP_HTTP_EXCEPTIONS": False,
"EXPLAIN_TEMPLATE_LOADING": False,
"PREFERRED_URL_SCHEME": "http",
"JSON_AS_ASCII": True,
"JSON_SORT_KEYS": True,
"JSONIFY_PRETTYPRINT_REGULAR": False,
"JSONIFY_MIMETYPE": "application/json",
"TEMPLATES_AUTO_RELOAD": None,
"MAX_COOKIE_SIZE": 4093,
}
)
sqlalchemy的默认配置⽂件
if (
'SQLALCHEMY_DATABASE_URI'not fig and
'SQLALCHEMY_BINDS'not fig
):
warnings.warn(
'Neither SQLALCHEMY_DATABASE_URI nor SQLALCHEMY_BINDS is set. '
'Defaulting SQLALCHEMY_DATABASE_URI to "sqlite:///:memory:".'
)
track_modifications = fig.setdefault(
'SQLALCHEMY_TRACK_MODIFICATIONS', None
)
4)创建数据查询模型并继承db.Model
⼆，直接使⽤SQLAlchemy连接mysql，不⽤通过flask框架
1)，SQLAlchemy和pymsql的安装，在使⽤SQLAlchemy连接mysql前需要先给Python安装MySQL驱动，由于MySQL不⽀持Python3，所以可以同pymsql与SQLAlchemy进⾏交互
pip install pymysql
pip install sqlalchemy
2)，连接数据库连接数据库的引擎参数形式
engine = create_engine("数据库类型+数据库驱动://数据库⽤户名：数据库密码@IP地址：端⼝号/数据库?编码...", 其它参数)
注意：charset是utf8⽽不是utf-8，不能带- 不然会包异常
engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/test_auto?charset=utf8",echo=True)
echo=True,  # 设置为True，程序运⾏时反馈执⾏过程中的关键对象，包括 ORM 构建的 sql 语句
pool_size=5, # 数据库连接池初始化的容量，保持的连接数，初始化时，并不产⽣连接。只有慢慢需要连接时，才会产⽣连接
max_overflow=10, # 连接池最⼤溢出容量，该容量+初始容量=最⼤容量。超出会堵塞等待，等待时间为timeout参数值默认30
pool_timeout=30, #从连接池⾥获取连接，如果此时⽆空闲的连接。且连接数已经到达了pool_size+max_overflow。此时获取连接的进程会等待pool_timeout秒。如果超过这个时间，还没有获得将会抛出异常。默认30秒
pool_recycle=7200 # 重连周期
create_engine()返回的是Engine的⼀个实例，代表了操作数据库的核⼼接⼝，处理数据库和数据库的API，可以直接使⽤ute()或者t(）来直接建⽴⼀个DBAPI的连接，但是如果我们要使⽤ORM，就不能直接使⽤engine，初次调⽤create_engine()并不会真正连接数据库，只有在真正执⾏⼀条命令的时候才会尝试建⽴连接，⽬的是节省资源
3)，映射说明
在使⽤ORM时主要有两个配置过程；⼀是数据库表的信息描述处理；⼆是将类映射到这些表上。它们在SQLAlchemy中⼀起完成，被称为Declarative，
它们在SQLAlchemy中⼀起完成，被称为Declarative。使⽤Declarative参与的ORM映射的类需要被定义为⼀个指定基类的⼦类，这个基类含有ORM映射中相关类和表的信息。这样的基类称为declarative base class。这个基类可以通过declarative_base来创建。
declarative import declarative_base
Base = declarative_base()
4)建⽴映射关系
  数据库与 Python 对象的映射主要在体现三个⽅⾯，其中前两者是⽐较常见的映射
数据库表 (table）映射为 Python 的类 (class)，称为 model
表的字段 (field) 映射为 Column
表的记录 (record）以类的实例 (instance) 来表⽰
在sqlalchemy中表字段的常见类型如下：
Interger：整型，映射到数据库中是int类型
Float：浮点类型，float
Double; 双精度类型
String; 字符串类型需要指定字符串的长度
Boolean; 布尔类型
Decimal: 定点类型，专门为解决浮点类型精度丢失的问题⽽设定。Decimal需要传⼊两个参数，第⼀个参数标记该字段能存储多少位数，第⼆个参数表⽰⼩数点后有⼜多少个⼩数位。
Enum：枚举类型；
Date：⽇期类型，年⽉⽇；
DateTime：时间类型，年⽉⽇时分毫秒；
Time：时间类型，时分秒；
Text：长字符串，可存储6万多个字符，text；
LongText：长⽂本类型，longtext.
在指定表字段的映射为Column时不但要指定表字段的数据类型，往往还需要指定哪些字段是主
primary_key    #是否为主键
unique            #是否唯⼀
index              #如果为True，为该列创建索引，提⾼查询效率
nullable            #是否允许为空
default              #默认值
name                #在数据表中的字段映射
autoincrement  #是否⾃动增长
onupdate          #更新时执⾏的函数
comment          #字段描述
5) 创建类
根据以上字段的信息了解可以⾃定义⼀个类
__tablename__指定表名(要注意⼤⼩写)
Column类指定对应的字段，必须指定
from sqlalchemy import create_engine,Column,Integer,String,Enum
declarative import declarative_base
Base = declarative_base()
engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/test_auto?charset=utf8",echo=True)
#学⽣
class Student(Base):  #继承Base
__tablename__ = "student"  #表名
id = Column(Integer, primary_key=True,comment="学⽣ID") #为主键
name = Column(String(60), default=None, nullable=False, comment="学⽣姓名") #学⽣名  nullable能否为空  False不能为空  True可以为空
phone = Column(String(11), default=None, nullable=True,comment="学⽣电话") #电话可以为空默认是None
gender = Column(Enum("男","⼥"), default="男", nullable=False, comment="性别") #性别 Enum枚举不能为空
#这个为可选项，只是增加对于表的描述，便于以后测试，也可以描述的更加详细
def__repr__(self):
return"<User(name='%s', phone='%s', gender='%s')>" % (self.name,self.phone, der)
if __name__ == '__main__':
# adata.drop_all(engine)  #删除数据库的表
在实际编码的时候，常见的⽅式是先在数据库中建表，然后再⽤代码操作数据库。上⾯这种声明式定义映射模型，对 Column 的声明是很枯燥的。如果表的字段很多，这种枯燥的代码编写也是很痛苦的事情。
解决办法有两个：
⽅法⼀：安装 sqlacodegen 库 (pip install sqlacodegen)，然后通过下⾯的命令，基于数据库中的表⾃动⽣成 model 映射的代码。sqlacodegen ⽤法如下：
sqlacodegen --tables 表名 --outfile=路径名称 database url
database url  是与sqlalchemy的相同
sqlacodegen --table teacher --outfile=teacher.py  mysql+pymysql://root:123456@127.0.0.1:3306/test_auto?charset=utf8
⽅法⼆，在构建 model 的时候，使⽤ autoload = True，sqlalchemy 依据数据库表的字段结构，⾃动加载 model 的 Column。使⽤这种⽅法时，在构建 model 之
前，Base 类要与 engine 进⾏绑定。下⾯的代码演⽰了 autoload 模式编写 model 映射的⽅法：
from sqlalchemy import  create_engine
declarative import declarative_base
from sqlalchemy.sql.schema import Table
engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/test_auto?charset=utf8",echo=True)
Base = declarative_base()
metadata = adata
metadata.bind = engine
class Course(Base):
__tablename__= Table("course", metadata, autoload=True )
6)创建会话
在⼀个会话中操作数据库，会话建⽴在连接上，连接被引擎管理。当第⼀次使⽤数据库时，从引擎维护的连接池中获取⼀个连接使⽤。
session对象多线程不安全。所以不同线程应该使⽤不⽤的session对象。Session类和engine有⼀个就
⾏
SQLAlchemy的Session是⽤于管理数据库操作的⼀个像容器⼀样的⼯⼚对象。Session⼯⼚对象中提供query(), add(), add_all(), commit(), delete(), flush(), rollback(), close()等⽅法单线程创建session
from sqlalchemy import create_engine,Column,Integer,String,Enum
declarative import declarative_base
import sessionmaker
engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/test_auto?charset=utf8",echo=True)
Base = declarative_base()
session_factory = sessionmaker(bind=engine)
Session = session_factory()
多线程利⽤数据库连接池创建session
#数据库模块model.py
import scoped_session
import sessionmaker
engine = create_engine(
"mysql+pymysql://root:123456@127.0.0.1:3306/test_auto?charset=utf8",echo=True,
max_overflow=10, # 超过连接池⼤⼩外最多创建的连接
pool_size=5, # 连接池⼤⼩
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进⾏⼀次连接的回收（重置）
)
session_factory = sessionmaker(bind=engine)
Session = scoped_session(session_factory)
####业务模块thread.py
import threading
from model import Session
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String(20))
fullname = Column(String(20))
password = Column(String(20))
age = Column(Integer)
class MyThread(threading.Thread):
def__init__(self, threadName):
super(MyThread, self).__init__()
self.name = threading.current_thread().name
def run(self):
session = Session() #每个线程都可以直接使⽤数据库模块定义的Session，执⾏时，每⼀个session都相当于⼀个connection
session.query(User).all()
user = User(name="hawk-%s"%self.name, fullname="xxxx",password="xxxx",age=10)
session.add(user)
time.sleep(1)
if self.name == "thread-9":
sessionmit()
if__name__ == "__main__":
arr = []
for i in xrange(10):
arr.append(MyThread('thread-%s' % i))
for i in arr:
i.start()
for i in arr:
i.join()
7) sqlalchemy对数据库的基本操作
#创建⼀个表
# adata.drop_all(engine)  #删除数据库的表
#添加  add()：增加⼀个对象
new_stu = Student(name='张三', gender="男", phone='158********')
Session.add(new_stu)  # 添加⼀个
Sessionmit()  # 需要调⽤commit()⽅法提交事务
#添加  add_all()：可迭代对象，元素是对象
new_stu1 = Student(name='李四', gender="⼥", phone='158********')
new_stu2 = Student(name='王⼆', gender="男", phone='158********')
new_stu3 = Student(name='往⼆', gender="⼥", phone='158********')
Session.add_all([new_stu1,new_stu2,new_stu3])  #⼀次性添加⼀批
Sessionmit()  # 需要调⽤commit()⽅法提交事务
#查后添加first(), all()等返回对应的查询对象
query() 不进⾏条件查询
Session.query(Student).all() #查询所有返回的对象是list列表
Session.query(Student).first() #查询第⼀个
query()进⾏条件筛选  Query 对象提供了 filter() ⽅法和 filter_by() ⽅法⽤于数据筛选。filter_by() 适⽤于简单的基于关键字参数的筛选。 filter() 适⽤于复杂条件的表达
Session.query(Student).filter(Student.id<=3).all()
Session.query(Student).filter(Student.name=='张三').first()
Session.query(Student).filter_by(name='张三').first()
如果我们要出所有id< =3 的数据只能⽤ filter() ⽅法，因为 filter_by 只⽀持关键字参数，不能实现
如果有两个限制条件是AND关系，可以直接使⽤两次filter()处理
Session.query(Student).filter(Student.name == '张三').der == '男').first()
#或者是
Session.query(Student).filter_by(name='张三').der=='男').first()
当然也有其他的展⽰⽅法
#filter operator :
==    #相等
!=    #不相等
like(’%关键字%’)  #通配符搜索
ilike()  #通配符搜索(不敏感)
.in_([‘张三’, ‘张三1’, ‘张三2’])  #存在于
~
User.name.in_([‘张三’, ‘张三1’, ‘张三2’])  #不存在于
==None    #为空
.filter(Student.name == ‘张三’, der == ‘⼥’)  #and
or_(Student.name == ‘张三’, Student.name == ‘张三2’)    #OR
#检索返回的列表，以及列表的标量：
all()    #返回所有
first()    #返回第⼀⾏
one()    #检查是不是只有⼀⾏结果如果存在则返回改数据如果不存在则报异常
one_or_none()    #检查是不是⼀⾏或者没有结果如果存在则返回该数据若不存在则返回None
⼤致常见条件的表达，因为⽐较直观
def test_filter_le(self):
Stu = Session.query(Student).filter(Student.id <= 3).all()
print(Stu)
def test_filter_ne(self):
Stu = Session.query(Student).filter(Student.id != 2).all()
print(Stu)
def test_filter_like(self):
Stu = Session.query(Student).filter(Student.id.like('%9')).all()
print(Stu)
def test_filter_in(self):
Stu = Session.query(Student).filter(Student.EDUCATION.in_(['Bachelor', 'Master'])).all()
print(Stu)
def test_filter_notin(self):
Stu = Session.query(Student).filter(~Student.EDUCATION.in_(['Bachelor', 'Master'])).all()
print(Stu)
def test_filter_isnull(self):
Stu = Session.query(Student).filter(Student.MARITAL_STAT == None).all()
print(Stu)
def test_filter_isnotnull(self):
Stu = Session.query(Student).filter(Student.MARITAL_STAT != None).all()
print(Stu)
def test_filter_and(self):
Stu = Session.query(Student).filter(Student.GENDER=='Female', Student.EDUCATION=='Bachelor').all()
print(Stu)
def test_filter_and2(self):
Stu = Session.query(Student).filter(and_(Student.GENDER=='Female', Student.EDUCATION=='Bachelor')).all() print(Stu)
def test_filter_or(self):
Stu = Session.query(Student).filter(or_(Student.MARITAL_STAT=='Single', Student.NR_OF_CHILDREN==0)).all() print(Stu)
#更新（注意要重新commit()提交）
stu = Session.query(Student).filter(Student.name == '张三').first()  #第⼀步查询出要修改的数据
print(f"修改之前={stu}")
Session.add(stu)  #第3步然后重新添加
Sessionmit()  #第4步最后再重新提交
#或者
Session.query(Student).filter(Student.name == '张三').update({"gender":"⼥"})
Sessionmit()  #最后再重新提交
# 删除
stu = Session.query(Student).filter(Student.name == '张三').first()  #查询数据
Session.delete(stu)  #删除数据
Sessionmit()  #再次提交
#或者
stu = Session.query(Student).filter(Student.name == '张三').delete()  #查询数据并删除
print(stu)  #输出的是删除的个数
Sessionmit()  #再次提交
8) 执⾏原⽣的SQL语句
从SQLAlchemy中导出text⽅法，可以通过text(SQL语句)嵌⼊使⽤SQL语句。
⽅法⼀：使⽤Session 进⾏执⾏原⽣的sql语句
from sqlalchemy import text
stu = Session.query(Student).filter(text('name="张三"')).first()
print(f'查询数据：{stu}')
通过Engine对象执⾏原⽣的SQL语句
  sql = "select * from  student WHERE name='张三'"
t() as conn:
stu = ute(sql)
result= stu.fetchall()
print(f'执⾏原⽣SQ结果={result}')
9)建⽴表与表之间的关系（⼀对多，多对⼀）

版权声明：本站内容均来自互联网，仅供演示用，请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系QQ:729038198，我们将在24小时内删除。