exporter-ex/DBSession.py

from sqlalchemy import create_engine ,Integer,String,Column
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from contextlib import contextmanager

HOST = 'rm-bp1k5853td1r25g3n690.mysql.rds.aliyuncs.com'
PORT = '3306'
DATABASE = 'devops'
USERNAME = 'devops'
PASSWORD = 'devops@123456'
MYSQL_URI = "mysql+pymysql://{username}:{password}@{host}:{port}/{db}?charset=utf8".format(username=USERNAME,
                                                                                           password=PASSWORD,
                                                                                           host=HOST, port=PORT,
                                                                                           db=DATABASE)

Base = declarative_base()


engine = create_engine(MYSQL_URI)
session_factory = sessionmaker(bind=engine)



@contextmanager
def session_maker(session=None):
    try:
        if session is None:
            session = session_factory()
        yield session
    except:
        session.rollback()
        raise
    else:
        session.commit()
    finally:
        session.close()