Skip to content

FastAPIProcessingService class

Bases: ProcessingService

A processing service that utilizes the FastAPI's BackgroundTasks to handle the execution of calls.

Notes:

  • This service is specifically designed for FastAPI applications and leverages FastAPI's BackgroundTasks to handle the callbacks' execution. This is useful for operations that need to happen after a request, but that the client doesn't really have to be waiting for the operation to complete before receiving the response.
Source code in pyventus/core/processing/fastapi/fastapi_processing_service.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
class FastAPIProcessingService(ProcessingService):
    """
    A processing service that utilizes the FastAPI's `BackgroundTasks` to handle the execution of calls.

    **Notes:**

    -   This service is specifically designed for FastAPI applications and leverages FastAPI's `BackgroundTasks`
        to handle the callbacks' execution. This is useful for operations that need to happen after a request, but
        that the client doesn't really have to be waiting for the operation to complete before receiving the response.
    """

    # Attributes for the FastAPIProcessingService
    __slots__ = ("__background_tasks",)

    def __init__(self, background_tasks: BackgroundTasks) -> None:
        """
        Initialize an instance of `FastAPIProcessingService`.

        :param background_tasks: The FastAPI `BackgroundTasks` object used to handle callbacks' execution.
        :return: None.
        """
        # Validate the background_tasks instance.
        if background_tasks is None or not isinstance(background_tasks, BackgroundTasks):
            raise PyventusException("The 'background_tasks' argument must be an instance of the BackgroundTasks.")

        # Store the BackgroundTasks instance.
        self.__background_tasks: BackgroundTasks = background_tasks

    def __repr__(self) -> str:
        """
        Retrieve a string representation of the instance.

        :return: A string representation of the instance.
        """
        return formatted_repr(
            instance=self,
            info=attributes_repr(
                background_tasks=self.__background_tasks,
            ),
        )

    @override
    def submit(self, callback: ProcessingServiceCallbackType, *args: Any, **kwargs: Any) -> None:
        # Add the callback to the background_tasks instance as a new task to be executed.
        self.__background_tasks.add_task(callback, *args, **kwargs)

Functions

__init__ 

__init__(background_tasks: BackgroundTasks) -> None

Initialize an instance of FastAPIProcessingService.

PARAMETER DESCRIPTION
background_tasks

The FastAPI BackgroundTasks object used to handle callbacks' execution.

TYPE: BackgroundTasks

RETURNS DESCRIPTION
None

None.
Source code in pyventus/core/processing/fastapi/fastapi_processing_service.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
def __init__(self, background_tasks: BackgroundTasks) -> None:
    """
    Initialize an instance of `FastAPIProcessingService`.

    :param background_tasks: The FastAPI `BackgroundTasks` object used to handle callbacks' execution.
    :return: None.
    """
    # Validate the background_tasks instance.
    if background_tasks is None or not isinstance(background_tasks, BackgroundTasks):
        raise PyventusException("The 'background_tasks' argument must be an instance of the BackgroundTasks.")

    # Store the BackgroundTasks instance.
    self.__background_tasks: BackgroundTasks = background_tasks

submit 

submit(callback: ProcessingServiceCallbackType, *args: Any, **kwargs: Any) -> None
Source code in pyventus/core/processing/fastapi/fastapi_processing_service.py 
56
57
58
59
@override
def submit(self, callback: ProcessingServiceCallbackType, *args: Any, **kwargs: Any) -> None:
    # Add the callback to the background_tasks instance as a new task to be executed.
    self.__background_tasks.add_task(callback, *args, **kwargs)