-
Notifications
You must be signed in to change notification settings - Fork 6
/
settings.py
156 lines (127 loc) · 5 KB
/
settings.py
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
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
import json
import logging
import logging.config
import os
from pathlib import Path
from typing import Literal
from httpx import ConnectError
from pydantic import Field, HttpUrl, field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
from ixmp4.core.exceptions import InvalidCredentials
from .auth import AnonymousAuth, ManagerAuth
from .credentials import Credentials
from .manager import ManagerConfig
from .toml import TomlConfig
from .user import local_user
logger = logging.getLogger(__name__)
here = Path(__file__).parent
class Settings(BaseSettings):
mode: Literal["production"] | Literal["development"] | Literal["debug"] = (
"production"
)
storage_directory: Path = Field(
"~/.local/share/ixmp4/", json_schema_extra={"env": "ixmp4_dir"}
)
secret_hs256: str = "default_secret_hs256"
migration_db_uri: str = "sqlite:///./run/db.sqlite"
manager_url: HttpUrl = Field("https://api.manager.ece.iiasa.ac.at/v1")
managed: bool = True
max_page_size: int = 10_000
default_page_size: int = 5_000
client_default_upload_chunk_size: int = 10_000
client_max_concurrent_requests: int = Field(2, lt=4)
client_max_request_retries: int = Field(3)
client_backoff_factor: int = Field(5)
client_timeout: int = Field(30)
model_config = SettingsConfigDict(env_prefix="ixmp4_", extra="allow")
def __init__(self, *args, **kwargs) -> None:
super().__init__(*args, **kwargs)
self.storage_directory.mkdir(parents=True, exist_ok=True)
self.database_dir = self.storage_directory / "databases"
self.database_dir.mkdir(exist_ok=True)
self.log_dir = self.storage_directory / "log"
self.log_dir.mkdir(exist_ok=True)
self.configure_logging(self.mode)
self._credentials = None
self._toml = None
self._default_auth = None
self._manager = None
logger.debug(f"Settings loaded: {self}")
@property
def credentials(self):
if self._credentials is None:
self.load_credentials()
return self._credentials
@property
def default_credentials(self):
try:
return self.credentials.get("default")
except KeyError:
pass
@property
def toml(self):
if self._toml is None:
self.load_toml_config()
return self._toml
@property
def default_auth(self):
if self._default_auth is None:
self.get_auth()
return self._default_auth
@property
def manager(self):
if self._manager is None:
self.load_manager_config()
return self._manager
def load_credentials(self):
credentials_config = self.storage_directory / "credentials.toml"
credentials_config.touch()
self._credentials = Credentials(credentials_config)
def get_auth(self):
if self.default_credentials is not None:
try:
self._default_auth = ManagerAuth(
*self.default_credentials, str(self.manager_url)
)
logger.info(
f"Connecting as user '{self._default_auth.get_user().username}'."
)
except InvalidCredentials:
logger.warning(f"Invalid credentials for {self.manager_url}.")
except ConnectError:
logger.warning(f"Unable to connect to {self.manager_url}.")
else:
self._default_auth = AnonymousAuth()
def load_manager_config(self):
self._manager = ManagerConfig(
str(self.manager_url), self.default_auth, remote=True
)
def load_toml_config(self):
if self.default_auth is not None:
toml_user = self.default_auth.get_user()
if not toml_user.is_authenticated:
toml_user = local_user
else: # if no connection to manager
toml_user = local_user
toml_config = self.storage_directory / "platforms.toml"
toml_config.touch()
self._toml = TomlConfig(toml_config, toml_user)
@field_validator("storage_directory")
def expand_user(cls, v):
# translate ~/asdf into /home/user/asdf
return Path.expanduser(v)
def configure_logging(self, config: str):
access_file = self.log_dir / "access.log"
debug_file = self.log_dir / "debug.log"
error_file = self.log_dir / "error.log"
os.environ.setdefault("IXMP4_ACCESS_LOG", str(access_file.absolute()))
os.environ.setdefault("IXMP4_DEBUG_LOG", str(debug_file.absolute()))
os.environ.setdefault("IXMP4_ERROR_LOG", str(error_file.absolute()))
logging_config = here / f"logging/{config}.json"
with open(logging_config, "r") as file:
config_dict = json.load(file)
logging.config.dictConfig(config_dict)
def check_credentials(self):
if self.default_credentials is not None:
username, password = self.default_credentials
ManagerAuth(username, password, str(self.manager_url))