Skip to content
Snippets Groups Projects
Commit 300f4fed authored by Krzysztof Mochocki's avatar Krzysztof Mochocki
Browse files

Refactor settings

Move defaults to seperate class
Move environment names to seperate class
Move default_factory function to Defaults class
parent 3c63d3c6
No related branches found
No related tags found
1 merge request!73Clive integration related fixes
from __future__ import annotations from __future__ import annotations
from datetime import timedelta from datetime import timedelta
from distutils.util import strtobool
from pathlib import Path from pathlib import Path
from typing import ClassVar
from pydantic import Field
from helpy import HttpUrl from helpy import HttpUrl
from helpy import Settings as HandleSettings from helpy import Settings as HandleSettings
...@@ -12,7 +12,20 @@ from helpy import Settings as HandleSettings ...@@ -12,7 +12,20 @@ from helpy import Settings as HandleSettings
class Settings(HandleSettings): class Settings(HandleSettings):
"""Defines parameters for beekeeper how to start and behave.""" """Defines parameters for beekeeper how to start and behave."""
working_directory: Path = Field(default_factory=lambda: Path.cwd()) class EnvironNames(HandleSettings.EnvironNames):
WORKING_DIRECTORY: ClassVar[str] = "BEEKEEPY_WORKING_DIRECTORY"
PROPAGATE_SIGINT: ClassVar[str] = "BEEKEEPY_PROPAGATE_SIGINT"
CLOSE_TIMEOUT: ClassVar[str] = "BEEKEEPY_CLOSE_TIMEOUT"
class Defaults(HandleSettings.Defaults):
WORKING_DIRECTORY: ClassVar[Path] = Path.cwd()
PROPAGATE_SIGINT: ClassVar[bool] = True
CLOSE_TIMEOUT: ClassVar[timedelta] = timedelta(seconds=10.0)
working_directory: Path = Defaults.default_factory(
EnvironNames.WORKING_DIRECTORY,
lambda x: (Settings.Defaults.WORKING_DIRECTORY if x is None else Path(x)),
)
"""Path, where beekeeper binary will store all it's data and logs.""" """Path, where beekeeper binary will store all it's data and logs."""
http_endpoint: HttpUrl | None = None # type: ignore[assignment] http_endpoint: HttpUrl | None = None # type: ignore[assignment]
...@@ -24,11 +37,17 @@ class Settings(HandleSettings): ...@@ -24,11 +37,17 @@ class Settings(HandleSettings):
binary_path: Path | None = None binary_path: Path | None = None
"""Alternative path to beekeeper binary.""" """Alternative path to beekeeper binary."""
propagate_sigint: bool = True propagate_sigint: bool = Defaults.default_factory(
EnvironNames.PROPAGATE_SIGINT,
lambda x: (Settings.Defaults.PROPAGATE_SIGINT if x is None else strtobool(x)),
)
"""If set to True (default), sigint will be sent to beekeeper without control of this library.""" """If set to True (default), sigint will be sent to beekeeper without control of this library."""
use_existing_session: str | None = None use_existing_session: str | None = None
"""If set, beekeeper will use given session while connecting to beeekeeper.""" """If set, beekeeper will use given session while connecting to beeekeeper."""
close_timeout: timedelta = Field(default_factory=lambda: timedelta(seconds=10.0)) close_timeout: timedelta = Defaults.default_factory(
"""This timeout varriable affects time handle waits before beekeepy closes.""" EnvironNames.CLOSE_TIMEOUT,
lambda x: (Settings.Defaults.CLOSE_TIMEOUT if x is None else timedelta(seconds=int(x))),
)
"""Affects time handle waits before beekeepy closes."""
...@@ -14,25 +14,37 @@ if TYPE_CHECKING: ...@@ -14,25 +14,37 @@ if TYPE_CHECKING:
from typing_extensions import Self from typing_extensions import Self
def _default_factory(env_name: str, default_factory: Callable[[str | None], Any]) -> Any:
env_value = environ.get(env_name)
return Field(default_factory=lambda: default_factory(env_value))
class CommunicationSettings(BaseModel): class CommunicationSettings(BaseModel):
DEFAULT_TIMEOUT: ClassVar[timedelta] = timedelta(seconds=5) class EnvironNames:
TIMEOUT: ClassVar[str] = "HELPY_COMMUNICATION_MAX_RETRIES"
max_retries: int = _default_factory("HELPY_COMMUNICATION_MAX_RETRIES", lambda x: int(x or 5)) PERIOD_BETWEEN_RETRIES: ClassVar[str] = "HELPY_COMMUNICATION_TIMEOUT_SECS"
RETRIES: ClassVar[str] = "HELPY_COMMUNICATION_PERIOD_BETWEEN_RETRIES_SECS"
class Defaults:
TIMEOUT: ClassVar[timedelta] = timedelta(seconds=5)
PERIOD_BETWEEN_RETRIES: ClassVar[timedelta] = timedelta(seconds=0.2)
RETRIES: ClassVar[int] = 5
@staticmethod
def default_factory(env_name: str, default_factory: Callable[[str | None], Any]) -> Any:
env_value = environ.get(env_name)
return Field(default_factory=lambda: default_factory(env_value))
max_retries: int = Defaults.default_factory(
EnvironNames.RETRIES,
lambda x: (CommunicationSettings.Defaults.RETRIES if x is None else timedelta(seconds=int(x))),
)
"""Amount of retries when sending request to service.""" """Amount of retries when sending request to service."""
timeout: timedelta = _default_factory( timeout: timedelta = Defaults.default_factory(
"HELPY_COMMUNICATION_TIMEOUT_SECS", EnvironNames.TIMEOUT,
lambda x: (CommunicationSettings.DEFAULT_TIMEOUT if x is None else timedelta(seconds=int(x))), lambda x: (CommunicationSettings.Defaults.TIMEOUT if x is None else timedelta(seconds=int(x))),
) )
"""Maximum time for request to finish.""" """Maximum time for single request to finish."""
period_between_retries: timedelta = _default_factory( period_between_retries: timedelta = Defaults.default_factory(
"HELPY_COMMUNICATION_PERIOD_BETWEEN_RETRIES_SECS", lambda x: timedelta(seconds=int(x or 1)) EnvironNames.PERIOD_BETWEEN_RETRIES,
lambda x: (CommunicationSettings.Defaults.PERIOD_BETWEEN_RETRIES if x is None else timedelta(seconds=int(x))),
) )
"""Period between failed request and next retry.""" """Period between failed request and next retry."""
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment