Merge pull request #67 from febus982/cleanup-tasks-code

Improve and document tasks application architecture

Author: febus982

architecture.png

@@ -45,7 +45,7 @@ routes --> Book
 servicers --> BookService
 servicers --> Book
 celery_worker ---> tasks
-tasks <-u-> BookService
+tasks <-u- BookService
 tasks -u-> Book
 
 'links internal to books domain

architecture.puml

@@ -35,13 +35,15 @@ class CeleryConfig(BaseModel):
     worker_send_task_events: bool = True
     task_send_sent_event: bool = True
 
-    beat_schedule: dict = {
-        "recurrent_example": {
-            "task": "domains.books.tasks.book_created",
-            "schedule": 5.0,
-            "args": ("a-random-book-id",),
-        },
-    }
+    # Recurring tasks triggered directly by Celery
+    beat_schedule: dict = {}
+    # beat_schedule: dict = {
+    #     "recurrent_example": {
+    #         "task": "domains.books.tasks.book_cpu_intensive_task",
+    #         "schedule": 5.0,
+    #         "args": ("a-random-book-id",),
+    #     },
+    # }
 
 
 class AppConfig(BaseSettings):

common/config.py

@@ -1,6 +1,7 @@
 from collections.abc import Iterable
 
 from anyio import to_thread
+from celery.result import AsyncResult
 from dependency_injector.wiring import Provide, inject
 from structlog import get_logger
 
@@ -9,6 +10,7 @@
 
 from ._data_access_interfaces import BookEventGatewayInterface, BookRepositoryInterface
 from .dto import Book, BookData
+from .tasks import book_cpu_intensive_task
 
 
 class BookService:
@@ -30,13 +32,13 @@ def __init__(
         self.event_gateway = event_gateway
 
     async def create_book(self, book: BookData) -> Book:
-        # 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.model_dump()
-        )
-        book_model = BookModel(**book_data_altered)
+        # Example of CPU intensive task, run in a celery task
+        book_task: AsyncResult = book_cpu_intensive_task.delay(book)
+        # task.get() would block the application, we run it in a thread to remain async
+        # we can also build a wrapper coroutine to do this using `asyncio.sleep`
+        # and poll the AsyncResult class in case we do not want to use threads
+        book_data_altered: BookData = await to_thread.run_sync(book_task.get)
+        book_model = BookModel(**book_data_altered.model_dump())
         book = Book.model_validate(
             await self.book_repository.save(book_model), from_attributes=True
         )
@@ -56,13 +58,6 @@ async def list_books(self) -> Iterable[Book]:
         return [Book.model_validate(x, from_attributes=True) for x in books]
 
     async def book_created_event_handler(self, book_id) -> None:  # pragma: no cover
-        # This is just an example placeholder,
-        # there's nothing to test.
+        # This is just an example placeholder, there's nothing to test.
         logger = get_logger()
         await logger.ainfo(f"Processed book crated event for id `{book_id}`")
-
-
-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

domains/books/service.py

@@ -1,19 +1,27 @@
-from asgiref.sync import async_to_sync
+"""
+Tasks defined in this module are considered intensive operations
+that happen as part of one of the BookService methods,
+therefore we shouldn't invoke again the book service directly
+from here.
+
+Tasks that invoke the BookService could exist (e.g. an event
+worker), there are 2 options to implement them:
+- Create a different module, that would behave similar to HTTP
+routes, and invoke the service from there.
+- Invoke the service using inversion of control.
+
+IMPORTANT: It's dangerous to have nested task when they depend
+on each other's result. If you find yourself in this scenario
+it is probably better to redesign your application. If this is
+not possible, then celery provides task synchronisation primitives.
+
+https://docs.celeryq.dev/en/stable/userguide/tasks.html#avoid-launching-synchronous-subtasks
+"""
 from celery import shared_task
 
-from domains.books.service import BookService
+from domains.books.dto import BookData
 
 
 @shared_task()
-def book_created(book_id):
-    book_service = BookService()
-
-    """
-    This might not work if we call directly hello() from an app with
-    an already running async loop. It would be great having either
-    native async support or a better solution to identify running loops.
-    For now, we'll assume this is always executed in the worker.
-    In order to call this directly we might have to use the opposite
-    `sync_to_async` wrapper from asgiref.sync
-    """
-    return async_to_sync(book_service.book_created_event_handler)(book_id)
+def book_cpu_intensive_task(book: BookData) -> BookData:
+    return book

domains/books/tasks.py

@@ -1,8 +1,9 @@
-from unittest.mock import AsyncMock, MagicMock
+from unittest.mock import AsyncMock, MagicMock, patch
 
 from domains.books import dto, service
 from domains.books._data_access_interfaces import BookEventGatewayInterface
 from domains.books.models import BookModel
+from domains.books.tasks import book_cpu_intensive_task
 
 
 async def test_create_book(book_repository):
@@ -15,7 +16,12 @@ async def test_create_book(book_repository):
         title="test",
         author_name="other",
     )
-    returned_book = await book_service.create_book(book)
+    mocked_task_return = MagicMock
+    mocked_task_return.get = MagicMock(return_value=book_cpu_intensive_task(book))
+    with patch.object(
+        book_cpu_intensive_task, "delay", return_value=mocked_task_return
+    ):
+        returned_book = await book_service.create_book(book)
     assert book.title == returned_book.title
     assert book.author_name == returned_book.author_name
     assert returned_book.book_id is not None

tests/domains/books/test_book_service.py

@@ -1,10 +1,11 @@
 from unittest.mock import MagicMock, patch
 
+from domains.books.dto import BookData
 from domains.books.service import BookService
-from domains.books.tasks import book_created
+from domains.books.tasks import book_cpu_intensive_task
 
 
 @patch.object(BookService, "book_created_event_handler", return_value=None)
 def test_book_created_task(mocked_task_handler: MagicMock):
-    book_created(123)
-    mocked_task_handler.assert_called_once_with(123)
+    b = BookData(title="AA", author_name="BB")
+    assert book_cpu_intensive_task(b) == b