-
Notifications
You must be signed in to change notification settings - Fork 1
Singleton Pattern Implementation
- Introduction
- TVEngineSingleton Implementation
- Thread-Safe Initialization
- Instance Access via get_instance()
- Singleton Reset for Testing
- Application to Core Components
- Usage Examples
- Common Pitfalls
The singleton pattern is implemented throughout the PyTradingView application to ensure global consistency of core components. This document details the implementation of the TVEngineSingleton mixin class and its application across key system components such as EventBus and TVBridge. The pattern ensures that only one instance of these critical components exists throughout the application lifecycle, providing centralized control and state management.
The TVEngineSingleton mixin class provides a thread-safe implementation of the singleton pattern for indicator engine components. It serves as a base class for various engine components that require global instance consistency.
classDiagram
class TVEngineSingleton {
-_instance : Optional[TVEngineSingleton]
-_lock : Optional[threading.Lock]
+__new__(cls, config : Optional[TVWidgetConfig]) TVEngineSingleton
+get_instance(cls, config : Optional[TVWidgetConfig]) TVEngineSingleton
+reset(cls) None
+setup(indicators_dir : Optional[str], auto_activate : bool, config : Optional[Dict[str, Any]]) TVEngineSingleton
+run(widget_config : Optional[Dict[str, Any]], indicators_dir : Optional[str], on_port : int) None
}
The TVEngineSingleton class implements thread-safe singleton creation through the use of threading.Lock. The __new__ method contains the core initialization guard mechanism that prevents multiple instance creation in concurrent environments.
The implementation follows a double-checked locking pattern:
- A class-level lock is created on first access
- The lock is acquired before checking instance existence
- Instance creation occurs only if no instance exists
- The same instance is returned for all subsequent requests
This approach ensures that even under high concurrency, only one instance of the singleton will be created, maintaining global consistency across threads.
The get_instance() class method serves as the standard way to access singleton instances throughout the application. This method provides a clean, consistent interface for obtaining the singleton instance without requiring direct calls to __new__.
When get_instance() is called:
- It first checks if an instance already exists
- If no instance exists, it creates one by calling the class constructor
- If an instance already exists, it returns the existing instance
- The method accepts an optional configuration parameter that is only used during the first instantiation
This pattern allows components to safely obtain the singleton instance at any point in the application lifecycle.
sequenceDiagram
participant Client
participant TVEngineSingleton
participant Logger
Client->>TVEngineSingleton : get_instance(config)
TVEngineSingleton->>TVEngineSingleton : Check _instance is None?
alt Instance does not exist
TVEngineSingleton->>TVEngineSingleton : Create new instance
TVEngineSingleton->>TVEngineSingleton : Set _initialized = False
TVEngineSingleton-->>Client : Return new instance
else Instance exists
TVEngineSingleton-->>Client : Return existing instance
end
The reset() class method provides a mechanism to reset the singleton instance, primarily intended for testing purposes. This method allows test suites to start with a clean state by destroying the existing singleton instance.
When reset() is called:
- It checks if an instance currently exists
- If the instance has a
deactivate_allmethod, it calls this method to properly clean up active indicators - It sets the
_instanceclass variable toNone - It logs a warning message about the reset operation
The method includes a clear warning that it should not be used in production environments, as resetting a singleton during normal application operation could lead to unpredictable behavior and state inconsistencies.
The singleton pattern is applied to several core components beyond the indicator engine, ensuring global consistency across the application architecture.
The EventBus class implements the singleton pattern to provide a centralized event distribution system. It maintains a registry of event subscribers and ensures that all components communicate through a single event bus instance.
classDiagram
class EventBus {
-_instance : Optional[EventBus]
-_subscribers : Dict[EventType, List[Callable]]
-_event_queue : asyncio.Queue
-_running : bool
+get_instance() EventBus
+subscribe(event_type : EventType, callback : Callable) None
+unsubscribe(event_type : EventType, callback : Callable) None
+publish(event_type : EventType, data : Optional[Dict], source : Optional[str]) None
+publish_sync(event_type : EventType, data : Optional[Dict], source : Optional[str]) None
}
The TVBridge class uses the singleton pattern to manage the connection between Python and Node.js components. It ensures that only one bridge instance handles communication, preventing port conflicts and connection race conditions.
classDiagram
class TVBridge {
-_instance : Optional[TVBridge]
-node_server_port : int
-is_connected_to_node : bool
-bridge_port : int
-bridge_http_app : FastAPI
-start_event : Event
-_config_provider : Optional[TVWidgetConfig]
-_chart_ready_callback : Optional[Callable]
+get_instance() TVBridge
+register_config_provider(config : TVWidgetConfig) None
+register_chart_ready_callback(callback : Callable) None
+start_http_server(on_port : int) None
+connect_to_node_server(max_retries : int, base_delay : float) bool
+call_node_server(params : TVMethodCall) TVMethodResponse
+run(on_port : int) None
}
Additional components that implement the singleton pattern include:
-
IndicatorRegistry: Manages registration and discovery of trading indicators -
TVObjectPool: Provides object pooling services with singleton semantics -
TVSubscribeManager: Manages subscription state across the application
These implementations follow similar patterns to TVEngineSingleton, using class-level instance tracking and get_instance() methods for access.
Proper singleton usage follows a consistent pattern across the application:
# Correct usage pattern
from pytradingview.indicators.engine.singleton_mixin import TVEngineSingleton
# Obtain the singleton instance
engine = TVEngineSingleton.get_instance(config)
# Use the instance for operations
engine.setup(indicators_dir="./indicators")
engine.run(on_port=8080)
# Access the same instance elsewhere in the application
same_engine = TVEngineSingleton.get_instance()
assert engine is same_engine # This will be TrueFor components like EventBus, the pattern is identical:
from pytradingview.core.TVEventBus import EventBus
# Get the singleton event bus
event_bus = EventBus.get_instance()
# Subscribe to events
async def handle_chart_ready(event):
print("Chart is ready!")
event_bus.subscribe(EventType.CHART_READY, handle_chart_ready)Several common pitfalls should be avoided when working with singletons in this application:
A common issue occurs when singleton initialization creates circular dependencies. For example, if TVBridge requires EventBus during initialization, and EventBus requires TVBridge, a circular dependency is created.
The application avoids this through:
- Deferred initialization using callback registration
- Lazy instantiation of dependent components
- Careful ordering of component startup
Using the reset() method in production can lead to:
- Loss of application state
- Disconnected components
- Memory leaks from orphaned references
The reset method should only be used in test teardown methods to ensure test isolation.
While the core singleton implementation is thread-safe, subclasses must ensure that their initialization code is also thread-safe. This includes:
- Proper synchronization of instance variable initialization
- Avoiding race conditions in setup methods
- Ensuring atomic operations on shared state