flask项目演示
打开一个flask项目
运行步骤
1. pycharm 打开项目
2. 安装依赖：虚拟环境用 pip install -r req.txt
3. 数据库建库：movie # root用户密码不是123需要改两个地方的代码，__init__.py和models.py文件。
4. 打开models.py 解开注释，最上面两行注释掉然后以脚本文件运行。
cd到models.py所在的文件执行命令行： python model.py # 把表创建然后插入一个超级管理员。
5. 回到项目根路径：python manage.py runserver
6. 后台：/admin 前台：/ 》》》路由
SQLAlchemy介绍和快速使用
sqlalchemy类似于一种orm框架
* django-orm：只能给django用的不能独立使用。
* sqlalchemy：可以独立使用，集成到web项目中
* peewee：？
* tortoise-orm：异步orm框架
安装：pip3 install sqlalchemy
组成部分
Engine：框架的引擎
Connection Pooling：数据库链接池
Dialect：选择链接数据库的DB API种类：mysql，sqllite...
Schema/Types：架构和类型
SQL Exprression Language、SQL表达式语言
能够操作的关系型数据库
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
原生操作的快速使用
使用步骤：
# 第一步：导入包
from threading import Thread
import sqlachemy
from sqlachemy import create_engine
from sqlachemy.engine.base import Engine
# 第二步：实例化得到一个engine
engine = create_engine(
"mysql+pymysql://root:[email protected]:3306/luffy?charset=utf8",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有，线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
# 第三步：通过engine拿到一个链接
# conn对象是从链接池中取出一个链接
def task():
conn = engine.raw_connection()
cursor = conn.cursor()
cursor.execute("select * from luffy_banner")
print(cursor.fetchall())
# 第三步：多线程测试
for i in range(20):
t = Thread(target=task)
t.start()
创建操作数据表
通过类 创建和删除表
第一步：导入一些依赖
第二步：创建成一个Base：Base = declarative_base()
第三步：写类都集成Base
class User(Base):
第四步：写字段。字段都是Column类的对象，通过参数控制字段类型是否可以为空，是否加索引。
id = Column(Integer, primary_key=True) # id 主键
name = Column(String(32), index=True, nullable=False)
第五步：定义表名、联合唯一、联合索引
__tablename__ = 'users' # 数据库表名称
# 定义联合索引，联合唯一
__table_args__ = (
UniqueConstraint('id', 'name', name='uix_id_name'), # 联合唯一
Index('ix_id_name', 'name', 'email'), # 联合索引
)
第六步：把被Base管理的所有表同步到数据库中【不能创建数据库也不能删除修改字段】
engine = create_engine(
"mysql+pymysql://root:[email protected]:3306/aaa?charset=utf8",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
Base.metadata.create_all(engine)
第七步：删除被Base管理的所有表
Base.metadata.drop_all(engine)
# 由元类实例化得到一个父类，然后继承这个父类再写表字段。
# 第一步：导入一些依赖
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index
import datetime
from sqlalchemy.orm import relationship
from sqlalchemy import create_engine
# 第二步：由元类创建一个父类
Base = declarative_base()
# 第三步：写模型类继承这个父类
class User(Base):
# 第四步：写字段所有字段都是Column的对象。用参数去控制类型
id = Column(Integer, primary_key=True) # id主键
name = Column(String(32), index=True, nullable=False) # varchar32 name列、索引不可为空
email = Column(String(32), unique=True) # 唯一
# datetime.datetime.now 不能加括号，加括号以后就是当前创建user类的时间数据就会写死。
ctime = Column(DateTime, default=datetime.datetime.now)
extra = Column(Text, nullable=True)
# 定义表名字
__tablename__ = 'users' # 数据库表名称
# 定义联合索引、联合唯一
__table_args__ = (
UniqueConstraint('id', 'name', name='uix_id_name'), # 联合唯一
Index('ix_id_name', 'name', 'email') # 联合索引
)
class Book(Base):
__tablename__ = 'books' # 数据库表名称
id = Column(Integer, primary_key=True) # id主键
name = Column(String(32), index=True, nullable=False) # varchar32 name列、索引不可为空
price = Column(Integer)
def __repr__(self):
return str(self.price)
class Publish(Base):
__tablename__ = 'publish'
id = Column(Integer, primary_key=True) # id主键
name = Column(String(32), nullable=False) # varchar32 name列、索引不可为空
# 第五步：sqlalchemy没有表迁移；只能创建出被Base管理所有表和删除所有的表。不能只删除其中一个表或者增加、修改、删除表字段。
### 一对多关系：一个爱好对应多人喜欢
class Hobby(Base):
__tablename__ = 'hobby'
id = Column(Integer, primary_key=True) # id主键
caption = Column(String(50), default='篮球')
class Person(Base):
__tablename__ = 'person'
id = Column(Integer, primary_key=True) # id主键
name = Column(String(32), index=True, nullable=True)
# hobby 指的是tablename而不是类名
# 关联字段写在多的一方，写在Person中，跟hobby表中id字段做外键关联
hobby_id = Column(Integer, ForeignKey("hobby.id"))
# 跟数据库无关不会新增字段只用于快速链表操作
# 基于对象的跨表查询：就要加这个字段取对象 person.hobby person.hobby_id
# backref用于反向查询的字段。
hobby = relationship('Hobby', backref='pers') # 如果有hobby对象要拿到所有人则需要hobby.pers 是一个列表
def __repr__(self):
return self.name
### 多对多关系
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'))
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)
# 与生成表结构无关，仅用于查询方便，放在哪个单表中都是可以的
# 方便于快速查询， 这个字段相当于django的manytomany能快速使用基于对象的跨表查询
girls = relationship('Girl', secondary='boy2girl', backref='boys')
def __str__(self):
return self.name
def __repr__(self):
return self.name
# sqlalchemy 不能创建数据库不能修改和删除字段
def init_db():
engine = create_engine(
"mysql+pymysql://root:[email protected]:3306/aaa?charset=utf8",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
# 创建出所有被Base管理的表(仅限于继承Base的模型类)
Base.metadata.create_all(engine)
def drop_db():
engine = create_engine(
"mysql+pymysql://root:[email protected]:3306/aaa?charset=utf8",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
# 删除所有被Base管理的表
Base.metadata.drop_all(engine)
if __name__ == '__main__':
init_db()
# drop_db()
sqlalchemy快速插入数据
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from models import User, Book
# 第一步：创建engine
engine = create_engine("mysql+pymysql://root:[email protected]:3306/aaa", max_overflow=0, pool_size=5)
# 第二步：通过engine获取session对象，这里的session和之前的cookie的session不是一个东西
Session = sessionmaker(bind=engine)
# 每次执行数据库操作时都需要创建一个Connection
session = Session()
# 第三步：通过session操作插入数据
# book = Book(name='西游记', price=23) # 直接从book表创建
user = User(name='xz', email='[email protected]', extra='好帅！')
# 这里是用add的方法创建
# session.add(book)
session.add(user)
session.commit()
session.close()
scoped_session线程安全
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from models import User, Book
# 第一步：创建engine
engine = create_engine("mysql+pymysql://root:[email protected]:3306/aaa", max_overflow=0, pool_size=5)
# 第二步：通过engine，获得session对象
Session = sessionmaker(bind=engine)
# 第三步：导入线程
from sqlalchemy.orm import scoped_session
from threading import Thread
session = scoped_session(Session)
def task(i):
user = User(name='肖战%s' % i, email='%[email protected]' % i, extra='大帅哥')
session.add(user)
session.commit()
session.close()
for i in range(50):
t = Thread(target=task, args=[i, ])
t.start()
注意
* session 是链接对象，如果集成到flask中，我们是吧session定义成全局。如果每个视图函数都定义一个session是有些麻烦的。
* sqlalchemy 帮助我们只需要定义一次session。可以实现能够在不同线程中使用自己的session，这种底层是基于local。
* 因为session = Session() 不是线程安全的，如果我们需要做到线程安全就需要在在scoped_session(Seesion) 获得session对象。
session = scoped_session(Session)
类装饰器
加在类上的装饰器
# 类装饰器：加在类上的装饰器
def auth(func):
def inner(*args, **kwargs):
print("我要开始了")
res = func(*args, **kwargs)
res.name = 'xcc'
return res
return inner
@auth # Person = auth(Person) # 加在类上的装饰器
class Person():
pass
# 类实例化对象可调用装饰器中的数据
p = Person() # 我要开始了
print(p.name) # xcc
装饰器是类
# 类装饰器：装饰器是类
class Auth():
def __init__(self, func):
self.func = func
def __call__(self, a, b):
print('我要开始了 ')
res = self.func(a, b)
return res
@Auth # add=Auth(add) 类实例化对象会触发Auth的 __init__
def add(a, b): # add 是Auth的对象
return a + b
res = add(4, 5) # add 加括号会触发Auth的__call__
print(res) # 9
一对多
表模型
# 由元类实例化得到一个父类，然后继承这个父类再写表字段。
# 第一步：导入一些依赖
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index
import datetime
from sqlalchemy.orm import relationship
from sqlalchemy import create_engine
# 第二步：由元类创建一个父类
Base = declarative_base()
# 第三步：写模型类继承这个父类
class User(Base):
# 第四步：写字段所有字段都是Column的对象。用参数去控制类型
id = Column(Integer, primary_key=True) # id主键
name = Column(String(32), index=True, nullable=False) # varchar32 name列、索引不可为空
email = Column(String(32), unique=True) # 唯一
# datetime.datetime.now 不能加括号，加括号以后就是当前创建user类的时间数据就会写死。
ctime = Column(DateTime, default=datetime.datetime.now)
extra = Column(Text, nullable=True)
# 定义表名字
__tablename__ = 'users' # 数据库表名称
# 定义联合索引、联合唯一
__table_args__ = (
UniqueConstraint('id', 'name', name='uix_id_name'), # 联合唯一
Index('ix_id_name', 'name', 'email') # 联合索引
)
class Book(Base):
__tablename__ = 'books' # 数据库表名称
id = Column(Integer, primary_key=True) # id主键
name = Column(String(32), index=True, nullable=False) # varchar32 name列、索引不可为空
price = Column(Integer)
def __repr__(self):
return str(self.price)
class Publish(Base):
__tablename__ = 'publish'
id = Column(Integer, primary_key=True) # id主键
name = Column(String(32), nullable=False) # varchar32 name列、索引不可为空
# 第五步：sqlalchemy没有表迁移；只能创建出被Base管理所有表和删除所有的表。不能只删除其中一个表或者增加、修改、删除表字段。
### 一对多关系：一个爱好对应多人喜欢
class Hobby(Base):
__tablename__ = 'hobby'
id = Column(Integer, primary_key=True) # id主键
caption = Column(String(50), default='篮球')
class Person(Base):
__tablename__ = 'person'
id = Column(Integer, primary_key=True) # id主键
name = Column(String(32), index=True, nullable=True)
# hobby 指的是tablename而不是类名
# 关联字段写在多的一方，写在Person中，跟hobby表中id字段做外键关联
hobby_id = Column(Integer, ForeignKey("hobby.id"))
# 跟数据库无关不会新增字段只用于快速链表操作
# 基于对象的跨表查询：就要加这个字段取对象 person.hobby person.hobby_id
# backref用于反向查询的字段。
hobby = relationship('Hobby', backref='pers') # 如果有hobby对象要拿到所有人则需要hobby.pers 是一个列表
def __repr__(self):
return self.name
### 多对多关系
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'))
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)
# 与生成表结构无关，仅用于查询方便，放在哪个单表中都是可以的
# 方便于快速查询， 这个字段相当于django的manytomany能快速使用基于对象的跨表查询
girls = relationship('Girl', secondary='boy2girl', backref='boys')
def __str__(self):
return self.name
def __repr__(self):
return self.name
# sqlalchemy 不能创建数据库不能修改和删除字段
def init_db():
engine = create_engine(
"mysql+pymysql://root:[email protected]:3306/aaa?charset=utf8",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
# 创建出所有被Base管理的表(仅限于继承Base的模型类)
Base.metadata.create_all(engine)
def drop_db():
engine = create_engine(
"mysql+pymysql://root:[email protected]:3306/aaa?charset=utf8",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
# 删除所有被Base管理的表
Base.metadata.drop_all(engine)
if __name__ == '__main__':
init_db()
# drop_db()
0、获取一个session对象
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from models import Person, Hobby
from sqlalchemy.orm import scoped_session
engine = create_engine("mysql+pymysql://root:[email protected]:3306/aaa", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = scoped_session(Session)
1、一对多关系新增
hobby = Hobby(caption='足球')
# 把hobby存入
session.add(hobby)
print(hobby.id)
### 两种方案
## 方案一：直接新增表关系
person = Person(name='wyb', hobby_id=1)
session.add(person)
## 方案二：创建对象新增
person = Person(name='xz', hobby=Hobby(caption='羽毛球'))
session.add(person)
## 方案三：将查询出来的结果作为新增数据。
hobby = session.query(Hobby).filter_by(id=2).first()
print(hobby)
person = Person(name='lss', hobby=hobby)
person = Person(name='lh', hobby_id=hobby.id)
session.add(person)
session.commit()
session.close()
2、基于对象的跨表查询
## 基于对象的跨表查询的正向查询
person = session.query(Person).filter_by(id=1).first()
print(person.hobby_id)
print(person.hobby.caption)
## 基于对象的跨表查询的反向
hobby = session.query(Hobby).filter_by(id=2).first()
# 可以查询到该喜好的所有人以列表形式
print(hobby.pers) # [lss, lh, lss, lh]
for p in hobby.pers:
print(p.name)
session.commit()
session.close()
多对多
0、获取一个session对象
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from models import Person, Hobby
from sqlalchemy.orm import scoped_session
engine = create_engine("mysql+pymysql://root:[email protected]:3306/aaa", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = scoped_session(Session)
1、多对多新增
## 手动操作第三张表的方式
boy = Boy(name='鹿晗')
girl = Girl(name='刘诗诗')
session.add(boy)
session.add(girl)
session.add_all([boy, girl])
## 建立关系：手动操作第三张表
b = Boy2Girl(girl_id=1, boy_id=1)
session.add(b)
### 通过关联关系
session.add(Boy(name='肖战', girls=[Girl(name='小红'), Girl(name='小绿')]))
## 上面语法可以岔开
girl1 = Girl(name='小红1')
girl2 = Girl(name='小绿1')
boy = Boy(name='肖战1', girls=[girl1, girl2])
session.add(boy)
session.commit()
session.close()
2、基于对象的跨表查询
### 基于对象的跨表查询
## 正向
boy = session.query(Boy).filter_by(id=2).first()
print(boy)
# 这个男生匹配的所有女生
print(boy.girls)
## 反向
girl = session.query(Girl).filter_by(id=2).first()
print(girl)
print(girl.boys)
session.commit()
session.close()
基本增删改查
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from models import Boy, Girl, Boy2Girl, Hobby, Book, User, Person
from sqlalchemy.orm import scoped_session
from sqlalchemy.sql import text
engine = create_engine("mysql+pymysql://root:[email protected]:3306/aaa", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = scoped_session(Session)
## 1 基本增 add add_all
hobby = Hobby(caption='羽毛球')
book = Book(name='三国演义', price=22)
session.add(hobby)
session.add_all([hobby, book])
## 2 删除--->查询再删---》一般写项目都是软删除
hobby = session.query(Hobby).filter_by(caption='足球').delete()
print(hobby)
## 3 修改、更新 update F查询
res = session.query(User).filter(User.id > 0).update({"name": "lqz"})
# 类似于django的F查询 =》》当字符串相加
res = session.query(User).filter(User.id > 0).update({User.name: User.name + "099"}, synchronize_session=False)
# 数字相加
res = session.query(Book).filter(Book.price > 20).update({"price": Book.price + 1}, synchronize_session="evaluate")
print(res)
### 4 查询
## 查询所有 是list对象
res = session.query(User).all()
print(type(res))
print(len(res))
## 只查询某几个字段
# select name as xx, email from user;
res = session.query(User.name.label('xx'), User.email)
print(res) # 打出原生sql
print(res.all())
for item in res.all():
print(item[0])
## 查询： filer:写条件 filter_by：等于的值
res = session.query(User).filter(User.name == "lqz").all()
res = session.query(User).filter(User.name != "lqz").all()
res = session.query(User).filter(User.name != "lqz", User.email == '[email protected]').all()
res = session.query(User).filter_by(name='lqz099').all()
res = session.query(User).filter_by(name='lqz099', email='[email protected]').all()
print(len(res))
## 取一个 all了后是list，list 没有first方法
res = session.query(User).first()
print(res)
## 查询所有，使用占位符(了解) :value :name
res = session.query(User).filter(text("id:value and name=:name")).params(value=10, name='xcc022').order_by(
User.id).all()
## 自定义查询（了解）
res = session.query(User).from_statement(text("SELECT * FROM users where email=:email")).params(email='[email protected]').all()
print(res)
## 高级查询
# 表达式，and条件连接
res = session.query(User).filter(User.id > 1, User.name == 'lss').all()
# between 查询
res = session.query(User).filter(User.id.between(1, 3), User.name == 'lss').all()
# in 查询
res = session.query(User).filter(User.id.in_([1, 3, 4])).all()
res = session.query(User).filter(User.email.in_(['[email protected]', '[email protected]'])).all()
# ~非，除。。外
res = session.query(User).filter(~User.id.in_([1, 3, 4])).all()
# 二次筛选
res = session.query(User).filter(~User.id.in_(session.query(User.id).filter_by(name='lqz099'))).all()
# and or条件
from sqlalchemy import and_, or_
# or_包裹的都是or条件，and_包裹的都是and条件
res = session.query(User).filter(and_(User.id >= 3, User.name == 'lqz099')).all() # and条件
res = session.query(User).filter(or_(User.id < 2, User.name == 'eric')).all()
res = session.query(User).filter(
or_(
User.id < 2,
and_(User.name == 'lqz099', User.id > 3),
User.extra != ""
)).all()
## 通配符，以e开头，不以e开头
res = session.query(User).filter(User.email.like('%@%')).all()
res = session.query(User.id).filter(~User.name.like('e%'))
## 分页 # 一页2条，查第5页
# res = session.query(User)[2*5:2*5+2]
## 排序，根据条件降序排列（从大到小）、升序
res = session.query(User).order_by(User.email.desc()).all()
res = session.query(Book).order_by(Book.price.desc()).all()
res = session.query(Book).order_by(Book.price.asc()).all()
# 第一个条件重复后，再按第二个条件升序排
ret = session.query(User).order_by(User.name.desc(), User.id.asc()).all()
## 分组查询 group_by func
from sqlalchemy.sql import func
ret = session.query(User).group_by(User.extra).all()
## 分组之后取最大id，id之和，最小id
res = session.query(
func.max(User.id),
func.sum(User.id),
func.min(User.id)).group_by(User.extra).all()
print(res)
for item in res:
print(item[2])
# having 分组之后 group_by
ret = session.query(
func.max(User.id),
func.sum(User.id),
func.min(User.id)).group_by(User.extra).having(func.max(User.id) > 2).all()
## 链表操作
#select * from person,hobby where person.hobby_id=hobby.id;
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;
res = session.query(Person).join(Hobby).all()
# #isouter=True 外连，表示Person left join Favor，没有右连接，反过来即可
# select * from Person left Hobby on Person.hobby_id=Hobby.id;
res = session.query(Person).join(Hobby, isouter=True).all()
# 自己指定on条件（连表条件）,第二个参数，支持on多个条件，用and_,同上
res = session.query(Person).join(Hobby, Person.id == Hobby.id, isouter=True) # sql本身有问题，只是给你讲， 自己指定链接字段
# 右链接
res = session.query(Hobby).join(Person, isouter=True)
## 组合（了解）UNION 操作符用于合并两个或多个 SELECT 语句的结果集
# union和union all的区别？ union 并集（去重） union all 合集（不去重）
q1 = session.query(User).filter(User.id > 40)
q2 = session.query(User).filter(User.id > 38)
res = q1.union(q2).all()
q1 = session.query(User.email).filter(User.id > 40)
q2 = session.query(User.email).filter(User.id > 38)
res = q1.union_all(q2).all()
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()
session.commit()
session.close()
支持原生sql
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from models import Boy, Girl, Boy2Girl, Hobby, Book, User, Person
from sqlalchemy.orm import scoped_session
from sqlalchemy.sql import text
engine = create_engine("mysql+pymysql://root:[email protected]:3306/aaa", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = scoped_session(Session)
# 查询
cursor = session.execute('select * from users')
result = cursor.fetchall()
# 添加
cursor = session.execute('insert into users(name) values(:value)',params={"value":'lqz'})
session.commit()
print(cursor.lastrowid)
print(result)
session.commit()
session.close()
flask-sqlalchemy使用和flask-migrate使用
flask中使用sqlalchemy
from flask import Flask
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from models import Boy, Girl, Boy2Girl, Hobby, Book, User, Person
from sqlalchemy.orm import scoped_session
app = Flask(__name__)
engine = create_engine("mysql+pymysql://root:[email protected]:3306/aaa", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = scoped_session(Session)
@app.route('/index')
def index():
session.add(Book(name='西游记',price=34))
session.commit()
return 'index'
if __name__ == '__main__':
app.run()
flask-sqlalchemy使用
1. 导入 from falsk_sqlalchemy import SQLALchemy
2. 实例化得到对象
db = SQLALchemy()
3. 将db注册到app中
db.init_app(app)
4. 视图函数中使用session
全局的db.session # 线程安全的
5. models.py 中继承 Base
db.Base
6. 写字段
username = db.Column(db.String(80), unique=True, nullable=False)
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
db = SQLAlchemy()
db.init_app(app)
class User(db.Base):
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(80), unique=True, nullable=False)
@app.route('/index')
def index():
db.session.add()
return 'index'
if __name__ == '__main__':
app.run()
flask-migrate使用
python manage.py makemigrations # 记录变化
python manage.py migrate #把变化同步到数据库
# 使用步骤：
1 导入
from flask_script import Manager
from flask_migrate import Migrate, MigrateCommand
2 注册
manager = Manager(app)
# 使用flask_migrate的Migrate 包裹一下app和db（sqlalchemy对象）
Migrate(app, db)
3 给flask_script增加一个db命令
# 把命令增加到flask-script中去
manager.add_command('db', MigrateCommand)
4 出现3条命令
python manage.py db init # 只执行一次，做初始化操作，以后再也不执行了，多出一个migrations文件夹
python manage.py db migrate #等同于django 的makemigrations
python manage.py db upgrade #等同于django 的migrate