Services

calculator

This is an example app in order to give some examples on how to build a python app repo using good practices in terms of linting, testing and documentation automation

Calculator

Calculator App class

Source code in src/services/calculator.py
 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
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
class Calculator:
    """
    Calculator App class
    """

    def __init__(self, *args: Any, **kwds: Any) -> None:
        pass

    def __str__(self, *args: Any, **kwds: Any) -> Any:
        pass

    def __del__(self, *args: Any, **kwds: Any) -> Any:
        pass

    def __call__(self, *args: Any, **kwds: Any) -> Any:
        pass

    @staticmethod
    def suma(a: float = 1, b: float = 1) -> float:
        """Function that returns addition of two integer numbers

        Args:
            a (float, optional): first number to add. Defaults to 1.
            b (float, optional): second number to add. Defaults to 1.

        Returns:
            float: addition of a and b
        """
        if isinstance(a, str) or isinstance(b, str):
            raise AdditionError("Invalid input. Only numbers are allowed")

        return a + b

    @staticmethod
    def resta(a: float = 1, b: float = 1) -> float:
        """Function that returns difference of two integer numbers

        Args:
            a (float, optional): _description_. Defaults to 1.
            b (float, optional): _description_. Defaults to 1.

        Returns:
            float: difference of a and b
        """
        if isinstance(a, str) or isinstance(b, str):
            raise SubstractionError("Invalid input. Only numbers are allowed")

        return a - b

    @staticmethod
    def multiplicacion(a: float = 1, b: float = 1) -> float:
        """Function that multiplies two float values

        Args:
            a (float, optional): First parameter. Defaults to 1.
            b (float, optional): Second parameter. Defaults to 1.

        Returns:
            float: multiplication of a and b params
        """
        if isinstance(a, str) or isinstance(b, str):
            error_msg = "Invalid input. Only numbers are allowed"
            raise MultiplicationError(error_msg)

        return a * b

    @staticmethod
    def division(a: float = 1, b: float = 1) -> float | ZeroDivisionError:
        """Function that returns division of two values where
           b is non zero

        Args:
            a (float, optional): First param. Defaults to 1.
            b (float, optional): Second param (non-zero). Defaults to 1.

        Returns:
            float: division result of a / b where b is non-zero
        """
        if isinstance(a, str) or isinstance(b, str):
            raise DivisionError("Invalid input. Only numbers are allowed")

        try:
            return a / b
        except ZeroDivisionError as error:
            msg = "Division by 0 not allowed!"
            raise ZeroDivisionError(msg) from error

division(a=1, b=1) staticmethod

Function that returns division of two values where b is non zero

Parameters:
  • a (float, default: 1 ) –

    First param. Defaults to 1.

  • b (float, default: 1 ) –

    Second param (non-zero). Defaults to 1.

Returns:
  • float( float | ZeroDivisionError ) –

    division result of a / b where b is non-zero

Source code in src/services/calculator.py
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
@staticmethod
def division(a: float = 1, b: float = 1) -> float | ZeroDivisionError:
    """Function that returns division of two values where
       b is non zero

    Args:
        a (float, optional): First param. Defaults to 1.
        b (float, optional): Second param (non-zero). Defaults to 1.

    Returns:
        float: division result of a / b where b is non-zero
    """
    if isinstance(a, str) or isinstance(b, str):
        raise DivisionError("Invalid input. Only numbers are allowed")

    try:
        return a / b
    except ZeroDivisionError as error:
        msg = "Division by 0 not allowed!"
        raise ZeroDivisionError(msg) from error

multiplicacion(a=1, b=1) staticmethod

Function that multiplies two float values

Parameters:
  • a (float, default: 1 ) –

    First parameter. Defaults to 1.

  • b (float, default: 1 ) –

    Second parameter. Defaults to 1.

Returns:
  • float( float ) –

    multiplication of a and b params

Source code in src/services/calculator.py
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
@staticmethod
def multiplicacion(a: float = 1, b: float = 1) -> float:
    """Function that multiplies two float values

    Args:
        a (float, optional): First parameter. Defaults to 1.
        b (float, optional): Second parameter. Defaults to 1.

    Returns:
        float: multiplication of a and b params
    """
    if isinstance(a, str) or isinstance(b, str):
        error_msg = "Invalid input. Only numbers are allowed"
        raise MultiplicationError(error_msg)

    return a * b

resta(a=1, b=1) staticmethod

Function that returns difference of two integer numbers

Parameters:
  • a (float, default: 1 ) –

    description. Defaults to 1.

  • b (float, default: 1 ) –

    description. Defaults to 1.

Returns:
  • float( float ) –

    difference of a and b

Source code in src/services/calculator.py
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
@staticmethod
def resta(a: float = 1, b: float = 1) -> float:
    """Function that returns difference of two integer numbers

    Args:
        a (float, optional): _description_. Defaults to 1.
        b (float, optional): _description_. Defaults to 1.

    Returns:
        float: difference of a and b
    """
    if isinstance(a, str) or isinstance(b, str):
        raise SubstractionError("Invalid input. Only numbers are allowed")

    return a - b

suma(a=1, b=1) staticmethod

Function that returns addition of two integer numbers

Parameters:
  • a (float, default: 1 ) –

    first number to add. Defaults to 1.

  • b (float, default: 1 ) –

    second number to add. Defaults to 1.

Returns:
  • float( float ) –

    addition of a and b

Source code in src/services/calculator.py
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
@staticmethod
def suma(a: float = 1, b: float = 1) -> float:
    """Function that returns addition of two integer numbers

    Args:
        a (float, optional): first number to add. Defaults to 1.
        b (float, optional): second number to add. Defaults to 1.

    Returns:
        float: addition of a and b
    """
    if isinstance(a, str) or isinstance(b, str):
        raise AdditionError("Invalid input. Only numbers are allowed")

    return a + b

worker

This module enables the execution of tasks in different processes using Queues. The Executor class is a wrapper around the multiprocessing standard library. Ideal for CPU bound tasks that can be parallelized.

Executor

Executor class to manage worker processes.

Source code in src/services/worker.py
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
60
61
62
63
64
65
66
67
68
class Executor:
    """
    Executor class to manage worker processes.
    """

    def __init__(self, num_workers: int = 4) -> None:
        self.tasks_queue: Queue = Queue()
        self.workers = [
            Process(target=self.worker) for _ in range(num_workers)
        ]  # noqa: E501
        for worker in self.workers:
            worker.start()

    def add_task(self, task: Callable | None, *args: Sequence[Any]) -> None:
        """
        Add a task to the queue. Task should be a callable with its arguments.
        """
        self.tasks_queue.put((task, args))

    def worker(self) -> None:
        """
        Worker process: Retrieve tasks from the queue and execute them.
        """
        while True:
            task, args = self.tasks_queue.get()
            if task is None:
                # None is the signal to stop.
                break
            task(*args)

    def stop_workers(self) -> None:
        """
        Stop all worker processes.
        """
        for _ in self.workers:
            self.add_task(None)  # Signal to stop.
        for worker in self.workers:
            worker.join()

    def execute(self, task: Callable, *args: Sequence[Any]) -> None:
        """
        Execute method to add tasks. Now accepts a task and its arguments.
        """
        # Add tasks
        for arg in args:
            self.add_task(task, arg)

add_task(task, *args)

Add a task to the queue. Task should be a callable with its arguments.

Source code in src/services/worker.py
36
37
38
39
40
def add_task(self, task: Callable | None, *args: Sequence[Any]) -> None:
    """
    Add a task to the queue. Task should be a callable with its arguments.
    """
    self.tasks_queue.put((task, args))

execute(task, *args)

Execute method to add tasks. Now accepts a task and its arguments.

Source code in src/services/worker.py
62
63
64
65
66
67
68
def execute(self, task: Callable, *args: Sequence[Any]) -> None:
    """
    Execute method to add tasks. Now accepts a task and its arguments.
    """
    # Add tasks
    for arg in args:
        self.add_task(task, arg)

stop_workers()

Stop all worker processes.

Source code in src/services/worker.py
53
54
55
56
57
58
59
60
def stop_workers(self) -> None:
    """
    Stop all worker processes.
    """
    for _ in self.workers:
        self.add_task(None)  # Signal to stop.
    for worker in self.workers:
        worker.join()

worker()

Worker process: Retrieve tasks from the queue and execute them.

Source code in src/services/worker.py
42
43
44
45
46
47
48
49
50
51
def worker(self) -> None:
    """
    Worker process: Retrieve tasks from the queue and execute them.
    """
    while True:
        task, args = self.tasks_queue.get()
        if task is None:
            # None is the signal to stop.
            break
        task(*args)

task_function(data)

Example Job function that processes data.

Parameters:
  • data (Any) –

    Any input data required by the job.

Source code in src/services/worker.py
12
13
14
15
16
17
18
19
20
def task_function(data: Any) -> None:
    """
    Example Job function that processes data.

    Args:
        data (Any): Any input data required by the job.
    """
    print(f"Processing {data}")
    time.sleep(1)  # Simulate work