Update singleton_thread_locals.py (#252)

* Update singleton_thread_locals.py

fix

* Update singleton_thread_locals.py

* Update singleton_thread_locals.py
This commit is contained in:
RobinsonMa 2020-06-17 09:53:00 +08:00 committed by GitHub
parent a0d4af15e9
commit c136663071
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23

View File

@ -1,26 +1,26 @@
"""`ThreadLocalSingleton` providers example.""" """`ThreadLocalSingleton` providers example."""
import threading import threading
import Queue import queue
import dependency_injector.providers as providers import dependency_injector.providers as providers
def example(example_object, queue): def example(example_object, queue_object):
"""Put provided object in the provided queue.""" """Put provided object in the provided queue."""
queue.put(example_object) queue_object.put(example_object)
# Create thread-local singleton provider for some object (main thread): # Create thread-local singleton provider for some object (main thread):
thread_local_object = providers.ThreadLocalSingleton(object) thread_local_object = providers.ThreadLocalSingleton(object)
# Create singleton provider for thread-safe queue: # Create singleton provider for thread-safe queue:
queue = providers.Singleton(Queue.Queue) queue_factory = providers.ThreadSafeSingleton(queue.Queue)
# Create callable provider for example(), inject dependencies: # Create callable provider for example(), inject dependencies:
example = providers.DelegatedCallable(example, example = providers.DelegatedCallable(example,
example_object=thread_local_object, example_object=thread_local_object,
queue=queue) queue_object=queue_factory)
# Create factory provider for threads that are targeted to execute example(): # Create factory provider for threads that are targeted to execute example():
thread_factory = providers.Factory(threading.Thread, thread_factory = providers.Factory(threading.Thread,
@ -43,8 +43,8 @@ if __name__ == '__main__':
# Making some asserts (main thread): # Making some asserts (main thread):
all_objects = set() all_objects = set()
while not queue().empty(): while not queue_factory().empty():
all_objects.add(queue().get()) all_objects.add(queue_factory().get())
assert len(all_objects) == len(threads) assert len(all_objects) == len(threads)
# Queue contains same number of objects as number of threads where # Queue contains same number of objects as number of threads where