Flask 中使用SQLAlchemy創(chuàng)建臨時表

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, DateTime, Table, MetaData, BigInteger

Base = declarative_base()
class TempTable(Base):
    '''
    檢查部分時間有效的臨時表
    '''
    __tablename__ = 'temp_table'
    __table_args__ = {'prefixes': ['TEMPORARY']}
    id = Column(BigInteger, autoincrement=True, primary_key=True)
    tag_id = Column(String(36), nullable=False)
    hours = Column(String(2), nullable=False)

    def __repr__(self):
        return 'tag_id: %r' % self.tag_id

DATABASE_URI='mysql+pymysql://root:123123@127.0.0.1:3306/test?charset=utf8&connect_timeout=300'
engine = create_engine(DATABASE_URI)
Session = scoped_session(sessionmaker(bind=engine))

try:
    TempTable.__table__.create(Session.connection())      #創(chuàng)建臨時表
except Exception as err:
    logger.error(err)

try:
    TempTable.__table__.drop(Session.connection())    # 刪除臨時表，可以不用，會話完成時臨時表自動刪除。
    Session.commit()
except Exception as err:
    logger.exception(err)
    Session.rollback()

• 創(chuàng)建臨時表時使用：Session.connection()，使用Session.bind，是創(chuàng)建一個新會話，MySQL臨時表基于會話，可能存在 table not exists