flask-离线脚本with在上下文的应用

Posted 2021-02-12 weidaijie

流程图

注意：

将SQLAlchemy相关的所有功能都封装到db=flask_sqlalchemy.SQLAlchemy()对象中，

- 创建表
class User(db.Model):
pass

- 操作表
db.session

核心就是配置，通过db对象，操作models，为蓝图提供数据！

现在有一个需求，需要将数据库中的表删除或者生成数据库中的表，必须通过脚本来完成！

配置文件加载之后，将setttings.py中的属性添加到app.config对象中。如果有app对象，那么就可以得到以下信息：

- 应用上下文中的有：app/g 
- flask的配置文件：app.config中 
- app中包含了：SQLAlchemy相关的数据。

 离线脚本编写的目录结构

 

__init__.py

from flask import Flask
from flask_session import Session

# 第一步：导入并实例化SQLAlchemy
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()

from .views.account import ac
from .views.user import us

from .models import *

def create_app():
    app = Flask(__name__)
    app.config.from_object(‘settings.ProConfig‘)

    app.register_blueprint(ac)
    app.register_blueprint(us)

    # Flask-Session: 第一步示例Session
    # Session(app)

    # 第三步：依赖app中的配置文件
    db.init_app(app)

    return app

 settings.py

    # ##### SQLALchemy配置文件 #####
    SQLALCHEMY_DATABASE_URI = "mysql+pymysql://root:[email protected]:3306/s9day122?charset=utf8"
    SQLALCHEMY_POOL_SIZE = 10
    SQLALCHEMY_MAX_OVERFLOW = 5

    SQLALCHEMY_TRACK_MODIFICATIONS = False

 models.py

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column
from sqlalchemy import Integer,String,Text,Date,DateTime
from sqlalchemy import create_engine
from chun import db


class Users(db.Model):
    __tablename__ = ‘users‘

    id = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=False)
    # depart_id = Column(Integer)

 在视图函数中使用SQLAlchemy操作数据库

from flask import Blueprint
from chun import db
from chun 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‘

 编写离线脚本

 

#新建create_tablepy
from chun 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中获取配置

db就我们在__init__文件中实例化的对象，它包含了create_all（创建表）和drop_all（删除表）的命令，但是由于在使用db时我们需要用到app中关于数据库的配置（从上下文中取），但是这时项目没有运行，没有请求，在local类中没有app的内容，所以我们使用with

方法，利用上下文管理，将需要的内容添加到loacl对象中

关于with的使用

class Foo(object):
    def __enter__(self):
        ‘‘‘进入‘‘‘
        print("执行————enter")

    def __exit__(self,*args ):
        ‘‘‘退出‘‘‘
        print("执行——————exit")

obj = Foo()
with obj:
    print("我要执行的内容")

结果：
执行————enter
我要执行的内容
执行——————exit

 flask中with源码

   def __enter__(self):
        self.push()
        return self

    def __exit__(self, exc_type, exc_value, tb):
        self.pop(exc_value)

        if BROKEN_PYPY_CTXMGR_EXIT and exc_type is not None:
            reraise(exc_type, exc_value, tb)

 在上面的离线脚本中，进入时其实就是在__enter__中执行了push方法，将appcontext的对象放入到local中，而退出时则是在__exit__中执行了pop方法，将对象从local中删除

补充SQLAlchemy两种创建session的方式

方式一：
    import models
    from threading import Thread
    from sqlalchemy.orm import sessionmaker
    from sqlalchemy import create_engine

    engine =create_engine("mysql+pymysql://root:[email protected]:3306/s8day128db?charset=utf8",pool_size=2,max_overflow=0)
    XXXXXX = sessionmaker(bind=engine)

    def task():
        from sqlalchemy.orm.session import Session
        session = XXXXXX()

        data = session.query(models.Classes).all()
        print(data)

        session.close()

    for i in range(10):
        t = Thread(target=task)
        t.start()

方式二（推荐）：
    import models
    from threading import Thread
    from sqlalchemy.orm import sessionmaker
    from sqlalchemy import create_engine
    from sqlalchemy.orm import scoped_session

    engine =create_engine("mysql+pymysql://root:[email protected]:3306/s8day128db?charset=utf8",pool_size=2,max_overflow=0)
    XXXXXX = sessionmaker(bind=engine)

    session = scoped_session(XXXXXX)
    
    
    def task():

        # 1. 原来的session对象 = 执行session.registry()
        # 2. 原来session对象.query
        data = session.query(models.Classes).all()
        print(data)
        session.remove()

        

    for i in range(10):
        t = Thread(target=task)
        t.start()

 

方式二的内部其实就利用threading.loacl为每个线程单独创建了一个session

flask-session默认也是使用的第二种方式：scoped_session