engine.py

# -*- encoding: utf-8 -*-
# -----------------------------------------------------------------------------
# @File Name  : app.py
# @Time       : 2020/11/18 下午12:45
# @Author     : X. Peng
# @Email      : acepengxiong@163.com
# @Software   : PyCharm
# -----------------------------------------------------------------------------


# import logging
import redis
import os
import sys
import yaml
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from sqlalchemy.pool import NullPool

env = sys.argv[-1] if sys.argv[-1] in ["dev", "test", "prod"] else "prod"
work_dir = os.getcwd()
CFG_FILEPATH = work_dir + '/app/config/config.yaml'

config = yaml.load(open(CFG_FILEPATH, 'r'), Loader=yaml.FullLoader)

temp_img_save_folder = work_dir + '/app/templates/temp_img/productionenv/ReportTempImg/'
pdf_save_folder = work_dir + '/app/pdf/productionenv/AppHoldReport/'
template_folder = config[env]['oss']['imgs_url_prefix']
pdf_folder = config[env]['oss']['pdf_url_prefix']

tamp_product_engine = create_engine(
    'mysql+mysqldb://{user}:{password}@{host}:{port}/{db}?charset={charset}'.format(
        db=config[env]['MySQL']['tamp_product_db'],
        host=config[env]['MySQL']['host'],
        port=config[env]['MySQL']['port'],
        user=config[env]['MySQL']['user'],
        password=config[env]['MySQL']['password'],
        charset="utf8"
    ),
    poolclass=NullPool
)
tamp_order_engine = create_engine(
    'mysql+mysqldb://{user}:{password}@{host}:{port}/{db}?charset={charset}'.format(
        db=config[env]['MySQL']['tamp_order_db'],
        host=config[env]['MySQL']['host'],
        port=config[env]['MySQL']['port'],
        user=config[env]['MySQL']['user'],
        password=config[env]['MySQL']['password'],
        charset="utf8"
    ),
    poolclass=NullPool
)

tamp_user_engine = create_engine(
    'mysql+mysqldb://{user}:{password}@{host}:{port}/{db}?charset={charset}'.format(
        db=config[env]['MySQL']['tamp_user_db'],
        host=config[env]['MySQL']['host'],
        port=config[env]['MySQL']['port'],
        user=config[env]['MySQL']['user'],
        password=config[env]['MySQL']['password'],
        charset="utf8"
    ),
    poolclass=NullPool
)

tamp_fund_engine = create_engine(
    'mysql+mysqldb://{user}:{password}@{host}:{port}/{db}?charset={charset}'.format(
        db=config[env]['MySQL']['tamp_fund_db'],
        host=config[env]['MySQL']['host'],
        port=config[env]['MySQL']['port'],
        user=config[env]['MySQL']['user'],
        password=config[env]['MySQL']['password'],
        charset="utf8"
    ),
    poolclass=NullPool
)

tamp_diagnose_app_engine = create_engine(
    'mysql+mysqldb://{user}:{password}@{host}:{port}/{db}?charset={charset}'.format(
        db=config[env]['MySQL']['tamp_diagnose_app_db'],
        host=config[env]['MySQL']['host'],
        port=config[env]['MySQL']['port'],
        user=config[env]['MySQL']['user'],
        password=config[env]['MySQL']['password'],
        charset="utf8"
    ),
    poolclass=NullPool
)


redis = redis.StrictRedis(
    host=config[env]['redis']['host'],
    port=config[env]['redis']['port'],
    db=config[env]['redis']['db'],
    password=config[env]['redis']['password']
)

redis_host=config[env]['celery']['host'],
redis_port=config[env]['celery']['port'],
redis_db=config[env]['celery']['db'],
redis_password=config[env]['celery']['password']

# logging.basicConfig(level=logging.INFO,
#                     filename=work_dir + config[env]['log']['filename'],
#                     filemode=config[env]['log']['filemode'],
#                     format=config[env]['log']['format'],
#                     datefmt=config[env]['log']['datefmt']
#                     )

class TAMP_SQL(object):
    """[sqlalchemy 封装]

    Args:
        object ([type]): [description]

    Returns:
        [type]: [description]
    """

    def __init__(self, db_engine):
        # 创建DBSession类型:
        self.DBSession = scoped_session(sessionmaker(bind=db_engine))
        self.session = self.DBSession()

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        try:
            self.session.commit()
        except:
            self.session.rollback()
        finally:
            self.session.close()