Last active
June 23, 2023 15:43
-
-
Save VietThan/321fea4fe2da244b27353d7a0edd911d to your computer and use it in GitHub Desktop.
Thread-safe object to store your configs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from threading import Lock, Thread | |
from pathlib import Path | |
from typing import Optional, Union | |
import os | |
import logging | |
class SingletonMeta(type): | |
""" | |
This is a thread-safe implementation of Singleton. | |
Straight from https://refactoring.guru/design-patterns/singleton/python/example#example-1 | |
""" | |
_instances = {} | |
_lock: Lock = Lock() | |
""" | |
We now have a lock object that will be used to synchronize threads during | |
first access to the Singleton. | |
""" | |
def __call__(cls, *args, **kwargs): | |
""" | |
Possible changes to the value of the `__init__` argument do not affect | |
the returned instance. | |
""" | |
# Now, imagine that the program has just been launched. Since there's no | |
# Singleton instance yet, multiple threads can simultaneously pass the | |
# previous conditional and reach this point almost at the same time. The | |
# first of them will acquire lock and will proceed further, while the | |
# rest will wait here. | |
with cls._lock: | |
# The first thread to acquire the lock, reaches this conditional, | |
# goes inside and creates the Singleton instance. Once it leaves the | |
# lock block, a thread that might have been waiting for the lock | |
# release may then enter this section. But since the Singleton field | |
# is already initialized, the thread won't create a new object. | |
if cls not in cls._instances: | |
instance = super().__call__(*args, **kwargs) | |
cls._instances[cls] = instance | |
return cls._instances[cls] | |
class Configs(metaclass=SingletonMeta): | |
_dict = {} | |
""" | |
We'll use this property to prove that our Singleton really works. | |
""" | |
def __init__(self, arg_folder_path: Optional[Union[str, Path]] = None) -> None: | |
self._dict.update(os.environ) | |
logging.info(f"loading *.yaml in configs folder") | |
folder_path: Union[str, Path, None] = self._dict.get("AUTOMATION_CONFIG_FOLDER_PATH") | |
if not folder_path: | |
logging.info("Did not find config folder path with env variable AUTOMATION_CONFIG_FOLDER_PATH") | |
folder_path = arg_folder_path | |
if not folder_path: | |
logging.info("Did not find config folder path passed in as argument") | |
logging.info("Defaulting to configs/ folder relative to configs.py") | |
folder_path = Path(__file__).parent / "configs" | |
if folder_path: | |
import yaml | |
for yaml_file in Path(folder_path).glob("*.yaml"): | |
with open(yaml_file, "r") as f: | |
self._dict.update(yaml.safe_load(f.read())) | |
def __getitem__(self, key: str) -> str: | |
"""Allow usage of Configs object with square-bracket notation. | |
Example: | |
```python | |
configs = Configs() | |
redis_host = configs["redis_host"] | |
``` | |
Parameters | |
---------- | |
key : str | |
the config key name | |
Returns | |
------- | |
str | |
the config value | |
""" | |
return self._dict[key] | |
def __contains__(self, key: str) -> bool: | |
"""All | |
Parameters | |
---------- | |
key : str | |
the config key name | |
Returns | |
------- | |
bool | |
whether config exists in settings | |
""" | |
return key in self._dict | |
def __str__(self) -> str: | |
return str(self._dict) | |
def get(self, key: str, default: str = None) -> str: | |
return self._dict.get(key, default) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment