ets-labs/python-dependency-injector
1
1
Fork 0
mirror of https://github.com/ets-labs/python-dependency-injector.git synced 2025-07-18 12:02:23 +03:00
Code Issues Packages Projects Releases Wiki Activity

Update singleton_thread_locals.py

Browse Source
This commit is contained in:
RobinsonMa 2020-06-17 09:47:11 +08:00 committed by GitHub
parent 4e6fc276a9
commit fda3083280
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 6 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
examples/providers/singleton_thread_locals.py
View File

@ -6,21 +6,21 @@ 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.ThreadSafeSingleton(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