Skip to content

main

ies_pi_worker.main

Task

A task is a function that's executed in background. It's defined by an id and a Future. When the task is initialized, started_at is populated with the current timestamp, and as soon as the future is done, finished_at is populated in the same way.

Source code in src/ies_pi_worker/main.py 
 9
10
11
12
13
14
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
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
class Task:
    """
    A task is a function that's executed in background.
    It's defined by an `id` and a `Future`.
    When the task is initialized, `started_at` is populated with the current
    timestamp, and as soon as the future is done, `finished_at` is populated in
    the same way.
    """    
    def __init__(self, id: str, future: Future):
        self.id = id
        self.future = future
        self.started_at = datetime.now()
        self.finished_at = None

        self.future.add_done_callback(self._finish())

    def finished(self) -> bool:
        """Returns whether the task has been finished"""
        return self.finished_at is not None

    def elapsed(self) -> timedelta:
        """Returns the amount of time it took for the task to complete"""
        return self.finished_at - self.started_at

    def to_dict(self, timeout=0.01) -> dict:
        """
        to_dict returns a dict representation of the task. Along with `id`, 
        `started_at` and `finished_at` it also returns the result or the error
        of the task, if it has finished.

        Args:
            timeout (float, optional): The number of seconds to wait for the 
                                       result if the task isn't finished yet.
                                       Defaults to 0.01

        Returns:
            dict: {
                "id": str,
                "started_at": datetime,
                "finished_at": datetime,
                "result": any | None,
                "error": str | None,
            }
        """        
        result = None
        error = None

        try:
            result = self.future.result(timeout)
        except(TimeoutError):
            pass
        except Exception as e:
            import traceback
            traceback.print_exception(e)
            error = str(e)

        return {
            "id": self.id,
            "started_at": self.started_at.strftime(date_format),
            "finished_at": None if self.finished_at is None else 
                self.finished_at.strftime(date_format),
            "result": result,
            "error": error,
        }

    def _finish(self):
        """Returns an unbounded method used to set `finished_at`"""        
        def finish(_):
            self.finished_at = datetime.now()
        return finish

elapsed 

elapsed() -> timedelta

Returns the amount of time it took for the task to complete

Source code in src/ies_pi_worker/main.py 
29
30
31
def elapsed(self) -> timedelta:
    """Returns the amount of time it took for the task to complete"""
    return self.finished_at - self.started_at

finished 

finished() -> bool

Returns whether the task has been finished

Source code in src/ies_pi_worker/main.py 
25
26
27
def finished(self) -> bool:
    """Returns whether the task has been finished"""
    return self.finished_at is not None

to_dict 

to_dict(timeout=0.01) -> dict

to_dict returns a dict representation of the task. Along with id, started_at and finished_at it also returns the result or the error of the task, if it has finished.

Parameters:

Name Type Description Default
timeout float

The number of seconds to wait for the result if the task isn't finished yet. Defaults to 0.01

 0.01

Returns:

Name Type Description
dict dict

{ "id": str, "started_at": datetime, "finished_at": datetime, "result": any | None, "error": str | None,
dict

}
Source code in src/ies_pi_worker/main.py 
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
def to_dict(self, timeout=0.01) -> dict:
    """
    to_dict returns a dict representation of the task. Along with `id`, 
    `started_at` and `finished_at` it also returns the result or the error
    of the task, if it has finished.

    Args:
        timeout (float, optional): The number of seconds to wait for the 
                                   result if the task isn't finished yet.
                                   Defaults to 0.01

    Returns:
        dict: {
            "id": str,
            "started_at": datetime,
            "finished_at": datetime,
            "result": any | None,
            "error": str | None,
        }
    """        
    result = None
    error = None

    try:
        result = self.future.result(timeout)
    except(TimeoutError):
        pass
    except Exception as e:
        import traceback
        traceback.print_exception(e)
        error = str(e)

    return {
        "id": self.id,
        "started_at": self.started_at.strftime(date_format),
        "finished_at": None if self.finished_at is None else 
            self.finished_at.strftime(date_format),
        "result": result,
        "error": error,
    }

Worker

Worker is a wrapper around concurrent.futures that allows to run tasks and retrieve their results later.

It saves the tasks in memory, so they don't persist.

The tasks are short-lived, and after they're finished, they are available for a limited amount of time.

Source code in src/ies_pi_worker/main.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
class Worker:
    """
    Worker is a wrapper around `concurrent.futures` that allows to run tasks
    and retrieve their results later.

    It saves the tasks in memory, so they don't persist.

    The tasks are short-lived, and after they're finished, they are available 
    for a limited amount of time.
    """    
    def __init__(self, delete_tasks_after=default_delete_tasks_after):
        self.delete_tasks_after=delete_tasks_after
        self.lock = Lock()
        self.executor = ProcessPoolExecutor()
        self.tasks: dict[str, Task] = {}

    def shutdown(self):
        """shutdown cleans up the resources of the executor"""        
        self.executor.shutdown()

    def run(self, fn, *args, **kwargs) -> Task:
        """
        run allows to run any arbitrary task, much like the `submit` method of
        a `concurrent.futures` executor, but returns a `Task`, that will automatically 
        update with the result of the task.

        Args:
            fn (function): a callable to be executed as fn(*args, **kwargs).
                           its return values will be stored in the returned Task.

        Returns:
            Task: a Task that will automatically update with the result of the function.
        """        
        id = str(uuid4())
        future = self.executor.submit(fn, *args, **kwargs)

        with self.lock:
            task = Task(id, future)
            self.tasks[id] = task
            return task

    def get(self, id: str) -> Task:
        """
        get retrieves a Task by their id.

        Args:
            id (str): The id of the task.

        Returns:
            Task: The Task obtained
        """        
        self._cleanup()
        with self.lock:
            task = self.tasks[id]
            return task

    def _cleanup(self):
        """cleanup removes tasks older than self.delete_tasks_after"""        
        with self.lock:
            for id, task in self.tasks.copy().items():
                if not task.finished():
                    continue
                if task.elapsed() > self.delete_tasks_after:
                    self.tasks.pop(id)

get 

get(id: str) -> Task

get retrieves a Task by their id.

Parameters:

Name Type Description Default
id str

The id of the task.

 required

Returns:

Name Type Description
Task Task

The Task obtained
Source code in src/ies_pi_worker/main.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
def get(self, id: str) -> Task:
    """
    get retrieves a Task by their id.

    Args:
        id (str): The id of the task.

    Returns:
        Task: The Task obtained
    """        
    self._cleanup()
    with self.lock:
        task = self.tasks[id]
        return task

run 

run(fn, *args, **kwargs) -> Task

run allows to run any arbitrary task, much like the submit method of a concurrent.futures executor, but returns a Task, that will automatically update with the result of the task.

Parameters:

Name Type Description Default
fn function

a callable to be executed as fn(args, *kwargs). its return values will be stored in the returned Task.

 required

Returns:

Name Type Description
Task Task

a Task that will automatically update with the result of the function.
Source code in src/ies_pi_worker/main.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def run(self, fn, *args, **kwargs) -> Task:
    """
    run allows to run any arbitrary task, much like the `submit` method of
    a `concurrent.futures` executor, but returns a `Task`, that will automatically 
    update with the result of the task.

    Args:
        fn (function): a callable to be executed as fn(*args, **kwargs).
                       its return values will be stored in the returned Task.

    Returns:
        Task: a Task that will automatically update with the result of the function.
    """        
    id = str(uuid4())
    future = self.executor.submit(fn, *args, **kwargs)

    with self.lock:
        task = Task(id, future)
        self.tasks[id] = task
        return task

shutdown 

shutdown()

shutdown cleans up the resources of the executor

Source code in src/ies_pi_worker/main.py 
96
97
98
def shutdown(self):
    """shutdown cleans up the resources of the executor"""        
    self.executor.shutdown()