热门标签
热门文章
- 1idea远程连接Docker调试 debug_idea docker 调试
- 2无人机集群作战仿真研究现状及在城市作战中的应用前景（附最新无人集群项目内容）_研究无人机作战的必要性和可行性
- 3【2023年电工杯数学建模竞赛B题人工智能对大学生学习影响的评价】完整思路分析+完整代码_人工智能对大学生学习影响的评价数学建模
- 4Idea之Java代码Remote JVM Debug_idea remote jvm debug
- 5【项目总结】基于SpringBoot+Ansj分词+正倒排索引的Java文档搜索引擎项目总结_springboot引入ansj统计词频
- 6【蓝桥杯计算思维题】少儿编程 蓝桥杯青少组计算思维真题及详细解析第5套_蓝桥杯小学组真题
- 7FPGA学习笔记_fpga16‘h什么意思
- 8计算机专业有哪些【含金量超高竞赛】？_计算机含金量高的比赛
- 9Devops系统化，从零开始学习容器技术（更新中）_系统学习容器化技术
- 10C++const成员函数/取地址运算符重载_const 版本的取地址运算符
当前位置: article > 正文
python学习之flask sqlalchemy介绍和使用 表操作增删查改_sqlalchemy增删改查
作者：AllinToyou | 2024-05-23 14:43:00
赞
踩
sqlalchemy增删改查
flask sqlalchemy介绍和使用 表操作增删查改
内容详细
1 sqlalchemy介绍和快速使用
1.1 介绍
# SQLAlchemy是一个基于Python实现的ORM框架 # django的orm框架---》只能再django中用，不能单独用 # SQLAlchemy单独的，可以集成到任意框架中 # peewee：轻量级 # python的异步orm框架不多， sanic， fastapi---》一旦用了异步，后续所有都需要用异步---》操作mysql，aiomysql--》操作redis，使用aioredis # 公司选择 -第一：peewee-async -第二：框架是异步---》没有使用异步orm框架---》SQLAlchemy---》生成和迁移表---》查询操作数据用原生操作# 写django项目---》库和表已经有了 -正常操作django中建表模型---》迁移---》表 -反向生成models--》表---》models.py----》改表---》再反向生成 python manage.py inspectdb > app/models.py
- 1
- 2
- 3
- 4
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
1.1 执行原生sql
# 执行原生sql快速使用import time import threading import sqlalchemy from sqlalchemy import create_engine from sqlalchemy.engine.base import Engine # 第一步：创建engine engine = create_engine( "mysql+pymysql://root:123@127.0.0.1:3306/lqz?charset=utf8", max_overflow=0, # 超过连接池大小外最多创建的连接 pool_size=5, # 连接池大小 pool_timeout=30, # 池中没有线程最多等待的时间，否则报错 pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置） ) # 第二步：使用 def task(): conn=engine.raw_connection() # 从连接池中取一个连接 cursor=conn.cursor() sql="select * from signer" cursor.execute(sql) print(cursor.fetchall()) if __name__ == '__main__': for i in range(20): t=threading.Thread(target=task) t.start() # 查询mysql的客户端连接数
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 1
- 2
2 单表操作增删查改
2.1 表迁移
# 不能创建数据库（django orm也不能） # 只能做表的创建和删除，不能做表更改（django orm能)---》借助于第三方实现###### # 第一步：生成基类，所有表模型都要继承这个基类 # django 的orm继承一个父类，Base就是那个父类 ###### 第二步：写表模型，继承父类，写字段 （注意区别于django 的orm） # django的default--》可不可以传个函数内存地址---》插入的时候通过函数运算完得到的值 ###### 第三步：迁移，通过表模型，生成表
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 1
- 2
import datetime from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index # 第一步：生成基类，所有表模型都要继承这个基类 # django 的orm继承一个父类，Base就是那个父类 Base = declarative_base() # 第二步：写表模型，继承父类，写字段 （注意区别于django 的orm） # django的default--》可不可以传个函数内存地址---》插入的时候通过函数运算完得到的值 class Users(Base): 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, nullable=True) # 大文本，可以为空 __tablename__ = 'lqz_users' # 数据库表名称,如果不写，就报错 __table_args__ = ( # UniqueConstraint('id', 'name', name='uix_id_name'), # 联合唯一 # Index('ix_id_name', 'name', 'email'), # 联合索引 ) # 聚簇索(mysql主键自动建索引，聚簇索引，mysql基于聚簇索引构建的B+树)，一定会有，没有显示建主键，mysql会隐藏一个 # 辅助索引：手动建的叫辅助索引---》单独减了索引---》如果你的辅助索引过多，非常影响插入效率，适度建索引
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
import datetime from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index # 第三步：迁移，通过表模型，生成表 from models import Base engine = create_engine( "mysql+pymysql://root:123@127.0.0.1:3306/db01?charset=utf8", max_overflow=0, # 超过连接池大小外最多创建的连接 pool_size=5, # 连接池大小 pool_timeout=30, # 池中没有线程最多等待的时间，否则报错 pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置） ) def create_table(): # 通过engine这个连接配置，创建出所有使用Base管理的表 Base.metadata.create_all(engine) def delete_table(): # 通过engine这个连接配置，删除出所有使用Base管理的表 Base.metadata.drop_all(engine) if __name__ == '__main__': # create_table() delete_table()
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
2.2 简单的表操作
# 操作表,增加一条记录，以后都用conn/session操作
# 第一步：创建engin
# 第二步：通过session得到连接对象
# Session = sessionmaker(bind=engine)
# session = Session()
# # 第三步：实例化得到模型类的对象，增加到数据库中
# usr=Users(name='lqz001')
# session.add(usr)
# # 第四步：提交事务
# session.commit()
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
2.3 基于scoped_session实现线程安全
# 以后操作数据，都用session对象---》定义在flask的函数外部还是内部？ # 放内部没问题，每次都生成一个新的session，耗费资源 # 如果定义在函数外部，会存在 多线程并发使用同一个变量session，要把session做成并发安全的 Session = sessionmaker(bind=engine) session = scoped_session(Session) # 也是基于local，给每一个线程自己创造一个session# 只需要记住，如果是多线程使用，或者在web框架中，使用scoped_session生成session就可以了 # 集成到flask中，有flask-sqlalchemy第三方，内部已经处理了scoped_session # 全局用这个一个session，不用担心并发不安全 usr=Users(name='lqz002') session.add(usr) # 线程一用：取local中取线程1的那个session，如果就给，没有就重新创造一个 # 第四步：提交事务 session.commit()
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 1
- 2
- 3
- 4
- 5
- 6
测试线程安全
# 线程一用：取local中取线程1的那个session，如果就给，没有就重新创造一个 # 线程二用：取local中取线程2的那个session，如果就给，没有就重新创造一个# 测试：开3个线程，如果定义全局的session，在3个线程中用，session对象应该是同一个 Session = sessionmaker(bind=engine) session = Session() # session = scoped_session(Session) def task(): # usr=Users(name='lqz003') # session.add(usr) # session.commit() # print(session.registry.registry.value) # <sqlalchemy.orm.scoping.scoped_session object at 0x7f8fbceeea60> print(session) # <sqlalchemy.orm.scoping.scoped_session object at 0x7f8fbceeea60> # 开3个线程，如果定义scoped_session，在3个线程中用，session对象应该是不是同一个，独有的 if __name__ == '__main__': for i in range(3): t=Thread(target=task) t.start()
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 1
- 2
- 3
3.4 基本增删查改
import datetime from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index from sqlalchemy.orm import sessionmaker from models import Users from sqlalchemy.orm import scoped_sessionfrom threading import Thread # 第三步：迁移，通过表模型，生成表 from models import Base from sqlalchemy.sql import text engine = create_engine( "mysql+pymysql://root:123@127.0.0.1:3306/db01?charset=utf8", max_overflow=0, # 超过连接池大小外最多创建的连接 pool_size=5, # 连接池大小 pool_timeout=30, # 池中没有线程最多等待的时间，否则报错 pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置） ) def create_table(): # 通过engine这个连接配置，创建出所有使用Base管理的表 Base.metadata.create_all(engine) def delete_table(): # 通过engine这个连接配置，删除出所有使用Base管理的表 Base.metadata.drop_all(engine) if __name__ == '__main__': # create_table() # delete_table() Session = sessionmaker(bind=engine) session = scoped_session(Session) #1 增加操作 # obj1 = Users(name="lqz003") # session.add(obj1) # # 增加多个，不同对象 # session.add_all([ # Users(name="lqz009"), # Users(name="lqz008"), # ]) # session.commit() # 2 删除操作---》查出来再删---》 # session.query(Users).filter(Users.id > 2).delete() # session.commit() # 3 修改操作--》查出来改 # 传字典 # session.query(Users).filter(Users.id > 0).update({"name": "lqz"}) # 类似于django的F查询 # 字符串加 # session.query(Users).filter(Users.id > 0).update({Users.name: Users.name + "099"}, synchronize_session=False) # # 数字加 # # session.query(Users).filter(Users.id > 0).update({"age": Users.age + 1}, synchronize_session="evaluate") # session.commit() # 4 查询操作----》 # r1 = session.query(Users).all() # 查询所有 # 只取age列，把name重命名为xx # select name as xx,age from user; # r2 = session.query(Users.name.label('xx'), Users.age).all() # filter传的是表达式，filter_by传的是参数 # r3 = session.query(Users).filter(Users.name == "lqz").all() # r3 = session.query(Users).filter(Users.id >= 1).all() # r4 = session.query(Users).filter_by(name='lqz').all() # r5 = session.query(Users).filter_by(name='lqz').first() #:value 和:name 相当于占位符，用params传参数 # r6 = session.query(Users).filter(text("id<:value and name=:name")).params(value=224, name='lqz').order_by(Users.id).all() # 自定义查询sql # r7 = session.query(Users).from_statement(text("SELECT * FROM users where name=:name")).params(name='lqz').all()
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 1
- 2
- 3
- 4
- 5
- 6
- 7
3.5 更多查询操作
# 更多查询 # 条件 # select * form user where name =lqz # ret = session.query(Users).filter_by(name='lqz').all() # 表达式，and条件连接 # select * from user where id >1 and name = lqz # ret = session.query(Users).filter(Users.id > 1, Users.name == 'lqz').all() # select * from user where id between 1,3 and name = lqz # ret = session.query(Users).filter(Users.id.between(1, 3), Users.name == 'lqz').all() # # 注意下划线 # select * from user where id in (1,3,4) # ret = session.query(Users).filter(Users.id.in_([1, 3, 4])).all() # # ~非，除。。外 # select * from user where id not in (1,3,4) # ret = session.query(Users).filter(~Users.id.in_([1, 3, 4])).all() # # 二次筛选 # ret = session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name='lqz'))).all() from sqlalchemy import and_, or_ # # # or_包裹的都是or条件，and_包裹的都是and条件 # ret = session.query(Users).filter(and_(Users.id > 3, Users.name == 'eric')).all() # ret = session.query(Users).filter(or_(Users.id < 2, Users.name == 'eric')).all() # ret = session.query(Users).filter( # or_( # Users.id < 2, # and_(Users.name == 'eric', Users.id > 3), # Users.extra != "" # )).all() # # # 通配符，以e开头，不以e开头 # ret = session.query(Users).filter(Users.name.like('e%')).all() # ret = session.query(Users).filter(~Users.name.like('e%')).all() # # # 限制，用于分页，区间 # ret = session.query(Users)[1:2] # # # 排序，根据name降序排列（从大到小） # ret = session.query(Users).order_by(Users.id.desc()).all() # # 第一个条件重复后，再按第二个条件升序排 # ret = session.query(Users).order_by(Users.name.desc(), Users.id.asc()).all() # # # 分组 from sqlalchemy.sql import func # select * from user group by user.extra; # ret = session.query(Users).group_by(Users.extra).all() # # 分组之后取最大id，id之和，最小id # select max(id),sum(id),min(id) from user group by name ; # ret = session.query( # func.max(Users.id), # func.sum(Users.id), # func.min(Users.id)).group_by(Users.name).all() # haviing筛选 # select max(id),sum(id),min(id) from user group by name having min(id)>2; # ret = session.query( # func.max(Users.id), # func.sum(Users.id), # func.min(Users.id)).group_by(Users.name).having(func.min(Users.id) > 2).all()# select max(id),sum(id),min(id) from user where id >=1 group by name having min(id)>2; # ret = session.query( # func.max(Users.id), # func.sum(Users.id), # func.min(Users.id)).filter(Users.id>=1).group_by(Users.name).having(func.min(Users.id) > 2).all() # # 连表（默认用forinkey关联） # select * from user,favor where user.id=favor.id # ret = session.query(Users, Favor).filter(Users.id == Favor.nid).all() # join表，默认是inner join # select * from Person inner join favor on person.favor=favor.id; # ret = session.query(Person).join(Favor).all() # isouter=True 外连，表示Person left join Favor，没有右连接，反过来即可 # ret = session.query(Person).join(Favor, isouter=True).all() # ret = session.query(Favor).join(Person, isouter=True).all() # 打印原生sql # aa = session.query(Person).join(Favor, isouter=True) # print(aa) # 自己指定on条件（连表条件）,第二个参数，支持on多个条件，用and_,同上 # select * from person left join favor on person.id=favor.id; # ret = session.query(Person).join(Favor, Person.id == Favor.id, isouter=True).all() # 组合（了解）UNION 操作符用于合并两个或多个 SELECT 语句的结果集 # union和union all的区别？ # q1 = session.query(Users.name).filter(Users.id > 2) # q2 = session.query(Favor.caption).filter(Favor.nid < 2) # ret = q1.union(q2).all() # # q1 = session.query(Users.name).filter(Users.id > 2) # q2 = session.query(Favor.caption).filter(Favor.nid < 2) # ret = q1.union_all(q2).all()
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
3.6 执行原生sql
# 执行原生sql # 查询 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)
- 1
- 2
- 3
- 4
- 1
- 2
- 3
- 4
# django 中执行原生sql---》原生sql用的较多
ret = models.Author.objects.raw('select * from book where nid>1')
# 把查回来的数据，直接映射给author对象--》name
print(ret)
for i in ret:
print(i)
print(ret.query)
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
3 一对多
3.1 表模型创建
class Hobby(Base): __tablename__ = 'hobby' id = Column(Integer, primary_key=True) caption = Column(String(50), default='篮球')class Person(Base): __tablename__ = 'person' nid = Column(Integer, primary_key=True) name = Column(String(32), index=True, nullable=True) # hobby指的是tablename而不是类名 hobby_id = Column(Integer, ForeignKey("hobby.id")) # 外键 # 跟数据库无关，不会新增字段，只用于快速链表操作 # 类名，backref用于反向查询 # 正向查询按字段，反向查询按 pers hobby = relationship('Hobby', backref='pers')
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 1
- 2
- 3
- 4
3.2 操作
# 一对多 import datetime from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index from sqlalchemy.orm import sessionmaker from models import Users from sqlalchemy.orm import scoped_session from threading import Thread # 第三步：迁移，通过表模型，生成表 from models import Base from sqlalchemy.sql import text engine = create_engine( "mysql+pymysql://root:123@127.0.0.1:3306/db01?charset=utf8", max_overflow=0, # 超过连接池大小外最多创建的连接 pool_size=5, # 连接池大小 pool_timeout=30, # 池中没有线程最多等待的时间，否则报错 pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置） ) def create_table(): # 通过engine这个连接配置，创建出所有使用Base管理的表 Base.metadata.create_all(engine) def delete_table(): # 通过engine这个连接配置，删除出所有使用Base管理的表 Base.metadata.drop_all(engine) if __name__ == '__main__': create_table() # delete_table() Session = sessionmaker(bind=engine) session = scoped_session(Session) from models import Hobby,Person #1 增加数据 # 方式一 # session.add_all([ # Hobby(caption='乒乓球'), # Hobby(caption='羽毛球'), # Person(name='张三', hobby_id=1), # Person(name='李四', hobby_id=1), # ]) # session.commit() # 方式二 # person = Person(name='张九', hobby=Hobby(caption='姑娘')) # session.add(person) # # 添加三 # hb = Hobby(caption='保龄球') # # 反向字段 # hb.pers = [Person(name='lqz01'), Person(name='lqz02')] # session.add(hb) # session.commit() #2 查询 # 正向查询 # person = session.query(Person).first() # print(person.name) # # 基于对象的跨表查询 # print(person.hobby.caption) # 反向查询 # v = session.query(Hobby).first() # print(v.caption) # print(v.pers) # 多条 # 链表查询 # select person.name ,hobby.caption from person left join bobby on person.hobby_id=hobby.id; # person_list=session.query(Person.name,Hobby.caption).join(Hobby,isouter=True).all() # # person_list = session.query(Person,Hobby).join(Hobby, isouter=True).all() # for row in person_list: # # print(row.name,row.caption) # print(row[0].name, row[1].caption) # person_list = session.query(Person).all() # for row in person_list: # print(row.name, row.hobby.caption) # # obj = session.query(Hobby).filter(Hobby.id == 1).first() # persons = obj.pers # print(persons) # session.close()
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
- 82
- 83
- 1
4 多对多
3.1 表模型创建
# boy girl 相亲，一个boy可以约多个女生，一个女生可以相多个男生 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) class Boy(Base): __tablename__ = 'boy' id = Column(Integer, primary_key=True, autoincrement=True) name = Column(String(64), unique=True, nullable=False) # 与生成表结构无关，仅用于查询方便,放在哪个单表中都可以 girls = relationship('Girl', secondary='boy2girl', backref='boys')
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 1
- 2
- 3
- 4
- 5
- 6
3.2 操作
# 多对多 import datetime from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index from sqlalchemy.orm import sessionmaker from models import Users from sqlalchemy.orm import scoped_session from threading import Thread # 第三步：迁移，通过表模型，生成表 from models import Base from sqlalchemy.sql import text engine = create_engine( "mysql+pymysql://root:123@127.0.0.1:3306/db01?charset=utf8", max_overflow=0, # 超过连接池大小外最多创建的连接 pool_size=5, # 连接池大小 pool_timeout=30, # 池中没有线程最多等待的时间，否则报错 pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置） ) def create_table(): # 通过engine这个连接配置，创建出所有使用Base管理的表 Base.metadata.create_all(engine) def delete_table(): # 通过engine这个连接配置，删除出所有使用Base管理的表 Base.metadata.drop_all(engine) from models import Boy,Girl,Boy2Girl if __name__ == '__main__': create_table() # delete_table() Session = sessionmaker(bind=engine) session = scoped_session(Session) # 1 增加数据 ## 方式一 # session.add_all([ # Boy(name='彭于晏'), # Boy(name='刘德华'), # Girl(name='刘亦菲'), # Girl(name='迪丽热巴'), # ]) # session.commit() # s2g = Boy2Girl(boy_id=1, girl_id=1) # session.add(s2g) # session.commit() ## 方式二 # boy = Boy(name='lqz') # boy.girls = [Girl(name='小红'), Girl(name='校花')] # session.add(boy) # session.commit() ### 方式三 # girl = Girl(name='小梅') # girl.boys = [Boy(name='lqz001'), Boy(name='lqz002')] # session.add(girl) # session.commit() # 基于对象的跨表查 # 使用relationship正向查询 # v = session.query(Boy).first() # print(v.name) # print(v.girls) # 使用relationship反向查询 v = session.query(Girl).first() print(v.name) print(v.boys)
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 1
5 flask集成
# Flask_SQLAlchemy 操作数据库 # flask_migrate 模拟django的表迁移 pip3 install flask_migrate# flask_migrate使用步骤 from flask_sqlalchemy import SQLAlchemy db = SQLAlchemy() # 全局SQLAlchemy app = Flask(__name__) app.config.from_object('settings.DevelopmentConfig') # 将db注册到app中，加载配置文件，flask-session，用一个类包裹一下app db.init_app(app) # flask_script创建命令 runserver命令 ，自定义名字 # 下面三句会创建出两个命令:runserver db 命令(flask_migrate) manager=Manager(app) Migrate(app, db) manager.add_command('db',MigrateCommand ) # 添加一个db命令，原来有了runserver命令了 # 直接使用命令迁移表即可 # 1 python3 manage.py db init #初始化，刚开始干，生成一个migrate文件夹 # 2 创建表，修改表 # python3 manage.py db migrate 等同于 makemigartions # python3 manage.py db upgrade 等同于migrate
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 1
- 2
- 3
# Flask_SQLAlchemy给你包装了基类，和session，以后拿到db
db = SQLAlchemy() # 全局SQLAlchemy
# 增删查改数据-->并发安全
db.session.query()
# 表模型要继承基表
class Users(db.Model):
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
文章转载自：https://www.cnblogs.com/ydy001/p/16299549.html，如存在版权问题，请联系。
声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/AllinToyou/article/detail/613254
推荐阅读
相关标签
