文章目录
一、SQLAlchemy介绍和快速使用
1.介绍
SQLAlchemy是一个基于Python实现的ORM框架。该框架建立在 DB API之上，使用关系对象映射进行数据库操作，简言之便是：将类和对象转换成SQL，然后使用数据API执行SQL并获取执行结果。
SQLAlchemy是一个企业级的orm框架，Django的orm框架，只能在Django框架中使用，不能在别的地方使用，而SQLAlchemy可以单独使用，也可以用在其他框架中
-flask ： sqlalchemy
-fastapi：sqlalchemy
python界的orm框架
-django orm
-sqlalchemy
-peewee
组成部分：
- Engine，框架的引擎
- Connection Pooling ，数据库连接池
- Dialect，选择连接数据库的DB API种类
- Schema/Types，架构和类型
- SQL Exprression Language，SQL表达式语言
安装
pip install sqlalchemy
SQLAlchemy本身无法操作数据库，其必须依赖pymsql等第三方插件，Dialect用于和数据API进行交流，根据配置文件的不同调用不同的数据库API，从而实现对数据库的操作，如：
MySQL-Python
mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>
pymysql
mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]
MySQL-Connector
mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname>
cx_Oracle
oracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value...]
更多：http://docs.sqlalchemy.org/en/latest/dialects/index.html
# django中如何反向生成models
python manage.py inspectdb > app/models.py
2.sqlalchemy原生操作
'使用sqlalchemy操作原生sql'
import pymysql
1.导入
from sqlalchemy import create_engine
from sqlalchemy.engine.base import Engine
2.创建引擎
engine = create_engine(
"mysql+pymysql://root:1234@127.0.0.1:3306/cnblogs",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
'================================'
3.使用引擎拿到链接
conn = engine.raw_connection()
4.剩下的和之前操作sql一样，使用pymysql
cursor=conn.cursor(pymysql.cursors.DictCursor)
cursor.execute('select * from article limit 10')
res = cursor.fetchall()
print(res)
'使用多线程测试：'
from threading import Thread
def task(arg):
conn = engine.raw_connection()
cursor = conn.cursor()
cursor.execute('select * from article')
res = cursor.fetchall()
print(res)
cursor.close()
conn.close()
for i in range(20):
t = Thread(target=task,args=(i,))
t.start()
二、sqlalchemy操作表
1.创建、删除表
1.导入
import datetime
from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index
Base = declarative_base() # Base当成models.Model
2.创建表模型
class User(Base):
__tablename__ = 'users' # 创建的表名
# 写字段
id = Column(Integer,primary_key=True,autoincrement=True) # 主键、自增
name = Column(String(32), index=True, nullable=False) # name列，索引，不可为空 普通索引
email = Column(String(32), unique=True) # 唯一索引
# datetime.datetime.now不能加括号，加了括号，以后永远是当前时间
ctime = Column(DateTime, default=datetime.datetime.now)
extra = Column(Text) # 大文本
3.没有命令，后期可以使用第三方模块，可以有命令操作
# 目前需要手动做
# sqlalchemy 不能创建数据库，能创建表，不能删除增加字段(第三方模块)
3.1.创建引擎
engine = create_engine( # 需要自己手动创建库
"mysql+pymysql://root:1234@127.0.0.1:3306/sqlalchemy01",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
3.2.把表模型同步到数据库中（能创建删除，不能修改）
Base.metadata.create_all(engine)
3.3.删除表
Base.metadata.drop_all(engine)
2.简单操作（orm）
from models import User
1.创建引擎
from sqlalchemy import create_engine
engine = create_engine( # 需要自己手动创建库
"mysql+pymysql://root:1234@127.0.0.1:3306/sqlalchemy01",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
2.orm操作，借助于engine 得到session(conn)对象
from sqlalchemy.orm import sessionmaker
Connection = sessionmaker(bind=engine)
conn = Connection() # 不是flask中的session，而是会话，conn的
3. 拿到(conn)会话对象后，进行orm操作
3.1 增加数据
user = User(name='jack',email='1@qq.com')
# 插入到数据库
conn.add(user) # 放个对象
# 提交
conn.commit()
# 关闭连接
conn.close()
3.2 查询数据
# 查询User表中id为1的所有数据-->是一个列表
res = conn.query(User).filter_by(id=1).all()
print(res)
3.3 删除数据
res = conn.query(User).filter_by(name='jack').delete()
print(res) # 返回删除的条数
conn.commit() # 提交
3.4 修改
res = conn.query(User).filter_by(name='jack').update({'name':'oscar','extra':'xxxx'})
conn.commit()
三、SQL表模型：一对多关系
1.表模型
import datetime
from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base, relationship
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index
Base = declarative_base() # Base 当成 models.Model
# 一对多 ：一个兴趣被多个人喜欢 一个人只喜欢一个兴趣
class Hobby(Base):
__tablename__ = 'hobby'
id = Column(Integer, primary_key=True)
caption = Column(String(50), default='篮球')
def __str__(self):
return self.caption
def __repr__(self):
return self.caption
class Person(Base):
__tablename__ = 'person'
id = Column(Integer, primary_key=True)
name = Column(String(32), index=True, nullable=True)
# hobby指的是tablename而不是类名，uselist=False
# 外键关联--》强外键--》物理外键
hobby_id = Column(Integer, ForeignKey("hobby.id")) # 加一个unique=True就是一对一的外键关系
# 跟数据库无关，不会新增字段，只用于快速连表操作
# 基于对象的跨表查询:就要加这个字段，取对象 person.hobby pserson.hobby_id
# 类名，backref用于反向查询
hobby = relationship('Hobby', backref='pers')
def __str__(self):
return self.name
def __repr__(self):
return self.name
if __name__ == '__main__':
# 创建引擎
engine = create_engine(
"mysql+pymysql://root:1234@127.0.0.1:3306/sqlalchemy02",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
# 把表模型同步到数据库中
Base.metadata.create_all(engine)
2.新增和基于对象的跨表查询
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from models import Hobby, Person
engine = create_engine(
"mysql+pymysql://root:1234@127.0.0.1:3306/sqlalchemy02",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
Session = sessionmaker(bind=engine)
session=Session()
'''一对多关系新增'''
# 增加 Hobby
hobby = Hobby(caption='足球')
hobby1 = Hobby()
session.add_all([hobby,hobby1])
session.commit()
'''三种新增方案'''
# 方案一
person = Person(name='jack',hobby_id=1)
person1 = Person(name='tom',hobby_id=2)
session.add_all([person,person1])
session.commit()
# 方案二
# 增加person表数据时直接增加关联的hobby表数据
hobby = Hobby(caption='乒乓球')
person = Person(name='oscar',hobby=hobby) # 这里的直接使用对象的方式，前提是表模型中必须有relationship
session.add(person)
person1 = Person(name='kami',hobby=Hobby(caption='棒球'))
session.add(person1)
session.commit()
# 方案三
hobby = session.query(Hobby).filter_by(id=1).first()
print(hobby)
person = Person(name='ankn',hobby=hobby)
person1 = Person(name='tank',hobby_id=hobby.id)
session.add_all([person,person1])
session.commit()
'''基于对象的跨表查询--正向'''
person = session.query(Person).filter_by(id=2).first()
# person = session.query(Person).filter_by(name='tom').first()
print(person) # tom
# 基于对象的跨表查询，正向查询
print(person.hobby.caption) # 篮球
print(person.hobby_id) # 2
'''基于对象的跨表查询--反向'''
hobby = session.query(Hobby).filter_by(id=1).first()
# hobby = session.query(Hobby).filter_by(caption='足球').first()
print(hobby) # 足球
# 基于对象的跨表查询，反向查询
print(hobby.pers) # [jack, ankn, tank] 是一个列表套对象
print(hobby.pers[1].name)
# for i in hobby.pers: # 循环拿出每一个对象
# print(i.name)
四、SQL表模型：多对多关系
1.表模型
import datetime
from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base, relationship
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index
Base = declarative_base() # Base 当成 models.Model
'''多对多'''
class Boy2Girl(Base):
__tablename__ = 'boy2girl'
id = Column(Integer, primary_key=True, autoincrement=True)
girl_id = Column(Integer, ForeignKey('girl.id'))
boy_id = Column(Integer, ForeignKey('boy.id'))
# boy = relationship('Boy', backref='boy')
class Girl(Base):
__tablename__ = 'girl'
id = Column(Integer, primary_key=True)
name = Column(String(64), unique=True, nullable=False)
def __str__(self):
return self.name
def __repr__(self):
return self.name
class Boy(Base):
__tablename__ = 'boy'
id = Column(Integer, primary_key=True, autoincrement=True)
name = Column(String(64), unique=True, nullable=False)
def __str__(self):
return self.name
def __repr__(self):
return self.name
# 与生成表结构无关，仅用于查询方便,放在哪个单表中都可以--等同于manytomany
# 方便快速查询，写了这个字段，相当于django 的manytomany，快速使用基于对象的跨表查询
girls = relationship('Girl', secondary='boy2girl', backref='boys')
if __name__ == '__main__':
# 创建引擎
engine = create_engine(
"mysql+pymysql://root:1234@127.0.0.1:3306/sqlalchemy02",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
# 把表模型同步到数据库中
Base.metadata.create_all(engine)
2.新增和基于对象的跨表查询
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from models import Girl, Boy, Boy2Girl
engine = create_engine(
"mysql+pymysql://root:1234@127.0.0.1:3306/sqlalchemy02",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
Session = sessionmaker(bind=engine)
session=Session()
'''多对多关系新增'''
1.手动操作第三张表，建立关系
# 增加Boy
boy = Boy(name='彭于晏')
boy1 = Boy(name='吴彦祖')
boy2 = Boy(name='金城武')
session.add(boy)
session.add_all([boy1, boy2])
# 增加Girl
girl = Girl(name='金玟庭')
girl1 = Girl(name='田振姬')
girl2 = Girl(name='赵美延')
session.add(girl)
session.add_all([girl1, girl2])
session.commit()
# 建立外键关系：手动操作第三张表
obj = Boy2Girl(boy_id=3, girl_id=1)
obj1 = Boy2Girl(boy_id=3, girl_id=2)
obj2 = Boy2Girl(boy_id=3, girl_id=3)
session.add_all([obj, obj1, obj2])
session.commit()
2.通过关联关系
girl = Girl(name='沈小婷')
girl1 = Girl(name='柳智敏')
boy = Boy(name='古天乐',girls=[girl,girl1]) # 可以多条
session.add(boy)
session.commit()
'''基于对象的跨表查询--正向'''
boy = session.query(Boy).filter_by(name='古天乐').first()
print(boy)
# 基于对象的跨表查询，正向查询
print(boy.girls) # 列表套对象
print(boy.girls[0])
for i in boy.girls: 循环取出对象
print(i.name)
'''基于对象的跨表查询--反向'''
girl = session.query(Girl).filter_by(name='赵美延').first()
print(girl)
# 基于对象的跨表查询，反向查询
print(girl.boys) # 列表套对象
五、基于Scoped_Session实现线程安全
session对象，集成到flask中去，要把session对象做成全局（大家公用），还是每个视图函数独有一个（每次都要实例化得到这个session对象）不是session_key，而是链接会话session
每个视图函数独有一个，每次都要实例化，
sqlalchemy提供了一种方式，让咱们使用全局的一个session，但是每个视图函数中使用的都是不同的request，session都是这种实现机制
sqlalchemy提供了一种，在不同线程中，虽然使用全局session，实际上每个线程自己独有一个session
1.线程安全
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
engine = create_engine("mysql+pymysql://root:1234@127.0.0.1:3306/sqlalchemy02", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
"""
session是链接对象，如果集成到flask中，我们是把session定义成全局还是每个视图函数一个session呢？
正常来讲要每个视图函数定义一个session，有些麻烦。
sqlalchemy 帮咱提供了一个只要定义一次的session，能够做到在不同线程中，使用的是自己的session，底层基于local。
"""
1.原来获得 session对象
# session=Session() # 放在flask中，这样如果做成全局，线程就不安全，想要安全就得在视图函数中每次都实例化得到
# print(type(session))
from sqlalchemy.orm import scoped_session
from threading import local
from models import User
2.使用scoped_session得到session对象，保证线程安全，本意就是每个线程使用自己独立的session
session=scoped_session(Session)
print(type(session))
############# 执行ORM操作 #############
obj = User(name='jack',email='1@qq.com')
session.add(obj)
# 提交事务
session.commit()
# # 关闭session
session.close()
2.研究
'''
研究1：session对象，现在不是原来的session对象了
session=Session() <class 'sqlalchemy.orm.session.Session'>
session=scoped_session(Session) <class 'sqlalchemy.orm.scoping.scoped_session'>
'''
'''
研究2：类上加装饰器，能够给对象，增加属性或方法
'''
# 加个showName方法和name属性
def showName():
print('xxxxx')
def add(func):
def inner(*args,**kwargs):
res = func(*args,**kwargs) # Person()
res.name = 'jack'
res.showName=showName
return res
return inner
@add # Person=add(Person)
class Person:
pass
p = Person()
print(p.name) # jack
p.showName()
'''
研究3：如何保证线程安全
补充from threading import local，local是一个类，类加括号实例对象
l=local() # 而local对象线程是安全的，不需要加锁
这个l，可以被多个线程并发操作， 而且不会数据错乱
如何实现的？内部通过线程id号作区分，取值赋值用的都是自己的那部分，例如
线程1：l.name='jack'
线程2：l.name='tom'
# 线程1上取name，永远取出来的是线程1当时放入的值，线程二也是一样
session：scoped_session独享
session.add------》return self._proxied.add()
self._proxied----》return self.registry()
self.registry = ThreadLocalRegistry(session_factory)
ThreadLocalRegistry(session_factory)--》这个对象加括号，会触发它的__call__
def __call__(self) -> _T:
try:
# 这个register就是self.registry = threading.local()，所以它是线程安全的
# 所以不同线程都是使用自己的value，也就是session对象
return self.registry.value # 不同线程有自己的一个session对象
except AttributeError:
val = self.registry.value = self.createfunc() # 所以这个就是Session()
return val
'''
六、基本增删查改和高级查询
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm import scoped_session
from models import Boy, Girl, Boy2Girl, Hobby, User, Person
from sqlalchemy.sql import text
engine = create_engine("mysql+pymysql://root:1234@127.0.0.1:3306/sqlalchemy02", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = scoped_session(Session) # 线程安全
'''1.基本增加：add、add_all'''
# add_all
hobby = Hobby(caption='羽毛球')
hobby1 = Hobby(caption='橄榄球')
session.add_all([hobby,hobby1])
# add
person = Person(name='彭于晏',hobby=hobby)
session.add(person)
session.commit()
'''2.删除'''
# 方式一
session.query(User).filter_by(id=1).delete()
# 方式二: session.delete(对象)
user = session.query(User).filter_by(id=1).first()
session.delete(user)
'''3.修改，更新'''
# 方式一
session.query(User).filter_by(id=1).update({'name':'oscar'})
session.commit()
# 方式二：类名.属性名，作为要修改的key
session.query(Boy).filter_by(id=4).update({Boy.name:'xxxx'})
session.commit()
# 方式三：
# 对象.name='xxx' session.add(对象)
boy=session.query(Boy).filter_by(id=1).first()
boy.name='xxl'
session.add(boy) # 有id就是修改，没有就是新增
session.commit()
# 类似于django的F查询
# 当字符串拼接
session.query(User).filter_by(id=1).update({'name':User.name+'_nb'},synchronize_session=False)
# 当字符串相加
session.query(User).filter_by(id=1).update({'id':User.id+6}, synchronize_session="evaluate")
session.commit()
'''4.查询--->基本查询'''
# 1.filter_by 写条件
res=session.query(User).filter_by(name='jack',id=1).first()
res=session.query(User).filter_by(name='jack').all() # 放在列表中 不是queryset对象
print(res)
# 2.filter 写表达式
res=session.query(User).filter(User.name=='jack').first()
res=session.query(User).filter(User.id>=3).all()
res=session.query(User).filter(User.name!='xxxx').all()
print(res)
# 3.只查表中某几个字段，并重命名
# select name as xx,age from user;
res=session.query(User.name.label('xx'), User.email)
# select name,email from user;
res=session.query(User.name, User.email).all()
res=session.query(User)
# 4.条件可以使用text自己拼凑 占位符
# select * from user where id< 224 and name=jack order by id
res = session.query(User).filter(text("id<:value and name=:name")).params(value=224, name='jack').order_by(User.id).all()
print(res)
# 5.直接写原生sql
# SELECT * FROM users where name=jack
res = session.query(User).from_statement(text("SELECT * FROM users where name=:name")).params(name='jack').all()
res = session.query(User).from_statement(text("SELECT * FROM users where name=:name")).params(name='张三')
print(res)
print('=======================================')
'高级查询'
'''表达式，and条件连接'''
res = session.query(User).filter(User.id > 1, User.name == 'jack').all() # and条件
'''between'''
# select * from User where user.id between 1 and 3 and name=jack
res = session.query(User).filter(User.id.between(1, 3), User.name == 'jack').all()
'''in 条件'''
res = session.query(User).filter(User.id.in_([1, 3, 4])).all()
res = session.query(User).filter(User.email.in_(['1@qq.com', '3@qq.com'])).all()
'''~非，除。。外'''
res = session.query(User).filter(~User.id.in_([1, 3, 4])).all()
'''二次筛选'''
# select * from User where id not in (select id from User where name=jack)
res = session.query(User).filter(~User.id.in_(session.query(User.id).filter_by(name='jack'))).all()
'''and or条件'''
from sqlalchemy import and_, or_
# or_包裹的都是or条件，and_包裹的都是and条件
res = session.query(User).filter(and_(User.id >= 3, User.name == 'jack')).all() # and条件
# 默认就是and条件，结果和上面一样
res = session.query(User).filter(User.id >= 3,User.name=='jack').all()
# select * from User where id < 2 or name = jack;
res = session.query(User).filter(or_(User.id < 2, User.name == 'jack')).all() # or条件
res = session.query(User).filter(
or_(
User.id < 2,
and_(User.name == 'jack', User.id > 3),
User.extra != ""
)).all()
'''通配符，模糊查询'''
# select * from User where email like %@%;
res = session.query(User).filter(User.email.like('%@%')).all()
res = session.query(User.id,User.name).filter(~User.name.like('j%')).all() # 不以什么开头
'''限制，用于分页，区间'''
res = session.query(User)[2 * 5:2 * 5 + 2] # 切片可以做运算
res = session.query(User)[2:4] # 从零开始，顾头不顾尾
'''排序，根据email降序排列（从大到小）'''
res = session.query(User).order_by(User.email.desc()).all()
res = session.query(User).order_by(User.email.asc()).all()
# 第一个条件重复后，再按第二个条件升序排
res = session.query(User).order_by(User.name.desc(), User.id.asc()).all()
'''分组查询'''
from sqlalchemy.sql import func
# select name from User group by name; 一旦分组后，就只能查询 分组后的字段和聚合函数的字段
res = session.query(User.name).group_by(User.name).all()
# 分组之后取最大id，id之和，最小id
res = session.query(
func.max(User.id), # 最大
func.sum(User.id), # 总和
func.min(User.id), # 最小
func.count(User.id), # 数量
User.name).group_by(User.name).all()
for item in res: # 循环拿出列表里面的每一个元组
print(item)
'''having筛选'''
# select max(id),sum(id),min(id) from User group by name having max(id)>2;
res = session.query(
func.max(User.id),
func.sum(User.id),
func.min(User.id)).group_by(User.name).having(func.max(User.id) > 2).all()
'''链表操作'''
# select * from person,hobby where person.hobby_id=hobby.id; # 原生sql
res = session.query(Person, Hobby).filter(Person.hobby_id == Hobby.id).all()
# join表，默认是inner join,自动按外键关联 内连接
# select * from Person inner Hobby on Person.hobby_id=Hobby.id; # 原生sql
res = session.query(Person).join(Hobby).all()
# isouter=True 外连，表示Person left join Favor，没有右连接，反过来即可
# select * from Person left Hobby on Person.hobby_id=Hobby.id; # 原生sql
res = session.query(Person).join(Hobby, isouter=True).all()
res = session.query(Hobby).join(Person, isouter=True).all() # 反过来顺序就是右链接
# 自己指定on条件（连表条件）,第二个参数，支持on多个条件，用and_,同上
res = session.query(Person).join(Hobby, Person.id == Hobby.id, isouter=True) # sql本身有问题，只是给你讲， 自己指定链接字段
'''union和union all'''
# 组合（了解）UNION 操作符用于合并两个或多个 SELECT 语句的结果集
q1 = session.query(Boy.id,Boy.name).filter(Boy.id > 1)
q2 = session.query(Girl.id,Girl.name).filter(Girl.id < 5)
# union 去重
res = q1.union(q2).all()
print(res)
print(len(res))
q1 = session.query(Boy.name).filter(Boy.id > 1)
q2 = session.query(Girl.name).filter(Girl.id < 5)
res = q1.union_all(q2).all()
print(res)
print(len(res))
'''一对多，基于链表跨表查（__链表）'''
# 方式一：直接连
res = session.query(Person, Hobby).filter(Person.hobby_id == Hobby.id, Hobby.id >= 2).all()
# 方式二：join连
res = session.query(Person).join(Hobby).filter(Person.id >= 2).all()
'''多对多关系，基于链表的跨表查'''
# 方式一：直接连
res = session.query(Boy, Girl, Boy2Girl).filter(Boy.id == Boy2Girl.boy_id, Girl.id == Boy2Girl.girl_id).all()
# 方式二：join连
res = session.query(Boy).join(Boy2Girl).join(Girl).filter(Person.id >= 2).all()
print(res)
session.commit()
session.close()
七、Flask-SQLAlchemy的使用
1.flask中集成sqlalchemy使用(自己做)
'创建一个flask项目'
'''
项目结构
src
-__init__.py
-db.py
-models.py
-settings.py
-views.py
templates
-index.html
app.py
'''
'app.py'
from src import app
if __name__ == '__main__':
app.run()
'settings.py'
DEBUG=True
'__init__.py'
from flask import Flask
app = Flask(__name__, template_folder='../templates')
app.config.from_pyfile('./settings.py')
# 注册蓝图
from src.views import bp_user
app.register_blueprint(bp_user)
'db.py'
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm import scoped_session
from sqlalchemy.sql import text
engine = create_engine("mysql+pymysql://root:1234@127.0.0.1:3306/sqlalchemy02", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = scoped_session(Session) # 线程安全
'models.py'
import datetime
from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base, relationship
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index
Base = declarative_base() # Base 当成 models.Model
# 单表
class User(Base):
__tablename__ = 'users' # 表名
# 写字段
id = Column(Integer, primary_key=True, autoincrement=True) # id 主键
name = Column(String(32), index=True, nullable=False) # name列，索引，不可为空
email = Column(String(32), unique=True)
# datetime.datetime.now不能加括号，加了括号，以后永远是当前时间
ctime = Column(DateTime, default=datetime.datetime.now)
extra = Column(Text)
def __str__(self):
return self.name
def __repr__(self):
return self.name
if __name__ == '__main__':
# 创建引擎
engine = create_engine(
"mysql+pymysql://root:1234@127.0.0.1:3306/sqlalchemy02",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
# 把表模型同步到数据库中
Base.metadata.create_all(engine)
'views.py'
from flask import render_template,jsonify
from flask import Blueprint
bp_user = Blueprint('user',__name__)
from .db import session
from .models import User
@bp_user.route('/')
def index():
# 把用户表中所有数据都查出来，返回给前端
res = session.query(User).all()
'''前后端分离传入数据'''
# 不能直接放入jsonify中，json序列化，只能序列化 数字、字符串、列表、布尔和字典
# res是列表套对象，是不能直接序列化的
# l = []
# for i in res:
# l.append({'name': i.name, 'email': i.email})
# return jsonify({'code': 100, 'msg': '成功', 'results': l})
'''前后端混合传入数据'''
return render_template('index.html',users=res)
'index.html'
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Title</title>
</head>
<body>
<h1>hello world</h1>
{% for user in users %}
<p>用户名：{{user.name}}----->邮箱：{{user.email}}</p>
{% endfor %}
</body>
</html>
2.flask中集成flask-sqlalchemy使用
# 第三方: flask-sqlalchemy 封装了用起来，更简洁
安装：pip install flask-sqlalchemy
'''借助于flask-sqlalchemy'''
1 导入 from flask_sqlalchemy import SQLAlchemy
2 实例化得到对象
db = SQLAlchemy()
3 将db注册到app中
db.init_app(app)
------2,3 可以合并为db = SQLAlchemy(app)--------
4 视图函数中使用session
全局的db.session # 线程安全的
5 models.py 中继承Model
db.Model
6 写字段
username = db.Column(db.String(80), unique=True, nullable=False)
7 配置文件中加入
SQLALCHEMY_DATABASE_URI = "mysql+pymysql://root@127.0.0.1:3306/ddd?charset=utf8"
SQLALCHEMY_POOL_SIZE = 5
SQLALCHEMY_POOL_TIMEOUT = 30
SQLALCHEMY_POOL_RECYCLE = -1
# 追踪对象的修改并且发送信号
SQLALCHEMY_TRACK_MODIFICATIONS = False
八、Flask-Migrate的使用
1.数据库肯定要自己创建
2.创建表，增加删除字段，需要手动做处理，而django 有两个命令自动做
python manage.py makemigrations # 记录变化
python manage.py migrate #把变化同步到数据库
-那有没有种方案，跟djagno一样，自动记录，自动迁移？
3.第三方模块：flask-migrate，就能完成跟django一样，只是命令稍微有些不同
# https://github.com/miguelgrinberg/Flask-Migrate/
安装：pip install Flask-Migrate --upgrade
4.使用步骤
-导入
from flask_migrate import Migrate
-注册
'需要先导入app和db才行'
migrate = Migrate(app, db) # flask就会多出几个命令
'''
flask --app manage:app db init # 初始化，第一次执行，以后再也不执行了，它执行完，会出现一个migrations文件夹
flask --app manage:app db migrate # django中的makemigrations 是一模一样
flask --app manage:app db upgrade # 跟django的migrate一样
如果启动文件就是叫app则可以省略--app manage:app，直接flask db migrate这样使用即可，其他命令也是
'''
flask中还有和django中的模块类似的模块，如cors、cache、restful等等,具体可自行查找了解