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
