lavkach3

1
import logging
2
import os
3

4
from dotenv import load_dotenv
5
from pydantic import BaseConfig
6
from pydantic_settings import BaseSettings
7

8
BaseConfig.arbitrary_types_allowed = True
9
logging.basicConfig(level=logging.INFO)
10
cwd = os.getcwd()
11
env_path = os.path.join(cwd, os.environ.get('DOTENV_PATH') or '.env')
12
load_dotenv(env_path)
13

14

15
class Config(BaseSettings):
16
    BROKER_PATH: str = 'false'
17
    ENV: str = os.environ.get("ENV") or 'local'
18
    DEBUG: bool = os.environ.get("DEBUG") or True
19
    APP_HOST: str = os.environ.get("APP_HOST") or 'localhost'
20
    APP_PORT: int = os.environ.get("APP_PORT") or '8000'
21
    DB_HOST: str = os.environ.get("DB_HOST") or 'localhost'
22
    DB_HOST_TEST: str = os.environ.get("DB_TEST") or 'localhost'
23
    DB_PORT: str = os.environ.get("DB_PORT") or '5433'
24
    DB_PORT_TEST: str = os.environ.get("DB_PORT_TEST") or '5433'
25
    DB_NAME: str = os.environ.get("DB_NAME") or 'lavkach'
26
    DB_NAME_TEST: str = os.environ.get("DB_NAME_TEST") or 'lavkach2'
27
    DB_USER: str = os.environ.get("DB_USER") or 'taxi'
28
    DB_USER_TEST: str = os.environ.get("DB_USER_TEST") or 'taxi'
29
    DB_PASS: str = os.environ.get("DB_PASS") or 'test'
30
    DB_PASS_TEST: str = os.environ.get("DB_PASS_TEST") or 'test'
31
    DB_SSL: str = os.environ.get('DB_SSL') or 'prefer'
32
    POSTGRES_TEST_DATABASE_URL: str = f'postgresql+asyncpg://{DB_USER_TEST}:{DB_PASS_TEST}@{DB_HOST_TEST}/postgres'
33
    TEST_WRITER_DB_URL: str = f'postgresql+asyncpg://{DB_USER_TEST}:{DB_PASS_TEST}@{DB_HOST_TEST}:{DB_PORT_TEST}/{DB_NAME_TEST}'
34
    WRITER_DB_URL: str = f'postgresql+asyncpg://{DB_USER}:{DB_PASS}@{DB_HOST}:{DB_PORT}/{DB_NAME}?ssl={DB_SSL}'
35
    READER_DB_URL: str = f'postgresql+asyncpg://{DB_USER}:{DB_PASS}@{DB_HOST}:{DB_PORT}/{DB_NAME}?ssl={DB_SSL}'
36
    JWT_SECRET_KEY: str = os.environ.get("JWT_SECRET_KEY") or 'secret'
37
    JWT_ALGORITHM: str = os.environ.get("JWT_ALGORITHM") or 'HS256'
38
    CELERY_BROKER_URL: str = "amqp://user:bitnami@localhost:5672/"
39
    CELERY_BACKEND_URL: str = "redis://:password123@localhost:5370/0"
40
    REDIS_HOST: str = os.environ.get("REDIS_HOST") or 'localhost'
41
    REDIS_PORT: int = os.environ.get("REDIS_PORT") or '5370'
42
    REDIS_PASSWORD: str = os.environ.get("REDIS_PASSWORD") or ''
43
    REDIS_SSL: bool = os.environ.get("REDIS_SSL") or False
44
    REDIS_CERT_PATH: str = os.environ.get("REDIS_CERT_PATH") or ''
45
    AWS_DEFAULT_REGION: str = os.environ.get("AWS_DEFAULT_REGION") or 'us-east-1'
46
    AWS_ACCESS_KEY_ID: str = os.environ.get("AWS_ACCESS_KEY_ID") or ''
47
    AWS_SECRET_ACCESS_KEY: str = os.environ.get("AWS_SECRET_ACCESS_KEY") or ''
48
    AWS_DEFAULT_BUCKET: str = os.environ.get("AWS_DEFAULT_BUCKET") or 'us-east-1'
49
    AWS_ENDPOINT_URL: str = os.environ.get("AWS_ENDPOINT_URL") or 'https://storage.yandexcloud.net'
50
    SUPERUSER_EMAIL: str = os.environ.get('SUPERUSER_EMAIL') or ''
51
    SUPERUSER_PASSWORD: str = os.environ.get('SUPERUSER_PASSWORD') or ''
52
    BASE_HOST: str = os.environ.get('BASIC_HOST') or '127.0.0.1'
53
    BASE_PORT: str = os.environ.get('BASIC_PORT') or '8888'
54
    BUS_HOST: str = os.environ.get('BUS_HOST') or '127.0.0.1'
55
    BUS_PORT: str = os.environ.get('BUS_PORT') or '8099'
56
    MESSAGE_HOST: str = os.environ.get('MESSAGE_HOST') or '127.0.0.5'
57
    MESSAGE_PORT: str = os.environ.get('MESSAGE_PORT') or '8099'
58
    INTERCO_TOKEN: str = os.environ.get('INTERCO_TOKEN') or 'netu'
59

60

61
class DevelopmentConfig(Config):
62
    ...
63

64
class DockerConfig(Config):
65
    BUS_HOST: str = 'bus_app'
66
    BUS_PORT: str = '80'
67

68
class LocalConfig(Config):
69
    ...
70

71

72
class ProductionConfig(Config):
73
    ...
74

75

76
class ModuleEnv:
77
    ...
78

79

80
def my_import(name):
81
    """
82
    Метод для удобного импорта адаптеров по пути
83
    """
84
    components = name.split('.')
85
    mod = __import__(components[0])
86
    for comp in components[1:]:
87
        mod = getattr(mod, comp)
88
    return mod
89

90

91
def get_config():
92
    env = os.getenv("ENV", "local")
93
    for name, value in os.environ.items():
94
        logging.info("{0}: {1}".format(name, value))
95
        pass
96
    config_type = {
97
        "dev": DevelopmentConfig(),
98
        "local": LocalConfig(),
99
        "docker": DockerConfig(),
100
        "prod": ProductionConfig(),
101
    }
102
    return config_type[env]
103

104

105
config: Config = get_config()
106