Merge pull request #53 from febus982/example_blocking_sync_routine

Add code example for thread worker routine

Author: febus982

domains/books/_service.py

@@ -1,5 +1,6 @@
 from collections.abc import Iterable
 
+from anyio import to_thread
 from dependency_injector.wiring import Provide, inject
 
 from ._data_access_interfaces import BookRepositoryInterface
@@ -21,8 +22,21 @@ def __init__(
         self.book_repository = book_repository
 
     async def create_book(self, book: BookData) -> Book:
-        return Book.from_orm(await self.book_repository.save(BookModel(**book.dict())))
+        # Example of CPU intensive task, run in a different thread
+        # Using processes could be better, but it would bring technical complexity
+        # https://anyio.readthedocs.io/en/3.x/subprocesses.html#running-functions-in-worker-processes
+        book_data_altered = await to_thread.run_sync(
+            some_cpu_intensive_blocking_task, book.dict()
+        )
+        book_model = BookModel(**book_data_altered)
+        return Book.from_orm(await self.book_repository.save(book_model))
 
     async def list_books(self) -> Iterable[Book]:
         books = await self.book_repository.find()
         return [Book.from_orm(x) for x in books]
+
+
+def some_cpu_intensive_blocking_task(book: dict) -> dict:
+    # This is just an example placeholder,
+    # there's nothing to test.
+    return book  # pragma: no cover

grpc_app/__init__.py

@@ -23,7 +23,7 @@ async def create_server(test_config: Optional[AppConfig] = None):
     add_BooksServicer_to_server(BooksServicer(), s)
     address = "0.0.0.0:9999"
     logger = get_logger()
-    logger.ainfo(f"[{os.getpid()}] Listening on {address}")
+    await logger.ainfo(f"[{os.getpid()}] Listening on {address}")
     s.add_insecure_port(address)
     return s