python3-開發進階Flask的基礎(5)
阿新 • • 發佈：2018-08-28
下載安裝 ini extra www als 協同開發 通配 化學 first 
內容概要：
- SQLAlchemy
- flsak-sqlalchemy
- flask-script
- flask-migrate
- Flask的目錄結構
一、SQLAlchemy
1、概述
SQLAlchemy是一個ORM的框架，ORM就是關系對象映射，具體可以參照Django中的ORM。
作用：幫助我們使用類和對象快速實現數據庫操作
數據庫：
-原生：MYSQLdb pymysql
區別就是 MYSQLdb 不支持python3 pymysql 都支持
ORM框架
SQLAlchemy
2、SQLAlchemy用法
1、安裝
pip3 install sqlalchemy
2、配置
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:[email protected]: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:[email protected]: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()
3、增刪改查的演示
from duoduo.test import Users from sqlalchemy.orm import sessionmaker from sqlalchemy import create_engine engine = create_engine( "mysql+pymysql://root:[email protected]: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<=3).first()#> < >= <= # # 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()增刪改查
4、單表常用操作
# ############################## 其他常用 ############################### # 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 < 2, Users.name == ‘duoduo‘)).all() # session.query(Users).filter( # or_( # Users.id < 2, # 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()單表常用操作
5、連表常用操作
在上面的配置裏，重新搞兩給表
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:[email protected]: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()ForeignKey
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:[email protected]: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()
執行原生SQ:
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:[email protected]: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
a、先下載安裝
pip3 install flask-sqlalchemy
b、項目下的__init__.py導入
#第一步 ：導入並實例化SQLALchemy from flask_sqlalchemy import SQLAlchemy db=SQLAlchemy() #一定要在藍圖導入的上面，是全局變量
#也要導入表的models.py的所有表
c、初始化
db.init_app(app) #在註冊藍圖的地方的下面
d、在配置文件中寫入配置
# ##### SQLALchemy配置文件 ##### SQLALCHEMY_DATABASE_URI = "mysql+pymysql://root:[email protected]:3306/duoduo?charset=utf8" SQLALCHEMY_POOL_SIZE = 10 SQLALCHEMY_MAX_OVERFLOW = 5
e、創建models.py中的類（對應數據庫中的表）
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)
f、生成表（使用app上下文）
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(‘多多‘) #結果 #進入 #多多 #出來
g、基於ORM對數據庫進行操作
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‘
這裏面db.session.add的源碼：
2、 flask-script
下載安裝
pip3 install flask-script
功能：
a、增加runsever
from （項目目錄） import create_app from flask_script import Manager app = create_app() manager = Manager(app) if __name__ == ‘__main__‘: # app.run() manager.run()
b、位置傳參
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()
c、關鍵字傳參
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
配置是依賴 flask-script
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()
補充工具：（協同開發的保證開發的環境一致性）
1、pipreqs
找到項目使用的所有組件的版本
pip3 install pipreqs
在項目中的命令行
pipreqs ./ --encoding=utf-8
在項目中找了文件requirements.txt
如何裝那些模塊？
pip install 的時候可以指定一個文件，他會自己讀每一行，然後安裝
pycharm 打開文件的時候也會提示下載
2、虛擬環境 （開發環境版本不能共存的）
安裝
pip3 install virtualenv
創建虛擬環境
#找一個存放虛擬環境的文件夾 virtualenv env1 --no-site-packages 創建環境 activate #激活環境 deactivate # 退出環境 #也可以用pycharm鼠標點點就好了，最新的版本就有虛擬環境goon功能 #當我們需要特定的環境，可以在電腦上有多高開發環境
python3-開發進階Flask的基礎(5)