env_config.py
3.02 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
"""
Compatibility accessors for infrastructure/runtime environment settings.
All values are derived from the unified application config. This module no
longer owns any independent loading or precedence rules.
"""
from __future__ import annotations
from typing import Any, Dict
from config.loader import get_app_config
from config.services_config import get_rerank_service_url
def _app():
return get_app_config()
def _runtime():
return _app().runtime
def _infra():
return _app().infrastructure
def _elasticsearch_dict() -> Dict[str, Any]:
cfg = _infra().elasticsearch
return {
"host": cfg.host,
"username": cfg.username,
"password": cfg.password,
}
def _redis_dict() -> Dict[str, Any]:
cfg = _infra().redis
return {
"host": cfg.host,
"port": cfg.port,
"snapshot_db": cfg.snapshot_db,
"password": cfg.password,
"socket_timeout": cfg.socket_timeout,
"socket_connect_timeout": cfg.socket_connect_timeout,
"retry_on_timeout": cfg.retry_on_timeout,
"cache_expire_days": cfg.cache_expire_days,
"embedding_cache_prefix": cfg.embedding_cache_prefix,
}
def _db_dict() -> Dict[str, Any]:
cfg = _infra().database
return {
"host": cfg.host,
"port": cfg.port,
"database": cfg.database,
"username": cfg.username,
"password": cfg.password,
}
ES_CONFIG = _elasticsearch_dict()
REDIS_CONFIG = _redis_dict()
DB_CONFIG = _db_dict()
RUNTIME_ENV = _runtime().environment
ES_INDEX_NAMESPACE = _runtime().index_namespace
DEEPL_AUTH_KEY = _infra().secrets.deepl_auth_key
DASHSCOPE_API_KEY = _infra().secrets.dashscope_api_key
API_HOST = _runtime().api_host
API_PORT = _runtime().api_port
INDEXER_HOST = _runtime().indexer_host
INDEXER_PORT = _runtime().indexer_port
EMBEDDING_HOST = _runtime().embedding_host
EMBEDDING_PORT = _runtime().embedding_port
EMBEDDING_TEXT_HOST = _runtime().embedding_host
EMBEDDING_TEXT_PORT = _runtime().embedding_text_port
EMBEDDING_IMAGE_HOST = _runtime().embedding_host
EMBEDDING_IMAGE_PORT = _runtime().embedding_image_port
TRANSLATION_HOST = _runtime().translator_host
TRANSLATION_PORT = _runtime().translator_port
RERANKER_HOST = _runtime().reranker_host
RERANKER_PORT = _runtime().reranker_port
API_BASE_URL = f"http://localhost:{API_PORT}" if API_HOST == "0.0.0.0" else f"http://{API_HOST}:{API_PORT}"
INDEXER_BASE_URL = (
f"http://localhost:{INDEXER_PORT}" if INDEXER_HOST == "0.0.0.0" else f"http://{INDEXER_HOST}:{INDEXER_PORT}"
)
EMBEDDING_TEXT_SERVICE_URL = _app().services.embedding.get_provider_config().get("text_base_url")
EMBEDDING_IMAGE_SERVICE_URL = _app().services.embedding.get_provider_config().get("image_base_url")
RERANKER_SERVICE_URL = get_rerank_service_url()
def get_es_config() -> Dict[str, Any]:
return dict(ES_CONFIG)
def get_db_config() -> Dict[str, Any]:
return dict(DB_CONFIG)
def get_redis_config() -> Dict[str, Any]:
return dict(REDIS_CONFIG)
def print_config() -> None:
config = _app().sanitized_dict()
print(config)