Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 5 additions & 2 deletions .github/workflows/python-package.yml
Original file line number Diff line number Diff line change
Expand Up @@ -5,9 +5,12 @@ name: Python package

on:
push:
branches: [ "develop" ]
branches:
- develop
- feature/*
pull_request:
branches: [ "master" ]
branches:
- master

jobs:
build:
Expand Down
Empty file added src/topmodels/utils/__init__.py
Empty file.
32 changes: 32 additions & 0 deletions src/topmodels/utils/meta.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,32 @@
"""
meta.py

This module provides a thread-safe Singleton metaclass implementation.
"""

from threading import Lock
from typing import Any, Dict, TypeVar

T = TypeVar("T", bound="SingletonMeta")

class SingletonMeta(type):
"""
Thread-safe Singleton metaclass.

Ensures that only one instance of a class using this metaclass exists.
"""

_instances: Dict[Any, Any] = {}
_lock: Lock = Lock()

def __call__(cls, *args: Any, **kwargs: Any) -> T:
"""
Returns the singleton instance of the class.

If the instance does not exist, it is created in a thread-safe manner.
"""
if cls not in cls._instances:
with cls._lock:
if cls not in cls._instances:
cls._instances[cls] = super().__call__(*args, **kwargs)
return cls._instances[cls]
82 changes: 82 additions & 0 deletions tests/unit_tests/utils/test_singleton_meta.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,82 @@
"""
Unit tests for the SingletonMeta metaclass.

This module tests the thread-safety and uniqueness of the SingletonMeta implementation.
"""

import threading
from threading import Thread

import pytest

from topmodels.utils.meta import SingletonMeta


@pytest.fixture(autouse=True)
def reset_singleton_instances() -> None:
"""
Pytest fixture that automatically resets the SingletonMeta instance cache before each test.

This fixture clears the `_instances` dictionary of the `SingletonMeta` metaclass,
ensuring that singleton instances do not persist between tests and each test runs
with a fresh singleton state.
"""
SingletonMeta._instances.clear()

class MySingleton(metaclass=SingletonMeta):
"""
Example singleton class using SingletonMeta.

Args:
value (int, optional): Value to store in the singleton instance. Defaults to 0.
"""
def __init__(self, value=0):
"""
Initialize the singleton instance.

Args:
value (int, optional): Value to store. Defaults to 0.
"""
self.value = value

def test__singleton_instance_uniqueness():
"""
Test that only one instance of the singleton is created,
regardless of how many times the class is instantiated.
"""
a = MySingleton(1)
b = MySingleton(2)
assert a is b
assert a.value == b.value

def test__singleton_value_persistence():
"""
Test that the singleton instance retains its value across instantiations.
"""
instance = MySingleton(42)
assert instance.value == 42
instance.value = 100
new_instance = MySingleton()
assert new_instance.value == 100

def test__singleton_thread_safety():
"""
Test that the singleton instance is unique and consistent across multiple threads.
"""
results = []

def create_instance(val):
instance = MySingleton(val)
results.append(instance)

threads: list[Thread] = [
threading.Thread(target=create_instance, args=(i,))
for i in range(10)
]
for t in threads:
t.start()
for t in threads:
t.join()

# All threads should have received the same instance
assert all(inst is results[0] for inst in results)