python-fastapi-mm-zhcs-yj-api/config.py

# -*- coding: utf-8 -*-
from pydantic import BaseSettings, Field
from functools import lru_cache
from typing import Optional
from dotenv.main import DotEnv
from extensions import logger
from contextlib import contextmanager
import os
import io
from io import StringIO

def my_get_stream(self):
    if isinstance(self.dotenv_path, StringIO):
        yield self.dotenv_path
    elif os.path.isfile(self.dotenv_path):
        with io.open(self.dotenv_path, encoding='utf-8') as stream:
            yield stream
    else:
        if self.verbose:
            logger.warning("File doesn't exist %s", self.dotenv_path)
        yield StringIO('')


DotEnv._get_stream = contextmanager(my_get_stream)

class Settings(BaseSettings):

    # 系统环境
    ENVIRONMENT: Optional[str] = Field(None, env="ENVIRONMENT")

    SECRET_KEY: str = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
    ALGORITHM: str = "HS256"

    IS_DEV: bool = False
    IS_STAGE: bool = False
    IS_PROD: bool = False

    REDIS_DB_URL = {
        'host': '127.0.0.1',
        'port': 6379,
        'password': 'c0b0Info@)!%',
        'db': 0
    }

    class Config:
        env_file = ".env"


class DevConfig(Settings):

    IS_DEV: bool = True
    IS_STAGE: bool = False
    IS_PROD: bool = False

    MYSQL_SERVER: Optional[str] = Field(..., env="DEV_MYSQL_SERVER")
    MYSQL_USER: Optional[str] = Field(..., env="DEV_MYSQL_USER")
    MYSQL_PASSWORD: Optional[str] = Field(..., env="DEV_MYSQL_PASSWORD")
    MYSQL_DB_NAME: Optional[str] = Field(..., env="DEV_MYSQL_DB_NAME")
    MYSQL_PORT: Optional[int] = Field(..., env="DEV_MYSQL_PORT")

    YJHTGL_WEB_ROOT_PATH: Optional[str] = Field(..., env="DEV_YJHTGL_WEB_ROOT_PATH")
    YJXP_WEB_ROOT_PATH: Optional[str] = Field(..., env="DEV_YJXP_WEB_ROOT_PATH")
    YJXP_CALLBACK_WEB_PATH: Optional[str] = Field(..., env="DEV_YJXP_CALLBACK_WEB_PATH")

class ProdConfig(Settings):

    IS_DEV: bool = False
    IS_STAGE: bool = False
    IS_PROD: bool = True

    MYSQL_SERVER: Optional[str] = Field(..., env="PROD_MYSQL_SERVER")
    MYSQL_USER: Optional[str] = Field(..., env="PROD_MYSQL_USER")
    MYSQL_PASSWORD: Optional[str] = Field(..., env="PROD_MYSQL_PASSWORD")
    MYSQL_DB_NAME: Optional[str] = Field(..., env="PROD_MYSQL_DB_NAME")
    MYSQL_PORT: Optional[int] = Field(..., env="PROD_MYSQL_PORT")

    YJHTGL_WEB_ROOT_PATH: Optional[str] = Field(..., env="PROD_YJHTGL_WEB_ROOT_PATH")
    YJXP_WEB_ROOT_PATH: Optional[str] = Field(..., env="PROD_YJXP_WEB_ROOT_PATH")
    YJXP_CALLBACK_WEB_PATH: Optional[str] = Field(..., env="PROD_YJXP_CALLBACK_WEB_PATH")


    REDIS_DB_URL = {
        'host': '127.0.0.1',
        'port': 6379,
        'password': 'c0b0Info@)!%',
        'db': 0
    }

class StageConfig(Settings):

    IS_DEV: bool = False
    IS_STAGE: bool = True
    IS_PROD: bool = False

    MYSQL_SERVER: Optional[str] = Field(..., env="STAGE_MYSQL_SERVER")
    MYSQL_USER: Optional[str] = Field(..., env="STAGE_MYSQL_USER")
    MYSQL_PASSWORD: Optional[str] = Field(..., env="STAGE_MYSQL_PASSWORD")
    MYSQL_DB_NAME: Optional[str] = Field(..., env="STAGE_MYSQL_DB_NAME")
    MYSQL_PORT: Optional[int] = Field(..., env="STAGE_MYSQL_PORT")

    YJHTGL_WEB_ROOT_PATH: Optional[str] = Field(..., env="STAGE_YJHTGL_WEB_ROOT_PATH")
    YJXP_WEB_ROOT_PATH: Optional[str] = Field(..., env="STAGE_YJXP_WEB_ROOT_PATH")
    YJXP_CALLBACK_WEB_PATH: Optional[str] = Field(..., env="STAGE_YJXP_CALLBACK_WEB_PATH")

    REDIS_DB_URL = {
        'host': '127.0.0.1',
        'port': 6379,
        'password': 'c0b0Info@)!%',
        'db': 0
    }


class FactoryConfig:
    """Returns a config instance dependending on the ENV_STATE variable."""

    def __init__(self, env_state: Optional[str]):
        self.env_state = env_state

    def __call__(self):

        if self.env_state == "development":
            return DevConfig()

        elif self.env_state == "production":
            return ProdConfig()
    
        elif self.env_state == "stage":
            return StageConfig()

        
        else:
            return DevConfig()


@lru_cache()
def get_configs():
    from dotenv import load_dotenv
    load_dotenv(encoding='utf-8')
    return FactoryConfig(Settings().ENVIRONMENT)()

settings = get_configs()