Skip to content

Commit

Permalink
Make settings a singleton
Browse files Browse the repository at this point in the history
  • Loading branch information
hugobessa committed Dec 10, 2024
1 parent 1fb6c5d commit 731d0c4
Show file tree
Hide file tree
Showing 6 changed files with 52 additions and 15 deletions.
20 changes: 19 additions & 1 deletion poetry.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

1 change: 1 addition & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ readme = "README.md"
[tool.poetry.dependencies]
python = "<3.13,>=3.9"
typing-extensions = "^4.12.2"
pytest-asyncio = "^0.24.0"


[tool.poetry.group.dev.dependencies]
Expand Down
4 changes: 3 additions & 1 deletion vintasend/app_settings.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,8 @@
import os
from typing import Any, TypedDict, cast

from vintasend.utils.singleton_utils import SingletonMeta


class NotificationSettingsDict(TypedDict):
NOTIFICATION_ADAPTERS: list[tuple[str, str]]
Expand Down Expand Up @@ -132,7 +134,7 @@ def get_config(setting_name: str, config: Any = None):
return {}


class NotificationSettings:
class NotificationSettings(metaclass=SingletonMeta):
NOTIFICATION_ADAPTERS: list[tuple[str, str]]
NOTIFICATION_BACKEND: str
NOTIFICATION_MODEL: str | None
Expand Down
16 changes: 10 additions & 6 deletions vintasend/services/helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -57,10 +57,11 @@ def get_notification_adapters(
config: Any = None,
) -> list[BaseNotificationAdapter]:
default_adapters = []
app_settings = NotificationSettings(config)
adapters_imports_strs_with_default = (
adapters_imports_strs
if adapters_imports_strs is not None
else NotificationSettings(config).NOTIFICATION_ADAPTERS
else app_settings.NOTIFICATION_ADAPTERS
)
for adapter_import_str, template_renderer_import_str in adapters_imports_strs_with_default:
adapter_kwargs: dict = {}
Expand All @@ -77,7 +78,7 @@ def get_notification_adapters(
try:
adapter = adapter_cls(
template_renderer_import_str,
backend if backend else NotificationSettings().NOTIFICATION_BACKEND,
backend if backend else app_settings.NOTIFICATION_BACKEND,
backend_kwargs,
config,
**adapter_kwargs,
Expand Down Expand Up @@ -106,10 +107,11 @@ def get_asyncio_notification_adapters(
config: Any = None,
) -> list[AsyncIOBaseNotificationAdapter]:
default_adapters = []
app_settings = NotificationSettings(config)
adapters_imports_strs_with_default = (
adapters_imports_strs
if adapters_imports_strs is not None
else NotificationSettings(config).NOTIFICATION_ADAPTERS
else app_settings.NOTIFICATION_ADAPTERS
)
for adapter_import_str, template_renderer_import_str in adapters_imports_strs_with_default:
adapter_kwargs: dict = {}
Expand All @@ -125,7 +127,7 @@ def get_asyncio_notification_adapters(
try:
adapter = adapter_cls(
template_renderer_import_str,
backend if backend else NotificationSettings().NOTIFICATION_BACKEND,
backend if backend else app_settings.NOTIFICATION_BACKEND,
backend_kwargs,
config,
**adapter_kwargs,
Expand All @@ -147,10 +149,11 @@ def get_asyncio_notification_adapters(
def get_notification_backend(
backend_import_str: str | None, backend_kwargs: dict | None = None, config: Any = None
) -> BaseNotificationBackend:
app_settings = NotificationSettings(config)
backend_import_str_with_fallback = (
backend_import_str
if backend_import_str is not None
else NotificationSettings(config).NOTIFICATION_BACKEND
else app_settings.NOTIFICATION_BACKEND
)

try:
Expand All @@ -177,10 +180,11 @@ def get_notification_backend(
def get_asyncio_notification_backend(
backend_import_str: str | None, backend_kwargs: dict | None = None, config: Any = None
) -> AsyncIOBaseNotificationBackend:
app_settings = NotificationSettings(config)
backend_import_str_with_fallback = (
backend_import_str
if backend_import_str is not None
else NotificationSettings(config).NOTIFICATION_BACKEND
else app_settings.NOTIFICATION_BACKEND
)

try:
Expand Down
5 changes: 5 additions & 0 deletions vintasend/services/notification_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
from collections.abc import Callable, Iterable
from typing import Any, ClassVar, Generic, TypeGuard, TypeVar, cast

from vintasend.app_settings import NotificationSettings
from vintasend.services.notification_backends.asyncio_base import AsyncIOBaseNotificationBackend


Expand Down Expand Up @@ -81,6 +82,10 @@ def __init__(
notification_backend_kwargs: dict | None = None,
config: Any = None,
):
# initialize the notification settings singleton for the first time
# to ensure all components have access to the same settings
NotificationSettings(config)

if isinstance(notification_backend, BaseNotificationBackend):
self.notification_backend = cast(B, notification_backend)
else:
Expand Down
21 changes: 14 additions & 7 deletions vintasend/utils/singleton_utils.py
Original file line number Diff line number Diff line change
@@ -1,13 +1,20 @@
from types import MappingProxyType
from typing import Any
from typing import Any, Generic, TypeVar, cast


class SingletonMeta(type):
_instances: MappingProxyType[str, Any] = MappingProxyType({})
class BaseSingletonMeta(type):
_instances: MappingProxyType[Any, Any]

def __call__(cls, *args, **kwargs):

T = TypeVar('T', bound=BaseSingletonMeta)


class SingletonMeta(Generic[T], BaseSingletonMeta):
_instances: MappingProxyType["SingletonMeta[T]", T] = MappingProxyType({})

def __call__(cls, *args: Any, **kwargs: Any) -> T:
if cls not in cls._instances:
cls._instances = dict(cls._instances)
cls._instances[cls] = super().__call__(*args, **kwargs)
cls._instances = MappingProxyType(cls._instances)
_instances: dict["SingletonMeta[T]", T] = dict(cls._instances)
_instances[cls] = cast(T, super().__call__(*args, **kwargs))
cls._instances = MappingProxyType(_instances)
return cls._instances[cls]

0 comments on commit 731d0c4

Please sign in to comment.