python-base--/fastapi-demo/config/__init__.py

# -*- coding:utf-8 -*-
# 配置文件初始化
# 主要是一些 db 的初始化创建
# 简单 Class 类函数的 初始化

import asyncio
import logging
import os
from configparser import ConfigParser
from logging.handlers import TimedRotatingFileHandler
from os.path import dirname

import redis as redis
import uvloop
from requests import adapters
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

adapters.DEFAULT_POOLSIZE = 100000

asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())


class CfgBaseInit(object):
    """
    config: 配置文件对象
    project_name: 项目名称
    start_mode: 项目启动模式
    """
    __REC_ENGINE = "REC_ENGINE"  # 环境变量以REC_ENGINE开头

    config = ConfigParser()
    config.read(os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "config", "config.ini"))

    # 遍历所有配置，保存为dict
    config_dict = {}
    for sc in config.sections():  # 循环每个Session
        items = config.items(sc)  # 返回结果为元组
        config_dict[sc] = {}
        for it in items:  # 循环遍历 每个属性
            k, v = it[0], it[1]
            config_value = os.getenv("_".join([__REC_ENGINE, sc, k]))  # 环境变量中是否存在
            if config_value:
                print(f"获取到环境变量：{sc}_{k}")
                v = config_value
            config_dict[sc][k] = v

    project_name = config.get("CONFIG", "project_name")
    project_path = dirname(dirname(__file__))
    bind = config.get("CONFIG", "host")


class Logger(object):
    # 日志
    log_path = CfgBaseInit.config.get("LOG", "log_path")
    when = CfgBaseInit.config.get("LOG", "when")
    backupCount = int(CfgBaseInit.config.get("LOG", "backupCount"))

    logging.basicConfig(
        format='%(asctime)s %(name)s %(process)d %(thread)d %(threadName)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s')
    logger = logging.getLogger(CfgBaseInit.project_name)
    logger.setLevel(logging.DEBUG)

    # 输出到文件
    handler = TimedRotatingFileHandler(log_path, when=when, backupCount=backupCount)
    logger.addHandler(handler)


logger = Logger.logger


class MySqlEngine(CfgBaseInit):
    """ mysql 引擎
    """
    import pymysql
    from urllib.parse import quote_plus as urlquote
    pymysql.install_as_MySQLdb()

    mysql_dict = CfgBaseInit.config_dict["MYSQL"]

    __user = mysql_dict['user']
    __password = mysql_dict['password']
    __host = mysql_dict['host']
    __port = mysql_dict['port']
    __db = mysql_dict['db']
    __engine = create_engine(f"mysql://{__user}:{urlquote(__password)}@{__host}:{__port}/{__db}",
                             pool_size=2,
                             max_overflow=0,
                             pool_recycle=280)  # 280秒重新连接一次
    DBSession = sessionmaker(bind=__engine)


class RedisInit(CfgBaseInit):
    """
        Redis 对象初始化
    """
    __redis_dict = CfgBaseInit.config_dict["REDIS"]
    __host = __redis_dict['host']
    __port = __redis_dict['port']
    __db = __redis_dict['db']
    pool = redis.ConnectionPool(host=__host, port=__port, db=__db, max_connections=None, decode_responses=True,
                                socket_keepalive=False)
    redis = redis.Redis(connection_pool=pool)