Async/Await examples

.github/workflows/python-app.yml

@@ -36,4 +36,4 @@ jobs:
         flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
     - name: Test with pytest
       run: |
-        pytest
+        pytest -vv

src/asynchronous/runner/concurrent.py

@@ -0,0 +1,15 @@
+"""
+Source: https://www.geeksforgeeks.org/python/await-in-python/
+"""
+import asyncio
+from asynchronous.runner.task.myasynctask import custom_async_task
+from asynchronous.runner.operation.myoperations import first_operation, second_operation
+
+# this coroutine runs both runner concurrently with asyncio.gather()
+async def running_concurrently():
+    await asyncio.gather(custom_async_task(first_operation, 3),
+                         custom_async_task(second_operation, 2))
+
+# starts the event loop, running and waiting for both runner to complete
+def main():
+    asyncio.run(running_concurrently())

src/asynchronous/runner/operation/myoperations.py

@@ -0,0 +1,9 @@
+"""
+Definition of two simple operations.
+"""
+
+def first_operation():
+    print("First operation")
+
+def second_operation():
+    print("Second operation")

src/asynchronous/runner/sequential.py

@@ -0,0 +1,15 @@
+"""
+Source: https://www.geeksforgeeks.org/python/await-in-python/
+"""
+import asyncio
+from asynchronous.runner.task.myasynctask import custom_async_task
+from asynchronous.runner.operation.myoperations import first_operation, second_operation
+
+# runs two runner sequentially, waiting for one to finish before starting the next
+async def running_sequentially():
+    await custom_async_task(first_operation, 3)
+    await custom_async_task(second_operation, 2)
+
+# executes both runner, totaling around 5 seconds of runtime
+def main():
+    asyncio.run(running_sequentially())

src/asynchronous/runner/task/myasynctask.py

@@ -0,0 +1,12 @@
+import asyncio
+
+async def custom_async_task(func, sleep_time):
+    """
+    Custom asynchronous task (coroutine)
+    :param func: operation to execute
+    :param sleep_time: sleep time
+    """
+    print(f"Execution of {func.__name__} started")
+    func()
+    await asyncio.sleep(sleep_time)
+    print(f"Execution of {func.__name__} completed")

src/asynchronous/tests/usage_test.py

@@ -0,0 +1,33 @@
+from unittest.mock import patch, call
+from asynchronous.runner import sequential, concurrent
+
+
+@patch('builtins.print', create=True)
+def test_sequential_run(mock_print):
+    sequential.main()
+
+    order_of_calls = [
+        call("Execution of first_operation started"),
+        call("First operation"),
+        call("Execution of first_operation completed"),
+        call("Execution of second_operation started"),
+        call("Second operation"),
+        call("Execution of second_operation completed")
+    ]
+
+    mock_print.assert_has_calls(order_of_calls, any_order=False)
+
+@patch('builtins.print', create=True)
+def test_concurrent_run(mock_print):
+    concurrent.main()
+
+    order_of_calls = [
+        call("Execution of first_operation started"),
+        call("First operation"),
+        call("Execution of second_operation started"),
+        call("Second operation"),
+        call("Execution of second_operation completed"),
+        call("Execution of first_operation completed")
+    ]
+
+    mock_print.assert_has_calls(order_of_calls, any_order=False)

src/asynchronous/usage1.py

@@ -0,0 +1,3 @@
+from asynchronous.runner import sequential
+
+sequential.main()

src/asynchronous/usage2.py

@@ -0,0 +1,3 @@
+from asynchronous.runner import concurrent
+
+concurrent.main()