使用sqlalchemy 创建临时表

Flask 中使用SQLAlchemy创建临时表

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())      #创建临时表
except Exception as err:
    logger.error(err)

try:
    TempTable.__table__.drop(Session.connection())    # 删除临时表，可以不用，会话完成时临时表自动删除。
    Session.commit()
except Exception as err:
    logger.exception(err)
    Session.rollback()

• 创建临时表时使用：Session.connection()，使用Session.bind，是创建一个新会话，MySQL临时表基于会话，可能存在 table not exists