python学习之flask 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.1 执行原生sql
# 执行原生sql快速使用
import timeimport threadingimport sqlalchemyfrom sqlalchemy import create_enginefrom sqlalchemy.engine.base import Engine# 第一步:创建engineengine = 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的客户端连接数
2 单表操作增删查改
2.1 表迁移
# 不能创建数据库(django orm也不能)# 只能做表的创建和删除,不能做表更改(django orm能)---》借助于第三方实现
###### # 第一步:生成基类,所有表模型都要继承这个基类# django 的orm继承一个父类,Base就是那个父类###### 第二步:写表模型,继承父类,写字段 (注意区别于django 的orm)# django的default--》可不可以传个函数内存地址---》插入的时候通过函数运算完得到的值###### 第三步:迁移,通过表模型,生成表
import datetimefrom sqlalchemy import create_enginefrom sqlalchemy.ext.declarative import declarative_basefrom 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会隐藏一个# 辅助索引:手动建的叫辅助索引---》单独减了索引---》如果你的辅助索引过多,非常影响插入效率,适度建索引
import datetimefrom sqlalchemy import create_enginefrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index# 第三步:迁移,通过表模型,生成表from models import Baseengine = 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()
2.2 简单的表操作
# 操作表,增加一条记录,以后都用conn/session操作# 第一步:创建engin# 第二步:通过session得到连接对象# Session = sessionmaker(bind=engine)# session = Session()# # 第三步:实例化得到模型类的对象,增加到数据库中# usr=Users(name='lqz001')# session.add(usr)# # 第四步:提交事务# session.commit()
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()
测试线程安全
# 线程一用:取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()
3.4 基本增删查改
import datetimefrom sqlalchemy import create_enginefrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Indexfrom sqlalchemy.orm import sessionmakerfrom models import Usersfrom sqlalchemy.orm import scoped_session
from threading import Thread# 第三步:迁移,通过表模型,生成表from models import Basefrom sqlalchemy.sql import textengine = 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()
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()
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)
# 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)
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用于反向查询 # 正向查询按字段,反向查询按 pershobby = relationship('Hobby', backref='pers')
3.2 操作
# 一对多import datetimefrom sqlalchemy import create_enginefrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Indexfrom sqlalchemy.orm import sessionmakerfrom models import Usersfrom sqlalchemy.orm import scoped_sessionfrom threading import Thread# 第三步:迁移,通过表模型,生成表from models import Basefrom sqlalchemy.sql import textengine = 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()
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')
3.2 操作
# 多对多import datetimefrom sqlalchemy import create_enginefrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Indexfrom sqlalchemy.orm import sessionmakerfrom models import Usersfrom sqlalchemy.orm import scoped_sessionfrom threading import Thread# 第三步:迁移,通过表模型,生成表from models import Basefrom sqlalchemy.sql import textengine = 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,Boy2Girlif __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)
5 flask集成
# Flask_SQLAlchemy 操作数据库# flask_migrate 模拟django的表迁移pip3 install flask_migrate
# flask_migrate使用步骤from flask_sqlalchemy import SQLAlchemydb = SQLAlchemy() # 全局SQLAlchemyapp = Flask(__name__)app.config.from_object('settings.DevelopmentConfig')# 将db注册到app中,加载配置文件,flask-session,用一个类包裹一下appdb.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
# Flask_SQLAlchemy给你包装了基类,和session,以后拿到dbdb = SQLAlchemy() # 全局SQLAlchemy# 增删查改数据-->并发安全db.session.query()# 表模型要继承基表class Users(db.Model):
文章转载自:https://www.cnblogs.com/ydy001/p/16299549.html,如存在版权问题,请联系。
python学习之flask sqlalchemy介绍和使用 表操作增删查改相关推荐
- MySQL学习(3)——表的增删查改(进阶)
1.数据库约束 1.1 约束类型 NOT NULL - 指示某列不能存储null值 UNIQUE - 保证某列的每行必须有唯一的值 DEFAULT - 规定没有给列赋值时的默认值 PRIMARY KE ...
- MySQL学习(2)——MySQL表的增删查改(基础)
1.CRUD CRUD 即增加(Create).查询(Retrieve).更新(Update).删除(Delete) 四个单词的首字母缩写 2.新增(Cteate) 语法: INSERT [INTO] ...
- 升级版Python学习教程:SQLAlchemy太庞大,不妨试试这位小清新-Peewee
SQLAlchemy 功能很强大,文档很丰富,是一个重量级的 ORM 框架.本篇Python学习教程给大家介绍一个小清新,轻量级 ORM 框架 Peewee,支持 Python 2.7+ 和 3.4+ ...
- day 45 SQLAlchemy,和增删查改
SQLAlchemy,和增删查改 SQLAlchemy: SQLAlchemy是Python编程语言下的一款ORM框架,该框架建立在数据库API之上,使用关系对象映射进行数据库操作,简言之便是:将对象 ...
- 捋一捋Python的文件属性和增删查改等(下)
正式的Python专栏第35篇,同学站住,别错过这个从0开始的文章! 前面写了文件的读取,按行读写等,这篇我们把文件处理等其他函数也过一过吧. 文件属性 前面我们更多集中学习了文件的读写(open)函 ...
- Python学生信息管理系统(增删查改、模糊查找、txt文件输出)# 谭子
一.系统需求说明 本项目计划实现一个学生管理系统,学生信息包括:姓名.性别.手机号码,系统包含以下功能. 模块 子模块 说明 查询模块 查询全部学生的信息 显示当前系统内所有学员的信息 查询模块 精准 ...
- TP框架增删改查需要掉ajax么6,TP6框架--EasyAdmin学习笔记:实现数据库增删查改
这是我写的学习EasyAdmin的第三章,这一章我给大家分享下如何进行数据库的增删查改 上一章链接:点击这里前往 上一章我们说到,我仿照官方案例,定义了一条路由goodsone和创建了对应数据库,我们 ...
- linux增删查改语句,mysql基础知识之增删查改使用介绍
mysql基础知识之增删查改使用介绍 本文主要介绍mysql常用的SELECT.INSERT.UPDATE.DELETE语句的使用,数据库的安装这里不做介绍,并且事先已经准备好相关数据. 本文中使用的 ...
- MongoDB入门学习(三):MongoDB的增删查改
对于我们这样的菜鸟来说,最重要的不是数据库的管理,也不是数据库的性能,更不是数据库的扩展,而是怎么用好这款数据库,也就是一个数据库提供的最核心的功能,增删查改. 由于MongoDB存储数据都是以文档的 ...
最新文章
- 浅谈图网络在视觉任务中的应用
- Android sqlite 数据库保存Date 类型
- python奇数和_请问python如何判断奇偶数?
- 了解ViewConfiguration
- 【资源】分享1套最适合Py开发的机器学习/大数据视频课程
- Mysql对事务的支持
- cnn图像二分类 python_TensorFlow2基础:CNN图像分类
- python怎么查询元素是否在列表中_python怎么判断某一元素是否在列表中
- Config Server高可用
- prettytensor 的使用
- 工业电气自动化及电工电子技能考核实训平台(高级版)
- xdg在Linux中的用法,Linux实用命令之xdg-open
- 卖计算机英语对话,英语购买电脑情景对话.doc
- linux用户态切换到内核态方法
- 设计模式之旅(三)--观察者模式
- 如何从零开始学习SEM?
- kubernetes / K8s 初始化失败问题
- 接入小程序客服(java版教程),处理第一次主动推送会话超时问题
- .properties文件加载失败
- Linux 命令 —— tree