python3-开发进阶Flask的基础(5)
2021-07-12 13:09
标签：传参 ack div 添加 cut 多个 oca index factory 内容概要： SQLAlchemy是一个ORM的框架，ORM就是关系对象映射，具体可以参照Django中的ORM。 作用：帮助我们使用类和对象快速实现数据库操作 数据库： -原生：MYSQLdb pymysql 区别就是 MYSQLdb 不支持python3 pymysql 都支持 ORM框架 SQLAlchemy 1、安装 2、配置 3、增删改查的演示 4、单表常用操作 5、连表常用操作 在上面的配置里，重新搞两给表 下面是实例演示：之前自己添加一点数据： 连接的方式： 执行原生SQ: a、先下载安装 b、项目下的__init__.py导入 c、初始化 d、在配置文件中写入配置 e、创建models.py中的类（对应数据库中的表） f、生成表（使用app上下文） 补充一个知识点： g、基于ORM对数据库进行操作 这里面db.session.add的源码： 下载安装 功能： a、增加runsever b、位置传参 c、关键字传参 安装 配置是依赖 flask-script 1、pipreqs 找到项目使用的所有组件的版本 在项目中的命令行 在项目中找了文件requirements.txt 如何装那些模块？ pip install 的时候可以指定一个文件，他会自己读每一行，然后安装 pycharm 打开文件的时候也会提示下载 2、虚拟环境 （开发环境版本不能共存的） 安装 创建虚拟环境 python3-开发进阶Flask的基础(5) 标签：传参 ack div 添加 cut 多个 oca index factory 原文地址：https://www.cnblogs.com/ManyQian/p/9532440.html
一、SQLAlchemy
1、概述
2、SQLAlchemy用法
pip3 install sqlalchemy
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column
from sqlalchemy import Integer,String,Text,Date,DateTime
from sqlalchemy import create_engine
Base = declarative_base()
class Users(Base):
__tablename__ = ‘users‘
id = Column(Integer, primary_key=True)
name = Column(String(32), index=True, nullable=False)
def create_all():
engine = create_engine(
"mysql+pymysql://root:123@127.0.0.1:3306/duoduo123?charset=utf8",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
Base.metadata.create_all(engine)
def drop_all():
engine = create_engine(
"mysql+pymysql://root:123@127.0.0.1:3306/duoduo123?charset=utf8",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
Base.metadata.drop_all(engine)
if __name__ == ‘__main__‘:
create_all()
from duoduo.test import Users
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
engine = create_engine(
"mysql+pymysql://root:123@127.0.0.1:3306/duoduo123?charset=utf8",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
SessionFactory = sessionmaker(bind=engine)
#根据Users类对user表进行增删改查
session=SessionFactory()
#增加
#单个
# obj=Users(name=‘many qian‘)
# session.add(obj)
# session.commit()
#多个
# session.add_all([
# Users(name=‘大娃‘),
# Users(name=‘二娃‘)
# ])
# session.commit()
#查 所有
# tes=session.query(Users).all()
# print(tes) #拿到的对象
# for row in tes:
# print(row.id,row.name)
# tes=session.query(Users).filter(Users.id==3) #> =
#
# for row in tes:
# print(row.id,row.name)
#
# tes=session.query(Users).filter(Users.id =
#
# print(tes.id,tes.name)
#删除
# tes=session.query(Users).filter(Users.id==3).delete() #> =
# session.commit()
#改
# session.query(Users).filter(Users.id ==4).update({Users.name:‘三娃‘})
# session.query(Users).filter(Users.id ==4).update({‘name‘:‘四娃‘})
# session.query(Users).filter(Users.id ==4).update({‘name‘:Users.name+‘NICE‘},synchronize_session=False)
# session.commit()
#
#
# session.close()
# ############################## 其他常用 ###############################
# 1. 指定列
# select id,name as cname from users;
# result = session.query(Users.id,Users.name.label(‘cname‘)).all()
# for item in result:
# print(item[0],item.id,item.cname)
# 2. 默认条件and
# session.query(Users).filter(Users.id > 1, Users.name == ‘duoduo‘).all()
# 3. between
# session.query(Users).filter(Users.id.between(1, 3), Users.name == ‘duoduo‘).all()
# 4. in
# session.query(Users).filter(Users.id.in_([1,3,4])).all()
# session.query(Users).filter(~Users.id.in_([1,3,4])).all()
# 5. 子查询
# session.query(Users).filter(Users.id.in_(session.query(Users.id).filter(Users.name==‘duoduo‘))).all()
# 6. and 和 or
# from sqlalchemy import and_, or_
# session.query(Users).filter(Users.id > 3, Users.name == ‘duoduo‘).all()
# session.query(Users).filter(and_(Users.id > 3, Users.name == ‘duoduo‘)).all()
# session.query(Users).filter(or_(Users.id
# session.query(Users).filter(
# or_(
# Users.id
# and_(Users.name == ‘duoduo‘, Users.id > 3),
# Users.extra != ""
# )).all()
# 7. filter_by
# session.query(Users).filter_by(name=‘duoduo‘).all()
# 8. 通配符
# ret = session.query(Users).filter(Users.name.like(‘d%‘)).all() #%任何的东西
# ret = session.query(Users).filter(~Users.name.like(‘d_‘)).all() #_ 只有一个字符
# 9. 切片
# result = session.query(Users)[1:2]
# 10.排序
# ret = session.query(Users).order_by(Users.name.desc()).all()
# ret = session.query(Users).order_by(Users.name.desc(), Users.id.asc()).all() #desc从大到小 asc从小到大排
# 11. group by
from sqlalchemy.sql import func
# ret = session.query(
# Users.depart_id,
# func.count(Users.id),
# ).group_by(Users.depart_id).all()
# for item in ret:
# print(item)
#
# from sqlalchemy.sql import func
#进行二次筛选，只能用having
# ret = session.query(
# Users.depart_id,
# func.count(Users.id),
# ).group_by(Users.depart_id).having(func.count(Users.id) >= 2).all()
# for item in ret:
# print(item)
# 12.union 去重 和 union all 上下拼接不去重
"""
select id,name from users
UNION
select id,name from users;
"""
# q1 = session.query(Users.name).filter(Users.id > 2)
# q2 = session.query(Users.name).filter(Users.id > 2)
# ret = q1.union(q2).all()
#
# q1 = session.query(Users.name).filter(Users.id > 2)
# q2 = session.query(Users.name).filter(Users.id > 2)
# ret = q1.union_all(q2).all()
session.close()
from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship
#按照上面配置的代码加上这些东西，配置的表删除
class Depart(Base):
__tablename__ = ‘depart‘
id = Column(Integer, primary_key=True)
title = Column(String(32), index=True, nullable=False)
class Users(Base):
__tablename__ = ‘users‘
id = Column(Integer, primary_key=True)
name = Column(String(32), index=True, nullable=False)
depart_id = Column(Integer,ForeignKey("depart.id"))
#跟表结构无关
dp = relationship("Depart", backref=‘pers‘)
#django的manytomany 在flask中两个ForeignKey完成
from duoduo.test import Users,Depart,Student,Course,Student2Course
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
engine = create_engine(
"mysql+pymysql://root:123@127.0.0.1:3306/duoduo123?charset=utf8",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
SessionFactory = sessionmaker(bind=engine)
#根据Users类对user表进行增删改查
session=SessionFactory()
# #1、查询所有用户
# # ret =session.query(Users).all()
# #
# # for i in ret:
# # print(i.id,i.name,i.depart_id)
#2、查询所有用户,所属部门名称
# ret=session.query(Users.id,Users.name,Depart.title).join(Depart,User.depart_id==Depart.id).all()
# #这里有默认ForeignKey ,User.depart_id==Depart.id 可以不加也行
# for i in ret:
# print(i.id ,i.name,i.title)
#SELECT users.id AS users_id, users.name AS users_name, depart.title AS depart_title
#FROM users INNER JOIN depart ON depart.id = users.depart_id
# q=session.query(Users.id,Users.name,Depart.title).join(Depart)
# print(q)
#3、relation字段：查询所有用户+所有部门名称
# ret=session.query(Users).all()
# for row in ret:
# print(row.id,row.name,row.depart_id,row.dp.title)
#4、relation字段：查询销售部所有人员
# obj=session.query(Depart).filter(Depart.title ==‘大娃‘).first()
# # for i in obj.pers:
# # print(i.id,i.name,obj.title)
#5、创建一个名称叫：IT部门，再在该部门中添加一个员工叫：多多
#方式一：
# d1=Depart(title=‘IT‘)
# session.add(d1)
# session.commit()
#
# u1=Users(name=‘duoduo‘,depart_id=d1.id)
# session.add(u1)
# session.commit()
#方式二：
# u1=Users(name=‘多多1‘,dp=Depart(title=‘IT‘))
# session.add(u1)
# session.commit()
#6、创建一个部门叫王者荣耀，这个部门添加多个员工：亚瑟，后裔，貂蝉
# d1=Depart(title=‘王者荣耀‘)
# d1.pers=[Users(name=‘亚瑟‘),Users(name=‘后裔‘),Users(name=‘貂蝉‘)]
#
# session.add(d1)
# session.commit()
#1、录入数据
# session.add_all([
# Student(name=‘大娃‘),
# Student(name=‘二娃‘),
# Course(title=‘物理‘),
# Course(title=‘化学‘),
#
# ])
# session.add_all([
# Student2Course(student_id=2,course_id=1),
# # Student2Course(student_id=1,course_id=2)
# ]
# )
#2、查每个人选了课程的名称，三张表进行关联
# ret=session.query(Student2Course.id,Student.name,Course.title).join(Student,Student2Course.student_id==Student.id,isouter=True).join(Course,Student2Course.course_id==Course.id,isouter=True).order_by(Student2Course.id.asc())
# for i in ret:
# print(i)
#3、‘大娃‘所选的所有课
# ret=session.query(Student2Course.id,Student.name,Course.title).join(Student,Student2Course.student_id==Student.id,isouter=True).join(Course,Student2Course.course_id==Course.id,isouter=True).filter(Student.name==‘大娃‘).order_by(Student2Course.id.asc())
#
# for i in ret:
# print(i)
# obj=session.query(Student).filter(Student.name==‘大娃‘).first()
# for item in obj.course_list:
# print(item.title)
#4选了‘化学‘课程的所有人的名字
# obj=session.query(Course).filter(Course.title==‘化学‘).first()
# for item in obj.student_list:
# print(item.name)
#创建一个课程，创建2个学，两个学生选新创建的课程
# obj=Course(title=‘体育‘)
# obj.student_list=[Student(name=‘五娃‘),Student(name=‘六娃‘)]
#
# session.add(obj)
# session.commit()
# session.close()
class Student(Base):
__tablename__ = ‘student‘
id = Column(Integer, primary_key=True)
name = Column(String(32), index=True, nullable=False)
course_list = relationship(‘Course‘, secondary=‘student2course‘, backref=‘student_list‘)
class Course(Base):
__tablename__ = ‘course‘
id = Column(Integer, primary_key=True)
title = Column(String(32), index=True, nullable=False)
class Student2Course(Base):
__tablename__ = ‘student2course‘
id = Column(Integer, primary_key=True, autoincrement=True)
student_id = Column(Integer, ForeignKey(‘student.id‘))
course_id = Column(Integer, ForeignKey(‘course.id‘))
__table_args__ = (
UniqueConstraint(‘student_id‘, ‘course_id‘, name=‘uix_stu_cou‘), # 联合唯一索引
# Index(‘ix_id_name‘, ‘name‘, ‘extra‘), # 联合索引
)
from duoduo.test import Users,Depart,Student,Course,Student2Course
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
engine = create_engine(
"mysql+pymysql://root:123@127.0.0.1:3306/duoduo123?charset=utf8",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
SessionFactory = sessionmaker(bind=engine)
#连接
#第一种方式
# #并发
# def task():
# #去连接池获取一个连接
# session=SessionFactory()
# ret=session.query(Student).all()
# print(ret)
# #将连接交还给连接池
# session.close()
#
#
# from threading import Thread
#
# for i in range(20):
# t=Thread(target=task)
# t.start()
#
#第二种方式
from sqlalchemy.orm import scoped_session
session=scoped_session(SessionFactory)
def task():
ret=session.query(Student).all()
print(ret)
session.remove() #连接断开
from threading import Thread
for i in range(20):
t=Thread(target=task)
t.start()
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session
from models import Student,Course,Student2Course
engine = create_engine(
"mysql+pymysql://root:123@127.0.0.1:3306/duoduo123?charset=utf8",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
SessionFactory = sessionmaker(bind=engine)
session = scoped_session(SessionFactory)
def task():
""""""
# 方式一：
"""
# 查询
# cursor = session.execute(‘select * from users‘)
# result = cursor.fetchall()
# 添加
cursor = session.execute(‘INSERT INTO users(name) VALUES(:value)‘, params={"value": ‘duoduo‘})
session.commit()
print(cursor.lastrowid)
"""
# 方式二：
"""
# conn = engine.raw_connection()
# cursor = conn.cursor()
# cursor.execute(
# "select * from t1"
# )
# result = cursor.fetchall()
# cursor.close()
# conn.close()
"""
# 将连接交还给连接池
session.remove()
from threading import Thread
for i in range(20):
t = Thread(target=task)
t.start()
二、Flask的第三方组件
1、flask-sqlalchemy
pip3 install flask-sqlalchemy
#第一步 ：导入并实例化SQLALchemy
from flask_sqlalchemy import SQLAlchemy
db=SQLAlchemy()
#一定要在蓝图导入的上面，是全局变量
#也要导入表的models.py的所有表db.init_app(app)
#在注册蓝图的地方的下面
# ##### SQLALchemy配置文件 #####
SQLALCHEMY_DATABASE_URI = "mysql+pymysql://root:123@127.0.0.1:3306/duoduo?charset=utf8"
SQLALCHEMY_POOL_SIZE = 10
SQLALCHEMY_MAX_OVERFLOW = 5
from sqlalchemy import Column
from sqlalchemy import Integer,String,Text,Date,DateTime
from （项目目录） import db
class Users(db.Model): #继承的db.Model
__tablename__ = ‘users‘
id = Column(Integer, primary_key=True)
name = Column(String(32), index=True, nullable=False)
# depart_id = Column(Integer)
from （项目目录） import db,create_app
app = create_app()
app_ctx = app.app_context() # app_ctx = app/g
with app_ctx: # __enter__,通过LocalStack放入Local中
db.create_all() # 调用LocalStack放入Local中获取app，再去app中获取配置
class Foo(object):
def __enter__(self):
print(‘进入‘)
def __exit__(self, exc_type, exc_val, exc_tb):
print(‘出来‘)
obj =Foo()
with obj: #with Foo()
print(‘多多‘)
#结果
#进入
#多多
#出来
from flask import Blueprint
from （项目目录） import db
from （项目目录） import models
us = Blueprint(‘us‘,__name__)
@us.route(‘/index‘)
def index():
# 使用SQLAlchemy在数据库中插入一条数据
# db.session.add(models.Users(name=‘多多‘,depart_id=1)) #插入数据
# db.session.commit()
# db.session.remove()
result = db.session.query(models.Users).all()
print(result)
db.session.remove()
return ‘Index‘
2、 flask-script
pip3 install flask-script
from （项目目录） import create_app
from flask_script import Manager
app = create_app()
manager = Manager(app)
if __name__ == ‘__main__‘:
# app.run()
manager.run()
from （项目目录） import create_app
from flask_script import Manager
app = create_app()
manager = Manager(app)
@manager.command
def custom(arg):
"""
自定义命令
python manage.py custom 123
:param arg:
:return:
"""
print(arg)
if __name__ == ‘__main__‘:
# app.run()
manager.run()
from flask_script import Manager
from (项目目录) import create_app
app = create_app()
manager = Manager(app)
@manager.option(‘-n‘, ‘--name‘, dest=‘name‘)
@manager.option(‘-u‘, ‘--url‘, dest=‘url‘)
def cmd(name, url):
"""
自定义命令
执行： python manage.py cmd -n qianduoduo -u http://www.baidu.com
:param name:
:param url:
:return:
"""
print(name, url)
if __name__ == ‘__main__‘:
# app.run()
manager.run()
3、flask-migrate
pip3 install flask-migrate
from flask_script import Manager
from flask_migrate import Migrate, MigrateCommand
from （项目目录） import create_app
from （项目目录） import db
app = create_app()
manager = Manager(app)
Migrate(app, db)
"""
# 数据库迁移命名
python manage.py db init #只执行一次
python manage.py db migrate # makemirations
python manage.py db upgrade # migrate
"""
manager.add_command(‘db‘, MigrateCommand)
if __name__ == ‘__main__‘:
# app.run()
manager.run()
补充工具：（协同开发的保证开发的环境一致性）
pip3 install pipreqs
pipreqs ./ --encoding=utf-8
pip3 install virtualenv
#找一个存放虚拟环境的文件夹
virtualenv env1 --no-site-packages 创建环境
activate #激活环境
deactivate # 退出环境
#也可以用pycharm鼠标点点就好了，最新的版本就有虚拟环境goon功能
#当我们需要特定的环境，可以在电脑上有多高开发环境