Flask——多app应用(了解即可,以弃用)、flask-script(制定命令)、sqlachemy(详细介绍)、SQLAlchemy
2022/1/2 2:07:20
本文主要是介绍Flask——多app应用(了解即可,以弃用)、flask-script(制定命令)、sqlachemy(详细介绍)、SQLAlchemy,对大家解决编程问题具有一定的参考价值,需要的程序猿们随着小编来一起学习吧!
文章目录
- 昨日回顾
- 1 多app应用(了解即可,以弃用)
- 2 flask-script(制定命令)
- 3 sqlachemy
- 3.0 概念
- 3.1 基本使用(原生sql)
- 3.2 orm使用
- 3.3 线程安全
- 3.4 基本增删查改
- 3.5 高级操作
- 3.6 多表操作
- 3.7 models.py
- 4 Flask-SQLAlchemy
昨日回顾
1 注册一个力扣,每天至少5道题 2 你平时爱浏览什么网站(cnblogs,掘金,简书,知乎,csdn),用什么搜索引擎,遇到问题去哪解决(官方文档,知乎提问,思否提问) 3 两万字旁白,个人介绍,项目描述,个人技能写的点(几百字) 练习说,自己拿出20分时间,自己复述,你的同学复述,个人介绍,技术点 4 falsk回顾 -基本使用 -路由本质:app.add_rule_url() -配置文件多种方式 -模板语言,include,xss攻击,处理xss攻击的本质原理 -请求和响应,render_template,redirect,jsonfy -CBV原理,dispatch,和FBV -flask session源码执行流程 -闪现 -蓝图 -请求扩展(before_first_request,before_request...) -中间件 -flask请求上下文源码,flask核心源码,flask执行流程,LocalProxy,代理模式,requset源码,session源码, -g对象 -信号(内置,自定义) -第三方插件 -flask-session() -flask-cache(未讲),自定义缓存 -wtforms -数据库连接池
1 多app应用(了解即可,以弃用)
### 多个app实例(弃用) from werkzeug.wsgi import DispatcherMiddleware from werkzeug.serving import run_simple from flask import Flask, current_app app1 = Flask('app01') app2 = Flask('app02') @app1.route('/index') def index(): return "app01" @app2.route('/index2') def index2(): return "app2" # http://www.oldboyedu.com/index # http://www.oldboyedu.com/sec/index2 dm = DispatcherMiddleware(app1, { '/sec': app2, }) if __name__ == "__main__": run_simple('localhost', 5000, dm) # 请求来了,会执行dm()--->__call__
2 flask-script(制定命令)
1 模拟出类似django的启动方式:python manage.py runserver 2 pip install flask-script 3 把excel的数据导入数据库,定制个命令,去执行(openpyxl) python manage.py insertdb -f xxx.excl -t aa 4 使用 from flask import Flask from flask_script import Manager app = Flask(__name__) manager=Manager(app) # 自定制命令 Python manage.py createsuperuser @manager.command def custom(arg): """ 自定义命令 python manage.py custom 123 :param arg: :return: """ print(arg) @manager.option('-n', '--name', dest='name') @manager.option('-u', '--url', dest='url') def cmd(name, url): """ 自定义命令(-n也可以写成--name) 执行: python manage.py cmd -n lqz -u http://www.oldboyedu.com 执行: python manage.py cmd --name lqz --url http://www.oldboyedu.com :param name: :param url: :return: """ print(name, url) if __name__ == '__main__': manager.run() 5 创建超级用户 6 现在有一万条excel用户,批量导入到数据库中 -1. navicate直接支持 -2. 自己写脚本,flask-script
3 sqlachemy
3.0 概念
1 sqlachemy:第三方orm框架(对象关系映射) -go 中gorm,xorm -python中:django orm,sqlachemy,peewee -老刘带你手写的:https://www.cnblogs.com/liuqingzheng/articles/9006025.html 2 django orm,只能在django中用,不能单独用 3 使用 pip install sqlachemy 4 SQLAlchemy本身无法操作数据库,其必须以来pymsql等第三方插件 5 补充:django orm反向生成models -python manage.py inspectdb > app/models.py
3.1 基本使用(原生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/flask?charset=utf8", max_overflow=0, # 超过连接池大小外最多创建的连接 pool_size=5, # 连接池大小 pool_timeout=30, # 池中没有线程最多等待的时间,否则报错 pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收(重置) ) # 第二步:创建连接(执行原生sql) conn = engine.raw_connection() # 第三步:获取游标对象 cursor = conn.cursor() # 第四步:具体操作 cursor.execute('select * from boy') res=cursor.fetchall() print(res) # 比pymysql优势在,有数据库连接池
3.2 orm使用
# 创建一个个类(继承谁?字段怎么写) 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 # 制造了一个类,作为所有模型类的基类 Base = declarative_base() class User(Base): __tablename__ = 'users' # 数据库表名称(固定写法),如果不写,默认以类名小写作为表的名字 id = Column(Integer, primary_key=True) # id 主键 # mysql中主键自动建索引:聚簇索引 # 其他建建的索引叫:辅助索引 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) # default默认值 # extra = Column(Text, nullable=True) #类似于djagno的 Meta # __table_args__ = ( # UniqueConstraint('id', 'name', name='uix_id_name'), #联合唯一 # Index('ix_id_name', 'name', 'email'), #索引 # ) # 创建表 def create_table(): # 创建engine对象 engine = create_engine( "mysql+pymysql://root:123@127.0.0.1:3306/aaa?charset=utf8", max_overflow=0, # 超过连接池大小外最多创建的连接 pool_size=5, # 连接池大小 pool_timeout=30, # 池中没有线程最多等待的时间,否则报错 pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收(重置) ) # 通过engine对象创建表 Base.metadata.create_all(engine) # 删除表 def drop_table(): # 创建engine对象 engine = create_engine( "mysql+pymysql://root:123@127.0.0.1:3306/aaa?charset=utf8", max_overflow=0, # 超过连接池大小外最多创建的连接 pool_size=5, # 连接池大小 pool_timeout=30, # 池中没有线程最多等待的时间,否则报错 pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收(重置) ) # 通过engine对象删除所有表 Base.metadata.drop_all(engine) if __name__ == '__main__': # create_table() drop_table() # 创建库?需要手动创建库 # 问题,sqlachemy支持修改字段吗?不支持
3.3 线程安全
#基于scoped_session实现线程安全 from sqlalchemy.orm import sessionmaker from sqlalchemy import create_engine from models import User # pycharm报错,不会影响我们 from sqlalchemy.orm import scoped_session # 1 制作engine engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/aaa", max_overflow=0, pool_size=5) # 2 制造一个 session 类(会话) Session = sessionmaker(bind=engine) # 得到一个类 # 3 得到一个session对象(线程安全的session) #现在的session已经不是session对象了 #为什么线程安全,还是用的local session = scoped_session(Session) # session=Session() # 4 创建一个对象 obj1 = User(name="2008") # 5 把对象通过add放入 session.add(obj1) # 6 提交 session.commit() session.close() # 类不继承Session类,但是有该类的所有方法(通过反射,一个个放进去) # scoped_session.add------->instrument(name)--->do函数内存地址---》现在假设我要这么用:session.add()--->do() # scoped_session.close----->instrument(name)--->do函数内存地址
3.4 基本增删查改
from sqlalchemy.orm import sessionmaker from sqlalchemy import create_engine from models import User,Person,Hobby from sqlalchemy.orm import scoped_session from sqlalchemy.sql import text engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/aaa", max_overflow=0, pool_size=5) Session = sessionmaker(bind=engine) # session = scoped_session(Session) session=Session() ####1 新增多个对象 # obj=User(name='xxx') # obj2=User(name='yyyy') # obj3=User(name='zzz') #新增同样对象 # session.add_all([obj,obj2,obj3]) #新增不同对象 # session.add_all([Person(name='lqz'),Hobby()]) ####2 简单删除(查到删除) # res=session.query(User).filter_by(name='2008').delete() # res=session.query(User).filter(User.id>=2).delete() # # 影响1行 # print(res) #### 3 修改 # res=session.query(User).filter_by(id=1).update({User.name:'ccc'}) # res=session.query(User).filter_by(id=1).update({'name':'ccc'}) # session.query(User).filter(User.id > 0).update({User.name: User.name + "099"}, synchronize_session=False) # 如果要把它转成字符串相加 # session.query(User).filter(User.id > 0).update({"age": User.age + 1}, synchronize_session="evaluate") ## 如果要把它转成数字相加 ####4 基本查询操作 # res=session.query(User).all() # print(type(res)) # res=session.query(User).first() # print(res) #filter传的是表达式,filter_by传的是参数 # res=session.query(User).filter(User.id==1).all() # res=session.query(User).filter(User.id>=1).all() # res=session.query(User).filter(User.id<1).all() # res=session.query(User).filter_by(name='ccc099').all() #了解 # res = session.query(User).from_statement(text("SELECT * FROM users where name=:name")).params(name='ccc099').all() # print(res) session.commit() # 并没有真正关闭连接,而是放回池中 session.close()
3.5 高级操作
from sqlalchemy.orm import sessionmaker from sqlalchemy import create_engine from models import User,Person,Hobby from sqlalchemy.sql import text engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/aaa", max_overflow=0, pool_size=5) Session = sessionmaker(bind=engine) session=Session() # 1 查询名字为lqz的所有user对象 # ret = session.query(User).filter_by(name='ccc099').all() # 2 表达式,and条件连接 # ret = session.query(User).filter(User.id > 1, User.name == 'egon').all() # 查找id在1和10之间,并且name=egon的对象 # ret = session.query(User).filter(User.id.between(1, 10), User.name == 'egon').all() # in条件(class_,因为这是关键字,不能直接用) # ret = session.query(User).filter(User.id.in_([1,3,4])).all() # 取反 ~ ret = session.query(User).filter(~User.id.in_([1,3,4])).all() #二次筛选 # select * # ret = session.query(User).filter(User.id.in_(session.query(User.id).filter_by(name='egon'))).all() # # select name,id 。。。。 # ret = session.query(User.id,User.name).filter(User.id.in_(session.query(User.id).filter_by(name='egon'))).all() ''' SELECT users.id AS users_id, users.name AS users_name FROM users WHERE users.id IN (SELECT users.id AS users_id FROM users WHERE users.name = %(name_1)s) ''' # from sqlalchemy import and_, or_ #or_包裹的都是or条件,and_包裹的都是and条件 #查询id>3并且name=egon的人 # ret = session.query(User).filter(and_(User.id > 3, User.name == 'egon')).all() # 查询id大于2或者name=ccc099的数据 # ret = session.query(User).filter(or_(User.id > 2, User.name == 'ccc099')).all() # ret = session.query(User).filter( # or_( # User.id < 2, # and_(User.name == 'egon', User.id > 3), # User.extra != "" # )).all() # print(ret) ''' select *from user where id<2 or (name=egon and id >3) or extra !='' ''' # 通配符,以e开头,不以e开头 # ret = session.query(User).filter(User.name.like('e%')).all() # ret = session.query(User).filter(~User.name.like('e%')).all() # 限制,用于分页,区间 limit # 前闭后开区间,1能取到,3取不到 ret = session.query(User)[1:3] ''' select * from users limit 1,2; ''' # 排序,根据name降序排列(从大到小) # ret = session.query(User).order_by(User.name.desc()).all() # ret = session.query(User).order_by(User.name.asc()).all() #第一个条件降序排序后,再按第二个条件升序排 # ret = session.query(User).order_by(User.id.asc(),User.name.desc()).all() # ret = session.query(User).order_by(User.name.desc(),User.id.asc()).all() # 分组 from sqlalchemy.sql import func # ret = session.query(User).group_by(User.name).all() #分组之后取最大id,id之和,最小id # sql 分组之后,要查询的字段只能有分组字段和聚合函数 # ret = session.query( # func.max(User.id), # func.sum(User.id), # func.min(User.id), # User.name).group_by(User.name).all() # ''' # select max(id),sum(id),min(id) from user group by name; # # ''' # for obj in ret: # print(obj[0],'----',obj[1],'-----',obj[2],'-----',obj[3]) # print(ret) #haviing筛选 # ret = session.query( # func.max(User.id), # func.sum(User.id), # func.min(User.id)).group_by(User.name).having(func.min(User.id) >2).all() ''' select max(id),sum(id),min(id) from user group by name having min(id)>2; ''' print(ret) session.commit() session.close()
3.6 多表操作
from sqlalchemy.orm import sessionmaker from sqlalchemy import create_engine from models import User,Person,Hobby,Boy,Girl,Boy2Girl from sqlalchemy.sql import text engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/aaa", max_overflow=0, pool_size=5) Session = sessionmaker(bind=engine) session=Session() ### 1 一对多插入数据 # obj=Hobby(caption='足球') # session.add(obj) # p=Person(name='张三',hobby_id=2) # session.add(p) ### 2 方式二(默认情况传对象有问题) ###### Person表中要加 hobby = relationship('Hobby', backref='pers') # p=Person(name='李四',hobby=Hobby(caption='美女')) # 等同于 # p=Person(name='李四2') # p.hobby=Hobby(caption='美女2') # session.add(p) ## 3 方式三,通过反向操作 # hb = Hobby(caption='人妖') # hb.pers = [Person(name='文飞'), Person(name='博雅')] # session.add(hb) #### 4 查询(查询:基于连表的查询,基于对象的跨表查询) ### 4.1 基于对象的跨表查询(子查询,两次查询) # 正查 # p=session.query(Person).filter_by(name='张三').first() # print(p) # print(p.hobby.caption) # 反查 # h=session.query(Hobby).filter_by(caption='人妖').first() # print(h.pers) ### 4.2 基于连表的跨表查(查一次) # 默认根据外键连表 # isouter=True 左外连,表示Person left join Hobby,没有右连接,反过来即可 # 不写 inner join # person_list=session.query(Person,Hobby).join(Hobby,isouter=True).all() # print(person_list) # print(person_list) # for row in person_list: # print(row[0].name,row[1].caption) # ''' # select * from person left join hobby on person.hobby_id=hobby.id # ''' # # ret = session.query(Person, Hobby).filter(Person.hobby_id == Hobby.id) # print(ret) # ''' # select * from user,hobby where user.id=favor.nid; # # ''' #join表,默认是inner join # ret = session.query(Person).join(Hobby) # # ret = session.query(Hobby).join(Person,isouter=True) # ''' # SELECT * # FROM person INNER JOIN hobby ON hobby.id = person.hobby_id # ''' # print(ret) # 指定连表字段(从来没用过) # ret = session.query(Person).join(Hobby,Person.nid==Hobby.id, isouter=True) # # ret = session.query(Person).join(Hobby,Person.hobby_id==Hobby.id, isouter=True).all() # print(ret) ''' SELECT * FROM person LEFT OUTER JOIN hobby ON person.nid = hobby.id ''' # print(ret) # 组合(了解)UNION 操作符用于合并两个或多个 SELECT 语句的结果集 # union和union all的区别? # q1 = session.query(User.name).filter(User.id > 2) # 6条数据 # q2 = session.query(User.name).filter(User.id < 8) # 2条数据 # q1 = session.query(User.id,User.name).filter(User.id > 2) # 6条数据 # q2 = session.query(User.id,User.name).filter(User.id < 8) # 2条数据 # ret = q1.union_all(q2).all() # ret1 = q1.union(q2).all() # print(ret) # print(ret1) # # q1 = session.query(User.name).filter(User.id > 2) # q2 = session.query(Hobby.caption).filter(Hobby.nid < 2) # ret = q1.union_all(q2).all() #### 多对多 # session.add_all([ # Boy(hostname='霍建华'), # Boy(hostname='胡歌'), # Girl(name='刘亦菲'), # Girl(name='林心如'), # ]) # session.add_all([ # Boy2Girl(girl_id=1, boy_id=1), # Boy2Girl(girl_id=2, boy_id=1) # ]) ##### 要有girls = relationship('Girl', secondary='boy2girl', backref='boys') # girl = Girl(name='张娜拉') # girl.boys = [Boy(hostname='张铁林'),Boy(hostname='费玉清')] # session.add(girl) # boy=Boy(hostname='蔡徐坤') # boy.girls=[Girl(name='谢娜'),Girl(name='巧碧螺')] # session.add(boy) # session.commit() # 基于对象的跨表查 # girl=session.query(Girl).filter_by(id=3).first() # print(girl.boys) #### 基于连表的跨表查询 # 查询蔡徐坤约过的所有妹子 ''' select girl.name from girl,boy,Boy2Girl where boy.id=Boy2Girl.boy_id and girl.id=Boy2Girl.girl_id where boy.name='蔡徐坤' ''' # ret=session.query(Girl.name).filter(Boy.id==Boy2Girl.boy_id,Girl.id==Boy2Girl.girl_id,Boy.hostname=='蔡徐坤').all() ''' select girl.name from girl inner join Boy2Girl on girl.id=Boy2Girl.girl_id inner join boy on boy.id=Boy2Girl.boy_id where boy.hostname='蔡徐坤' ''' # ret=session.query(Girl.name).join(Boy2Girl).join(Boy).filter(Boy.hostname=='蔡徐坤').all() ret=session.query(Girl.name).join(Boy2Girl).join(Boy).filter_by(hostname='蔡徐坤').all() print(ret) ### 执行原生sql(用的最多的) ### django中orm如何执行原生sql # # cursor = session.execute('insert into users(name) values(:value)',params={"value":'xxx'}) # print(cursor.lastrowid) # session.commit() session.close()
3.7 models.py
# 创建一个个类(继承谁?字段怎么写) 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 relationship # 制造了一个类,作为所有模型类的基类 Base = declarative_base() class User(Base): __tablename__ = 'users' # 数据库表名称(固定写法),如果不写,默认以类名小写作为表的名字 id = Column(Integer, primary_key=True) # id 主键 # mysql中主键自动建索引:聚簇索引 # 其他建建的索引叫:辅助索引 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) # default默认值 # extra = Column(Text, nullable=True) #类似于djagno的 Meta # __table_args__ = ( # UniqueConstraint('id', 'name', name='uix_id_name'), #联合唯一 # Index('ix_id_name', 'name', 'email'), #索引 # ) def __str__(self): return self.name def __repr__(self): # python是强类型语言 return self.name+str(self.id) # 一对多关系 # 一个Hobby可以有很多人喜欢 # 一个人只能由一个Hobby 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而不是类名,uselist=False # 一对多的关系,关联字段写在多的一方 hobby_id = Column(Integer, ForeignKey("hobby.id")) # 默认可以为空 # 跟数据库无关,不会新增字段,只用于快速链表操作 # 类名,backref用于反向查询 hobby = relationship('Hobby', backref='pers') # 多对多关系 # 实实在在存在的表 class Boy2Girl(Base): __tablename__ = 'boy2girl' id = Column(Integer, primary_key=True, autoincrement=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) hostname = Column(String(64), unique=True, nullable=False) # 与生成表结构无关,仅用于查询方便,放在哪个单表中都可以 # secondary 通过哪个表建关联,跟django中的through一模一样 girls = relationship('Girl', secondary='boy2girl', backref='boys') # 创建表 def create_table(): # 创建engine对象 engine = create_engine( "mysql+pymysql://root:123@127.0.0.1:3306/aaa?charset=utf8", max_overflow=0, # 超过连接池大小外最多创建的连接 pool_size=5, # 连接池大小 pool_timeout=30, # 池中没有线程最多等待的时间,否则报错 pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收(重置) ) # 通过engine对象创建表 Base.metadata.create_all(engine) # 删除表 def drop_table(): # 创建engine对象 engine = create_engine( "mysql+pymysql://root:123@127.0.0.1:3306/aaa?charset=utf8", max_overflow=0, # 超过连接池大小外最多创建的连接 pool_size=5, # 连接池大小 pool_timeout=30, # 池中没有线程最多等待的时间,否则报错 pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收(重置) ) # 通过engine对象删除所有表 Base.metadata.drop_all(engine) if __name__ == '__main__': create_table() # 原来已经存在user表,再执行一次不会有问题 # drop_table() # 创建库?手动创建库 # 问题,sqlachemy支持修改字段吗?不支持
4 Flask-SQLAlchemy
1 Flask-SQLAlchemy 2 flask-migrate -python3 manage.py db init 初始化:只执行一次 -python3 manage.py db migrate 等同于 makemigartions -python3 manage.py db upgrade 等同于migrate 4 Flask-SQLAlchemy如何使用 1 from flask_sqlalchemy import SQLAlchemy 2 db = SQLAlchemy() 3 db.init_app(app) 4 以后在视图函数中使用 -db.session 就是咱们讲的session 5 flask-migrate的使用(表创建,字段修改) 1 from flask_migrate import Migrate,MigrateCommand 2 Migrate(app,db) 3 manager.add_command('db', MigrateCommand) 6 直接使用 -python3 manage.py db init 初始化:只执行一次,创建migrations文件夹 -python3 manage.py db migrate 等同于 makemigartions -python3 manage.py db upgrade 等同于migrate
这篇关于Flask——多app应用(了解即可,以弃用)、flask-script(制定命令)、sqlachemy(详细介绍)、SQLAlchemy的文章就介绍到这儿,希望我们推荐的文章对大家有所帮助,也希望大家多多支持为之网!
- 2024-05-29Elasticsearch慢查询日志配置
- 2024-05-29揭秘华为如此多成功项目的产品关键——Charter模板
- 2024-05-29海外IDC业务拓展的7大挑战
- 2024-05-29InLine Chat功能优化对标Github Copilot,CodeGeeX带来更高效、更直观的编程体验!
- 2024-05-29CodeGeeX 智能编程助手 6 项功能升级,在Visual Studio插件市场霸榜2周!
- 2024-05-29AutoMQ 生态集成 Apache Doris
- 2024-05-292024年IDC行业的深度挖掘:机遇、挑战与未来展望
- 2024-05-29五款扩展组件齐发 —— Volcano、Keda、Crane-scheduler 等,邀你体验
- 2024-05-29AutoMQ 对象存储数据高效组织的秘密: Compaction
- 2024-05-29活动预告|来 GIAC 大会听大数据降本利器:AutoMQ 基于云原生重新设计的 Kafka