We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Tasks API `start_timestamp` is not set, `task_duration` is wrong
**Describe the bug**
I noticed that System Tasks durations in the web UI are not right. The source of that data is the tasks API, and it's being calculated incorrectly there. `start_timestamp` is not part of the task schema, so it's not possible to calculate it yourself either.

**To Reproduce**
Check System Tasks page in admin interface, or `GET /admin/system_tasks/`

**Expected behavior**
Duration is correct, and `start_timestamp` is set

**Screenshots**
![image](https://user-images.githubusercontent.com/16788697/224092395-bacc6038-e263-41bf-801b-c0b3c00de36f.png)
![image](https://user-images.githubusercontent.com/16788697/224098644-eb942b08-9918-4cbb-9068-d99172dd62c1.png)

**Version and Deployment (please complete the following information):**
 - authentik version: `gh-main`
 - Deployment: docker-compose


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `authentik/events/monitored_tasks.py`
Content:
```
1 """Monitored tasks"""
2 from dataclasses import dataclass, field
3 from datetime import datetime
4 from enum import Enum
5 from timeit import default_timer
6 from typing import Any, Optional
7 
8 from celery import Task
9 from django.core.cache import cache
10 from django.utils.translation import gettext_lazy as _
11 from structlog.stdlib import get_logger
12 
13 from authentik.events.apps import GAUGE_TASKS
14 from authentik.events.models import Event, EventAction
15 from authentik.lib.utils.errors import exception_to_string
16 
17 LOGGER = get_logger()
18 CACHE_KEY_PREFIX = "goauthentik.io/events/tasks/"
19 
20 
21 class TaskResultStatus(Enum):
22     """Possible states of tasks"""
23 
24     SUCCESSFUL = 1
25     WARNING = 2
26     ERROR = 4
27     UNKNOWN = 8
28 
29 
30 @dataclass
31 class TaskResult:
32     """Result of a task run, this class is created by the task itself
33     and used by self.set_status"""
34 
35     status: TaskResultStatus
36 
37     messages: list[str] = field(default_factory=list)
38 
39     # Optional UID used in cache for tasks that run in different instances
40     uid: Optional[str] = field(default=None)
41 
42     def with_error(self, exc: Exception) -> "TaskResult":
43         """Since errors might not always be pickle-able, set the traceback"""
44         self.messages.append(str(exc))
45         return self
46 
47 
48 @dataclass
49 class TaskInfo:
50     """Info about a task run"""
51 
52     task_name: str
53     start_timestamp: float
54     finish_timestamp: float
55     finish_time: datetime
56 
57     result: TaskResult
58 
59     task_call_module: str
60     task_call_func: str
61     task_call_args: list[Any] = field(default_factory=list)
62     task_call_kwargs: dict[str, Any] = field(default_factory=dict)
63 
64     task_description: Optional[str] = field(default=None)
65 
66     @staticmethod
67     def all() -> dict[str, "TaskInfo"]:
68         """Get all TaskInfo objects"""
69         return cache.get_many(cache.keys(CACHE_KEY_PREFIX + "*"))
70 
71     @staticmethod
72     def by_name(name: str) -> Optional["TaskInfo"]:
73         """Get TaskInfo Object by name"""
74         return cache.get(CACHE_KEY_PREFIX + name, None)
75 
76     def delete(self):
77         """Delete task info from cache"""
78         return cache.delete(CACHE_KEY_PREFIX + self.task_name)
79 
80     def update_metrics(self):
81         """Update prometheus metrics"""
82         start = default_timer()
83         if hasattr(self, "start_timestamp"):
84             start = self.start_timestamp
85         try:
86             duration = max(self.finish_timestamp - start, 0)
87         except TypeError:
88             duration = 0
89         GAUGE_TASKS.labels(
90             task_name=self.task_name,
91             task_uid=self.result.uid or "",
92             status=self.result.status,
93         ).set(duration)
94 
95     def save(self, timeout_hours=6):
96         """Save task into cache"""
97         key = CACHE_KEY_PREFIX + self.task_name
98         if self.result.uid:
99             key += f":{self.result.uid}"
100             self.task_name += f":{self.result.uid}"
101         self.update_metrics()
102         cache.set(key, self, timeout=timeout_hours * 60 * 60)
103 
104 
105 class MonitoredTask(Task):
106     """Task which can save its state to the cache"""
107 
108     # For tasks that should only be listed if they failed, set this to False
109     save_on_success: bool
110 
111     _result: Optional[TaskResult]
112 
113     _uid: Optional[str]
114 
115     def __init__(self, *args, **kwargs) -> None:
116         super().__init__(*args, **kwargs)
117         self.save_on_success = True
118         self._uid = None
119         self._result = None
120         self.result_timeout_hours = 6
121         self.start = default_timer()
122 
123     def set_uid(self, uid: str):
124         """Set UID, so in the case of an unexpected error its saved correctly"""
125         self._uid = uid
126 
127     def set_status(self, result: TaskResult):
128         """Set result for current run, will overwrite previous result."""
129         self._result = result
130 
131     # pylint: disable=too-many-arguments
132     def after_return(self, status, retval, task_id, args: list[Any], kwargs: dict[str, Any], einfo):
133         super().after_return(status, retval, task_id, args, kwargs, einfo=einfo)
134         if not self._result:
135             return
136         if not self._result.uid:
137             self._result.uid = self._uid
138         info = TaskInfo(
139             task_name=self.__name__,
140             task_description=self.__doc__,
141             start_timestamp=self.start,
142             finish_timestamp=default_timer(),
143             finish_time=datetime.now(),
144             result=self._result,
145             task_call_module=self.__module__,
146             task_call_func=self.__name__,
147             task_call_args=args,
148             task_call_kwargs=kwargs,
149         )
150         if self._result.status == TaskResultStatus.SUCCESSFUL and not self.save_on_success:
151             info.delete()
152             return
153         info.save(self.result_timeout_hours)
154 
155     # pylint: disable=too-many-arguments
156     def on_failure(self, exc, task_id, args, kwargs, einfo):
157         super().on_failure(exc, task_id, args, kwargs, einfo=einfo)
158         if not self._result:
159             self._result = TaskResult(status=TaskResultStatus.ERROR, messages=[str(exc)])
160         if not self._result.uid:
161             self._result.uid = self._uid
162         TaskInfo(
163             task_name=self.__name__,
164             task_description=self.__doc__,
165             start_timestamp=self.start,
166             finish_timestamp=default_timer(),
167             finish_time=datetime.now(),
168             result=self._result,
169             task_call_module=self.__module__,
170             task_call_func=self.__name__,
171             task_call_args=args,
172             task_call_kwargs=kwargs,
173         ).save(self.result_timeout_hours)
174         Event.new(
175             EventAction.SYSTEM_TASK_EXCEPTION,
176             message=f"Task {self.__name__} encountered an error: {exception_to_string(exc)}",
177         ).save()
178 
179     def run(self, *args, **kwargs):
180         raise NotImplementedError
181 
182 
183 def prefill_task(func):
184     """Ensure a task's details are always in cache, so it can always be triggered via API"""
185     status = TaskInfo.by_name(func.__name__)
186     if status:
187         return func
188     TaskInfo(
189         task_name=func.__name__,
190         task_description=func.__doc__,
191         result=TaskResult(TaskResultStatus.UNKNOWN, messages=[_("Task has not been run yet.")]),
192         task_call_module=func.__module__,
193         task_call_func=func.__name__,
194         # We don't have real values for these attributes but they cannot be null
195         start_timestamp=default_timer(),
196         finish_timestamp=default_timer(),
197         finish_time=datetime.now(),
198     ).save(86400)
199     LOGGER.debug("prefilled task", task_name=func.__name__)
200     return func
201
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/authentik/events/monitored_tasks.py b/authentik/events/monitored_tasks.py
--- a/authentik/events/monitored_tasks.py
+++ b/authentik/events/monitored_tasks.py
@@ -111,6 +111,7 @@
     _result: Optional[TaskResult]
 
     _uid: Optional[str]
+    start: Optional[float] = None
 
     def __init__(self, *args, **kwargs) -> None:
         super().__init__(*args, **kwargs)
@@ -118,7 +119,6 @@
         self._uid = None
         self._result = None
         self.result_timeout_hours = 6
-        self.start = default_timer()
 
     def set_uid(self, uid: str):
         """Set UID, so in the case of an unexpected error its saved correctly"""
@@ -128,6 +128,10 @@
         """Set result for current run, will overwrite previous result."""
         self._result = result
 
+    def before_start(self, task_id, args, kwargs):
+        self.start = default_timer()
+        return super().before_start(task_id, args, kwargs)
+
     # pylint: disable=too-many-arguments
     def after_return(self, status, retval, task_id, args: list[Any], kwargs: dict[str, Any], einfo):
         super().after_return(status, retval, task_id, args, kwargs, einfo=einfo)
@@ -138,7 +142,7 @@
         info = TaskInfo(
             task_name=self.__name__,
             task_description=self.__doc__,
-            start_timestamp=self.start,
+            start_timestamp=self.start or default_timer(),
             finish_timestamp=default_timer(),
             finish_time=datetime.now(),
             result=self._result,
@@ -162,7 +166,7 @@
         TaskInfo(
             task_name=self.__name__,
             task_description=self.__doc__,
-            start_timestamp=self.start,
+            start_timestamp=self.start or default_timer(),
             finish_timestamp=default_timer(),
             finish_time=datetime.now(),
             result=self._result,

{"golden_diff": "diff --git a/authentik/events/monitored_tasks.py b/authentik/events/monitored_tasks.py\n--- a/authentik/events/monitored_tasks.py\n+++ b/authentik/events/monitored_tasks.py\n@@ -111,6 +111,7 @@\n     _result: Optional[TaskResult]\n \n     _uid: Optional[str]\n+    start: Optional[float] = None\n \n     def __init__(self, *args, **kwargs) -> None:\n         super().__init__(*args, **kwargs)\n@@ -118,7 +119,6 @@\n         self._uid = None\n         self._result = None\n         self.result_timeout_hours = 6\n-        self.start = default_timer()\n \n     def set_uid(self, uid: str):\n         \"\"\"Set UID, so in the case of an unexpected error its saved correctly\"\"\"\n@@ -128,6 +128,10 @@\n         \"\"\"Set result for current run, will overwrite previous result.\"\"\"\n         self._result = result\n \n+    def before_start(self, task_id, args, kwargs):\n+        self.start = default_timer()\n+        return super().before_start(task_id, args, kwargs)\n+\n     # pylint: disable=too-many-arguments\n     def after_return(self, status, retval, task_id, args: list[Any], kwargs: dict[str, Any], einfo):\n         super().after_return(status, retval, task_id, args, kwargs, einfo=einfo)\n@@ -138,7 +142,7 @@\n         info = TaskInfo(\n             task_name=self.__name__,\n             task_description=self.__doc__,\n-            start_timestamp=self.start,\n+            start_timestamp=self.start or default_timer(),\n             finish_timestamp=default_timer(),\n             finish_time=datetime.now(),\n             result=self._result,\n@@ -162,7 +166,7 @@\n         TaskInfo(\n             task_name=self.__name__,\n             task_description=self.__doc__,\n-            start_timestamp=self.start,\n+            start_timestamp=self.start or default_timer(),\n             finish_timestamp=default_timer(),\n             finish_time=datetime.now(),\n             result=self._result,\n", "issue": "Tasks API `start_timestamp` is not set, `task_duration` is wrong\n**Describe the bug**\r\nI noticed that System Tasks durations in the web UI are not right. The source of that data is the tasks API, and it's being calculated incorrectly there. `start_timestamp` is not part of the task schema, so it's not possible to calculate it yourself either.\r\n\r\n**To Reproduce**\r\nCheck System Tasks page in admin interface, or `GET /admin/system_tasks/`\r\n\r\n**Expected behavior**\r\nDuration is correct, and `start_timestamp` is set\r\n\r\n**Screenshots**\r\n![image](https://user-images.githubusercontent.com/16788697/224092395-bacc6038-e263-41bf-801b-c0b3c00de36f.png)\r\n![image](https://user-images.githubusercontent.com/16788697/224098644-eb942b08-9918-4cbb-9068-d99172dd62c1.png)\r\n\r\n**Version and Deployment (please complete the following information):**\r\n - authentik version: `gh-main`\r\n - Deployment: docker-compose\r\n\n", "before_files": [{"content": "\"\"\"Monitored tasks\"\"\"\nfrom dataclasses import dataclass, field\nfrom datetime import datetime\nfrom enum import Enum\nfrom timeit import default_timer\nfrom typing import Any, Optional\n\nfrom celery import Task\nfrom django.core.cache import cache\nfrom django.utils.translation import gettext_lazy as _\nfrom structlog.stdlib import get_logger\n\nfrom authentik.events.apps import GAUGE_TASKS\nfrom authentik.events.models import Event, EventAction\nfrom authentik.lib.utils.errors import exception_to_string\n\nLOGGER = get_logger()\nCACHE_KEY_PREFIX = \"goauthentik.io/events/tasks/\"\n\n\nclass TaskResultStatus(Enum):\n    \"\"\"Possible states of tasks\"\"\"\n\n    SUCCESSFUL = 1\n    WARNING = 2\n    ERROR = 4\n    UNKNOWN = 8\n\n\n@dataclass\nclass TaskResult:\n    \"\"\"Result of a task run, this class is created by the task itself\n    and used by self.set_status\"\"\"\n\n    status: TaskResultStatus\n\n    messages: list[str] = field(default_factory=list)\n\n    # Optional UID used in cache for tasks that run in different instances\n    uid: Optional[str] = field(default=None)\n\n    def with_error(self, exc: Exception) -> \"TaskResult\":\n        \"\"\"Since errors might not always be pickle-able, set the traceback\"\"\"\n        self.messages.append(str(exc))\n        return self\n\n\n@dataclass\nclass TaskInfo:\n    \"\"\"Info about a task run\"\"\"\n\n    task_name: str\n    start_timestamp: float\n    finish_timestamp: float\n    finish_time: datetime\n\n    result: TaskResult\n\n    task_call_module: str\n    task_call_func: str\n    task_call_args: list[Any] = field(default_factory=list)\n    task_call_kwargs: dict[str, Any] = field(default_factory=dict)\n\n    task_description: Optional[str] = field(default=None)\n\n    @staticmethod\n    def all() -> dict[str, \"TaskInfo\"]:\n        \"\"\"Get all TaskInfo objects\"\"\"\n        return cache.get_many(cache.keys(CACHE_KEY_PREFIX + \"*\"))\n\n    @staticmethod\n    def by_name(name: str) -> Optional[\"TaskInfo\"]:\n        \"\"\"Get TaskInfo Object by name\"\"\"\n        return cache.get(CACHE_KEY_PREFIX + name, None)\n\n    def delete(self):\n        \"\"\"Delete task info from cache\"\"\"\n        return cache.delete(CACHE_KEY_PREFIX + self.task_name)\n\n    def update_metrics(self):\n        \"\"\"Update prometheus metrics\"\"\"\n        start = default_timer()\n        if hasattr(self, \"start_timestamp\"):\n            start = self.start_timestamp\n        try:\n            duration = max(self.finish_timestamp - start, 0)\n        except TypeError:\n            duration = 0\n        GAUGE_TASKS.labels(\n            task_name=self.task_name,\n            task_uid=self.result.uid or \"\",\n            status=self.result.status,\n        ).set(duration)\n\n    def save(self, timeout_hours=6):\n        \"\"\"Save task into cache\"\"\"\n        key = CACHE_KEY_PREFIX + self.task_name\n        if self.result.uid:\n            key += f\":{self.result.uid}\"\n            self.task_name += f\":{self.result.uid}\"\n        self.update_metrics()\n        cache.set(key, self, timeout=timeout_hours * 60 * 60)\n\n\nclass MonitoredTask(Task):\n    \"\"\"Task which can save its state to the cache\"\"\"\n\n    # For tasks that should only be listed if they failed, set this to False\n    save_on_success: bool\n\n    _result: Optional[TaskResult]\n\n    _uid: Optional[str]\n\n    def __init__(self, *args, **kwargs) -> None:\n        super().__init__(*args, **kwargs)\n        self.save_on_success = True\n        self._uid = None\n        self._result = None\n        self.result_timeout_hours = 6\n        self.start = default_timer()\n\n    def set_uid(self, uid: str):\n        \"\"\"Set UID, so in the case of an unexpected error its saved correctly\"\"\"\n        self._uid = uid\n\n    def set_status(self, result: TaskResult):\n        \"\"\"Set result for current run, will overwrite previous result.\"\"\"\n        self._result = result\n\n    # pylint: disable=too-many-arguments\n    def after_return(self, status, retval, task_id, args: list[Any], kwargs: dict[str, Any], einfo):\n        super().after_return(status, retval, task_id, args, kwargs, einfo=einfo)\n        if not self._result:\n            return\n        if not self._result.uid:\n            self._result.uid = self._uid\n        info = TaskInfo(\n            task_name=self.__name__,\n            task_description=self.__doc__,\n            start_timestamp=self.start,\n            finish_timestamp=default_timer(),\n            finish_time=datetime.now(),\n            result=self._result,\n            task_call_module=self.__module__,\n            task_call_func=self.__name__,\n            task_call_args=args,\n            task_call_kwargs=kwargs,\n        )\n        if self._result.status == TaskResultStatus.SUCCESSFUL and not self.save_on_success:\n            info.delete()\n            return\n        info.save(self.result_timeout_hours)\n\n    # pylint: disable=too-many-arguments\n    def on_failure(self, exc, task_id, args, kwargs, einfo):\n        super().on_failure(exc, task_id, args, kwargs, einfo=einfo)\n        if not self._result:\n            self._result = TaskResult(status=TaskResultStatus.ERROR, messages=[str(exc)])\n        if not self._result.uid:\n            self._result.uid = self._uid\n        TaskInfo(\n            task_name=self.__name__,\n            task_description=self.__doc__,\n            start_timestamp=self.start,\n            finish_timestamp=default_timer(),\n            finish_time=datetime.now(),\n            result=self._result,\n            task_call_module=self.__module__,\n            task_call_func=self.__name__,\n            task_call_args=args,\n            task_call_kwargs=kwargs,\n        ).save(self.result_timeout_hours)\n        Event.new(\n            EventAction.SYSTEM_TASK_EXCEPTION,\n            message=f\"Task {self.__name__} encountered an error: {exception_to_string(exc)}\",\n        ).save()\n\n    def run(self, *args, **kwargs):\n        raise NotImplementedError\n\n\ndef prefill_task(func):\n    \"\"\"Ensure a task's details are always in cache, so it can always be triggered via API\"\"\"\n    status = TaskInfo.by_name(func.__name__)\n    if status:\n        return func\n    TaskInfo(\n        task_name=func.__name__,\n        task_description=func.__doc__,\n        result=TaskResult(TaskResultStatus.UNKNOWN, messages=[_(\"Task has not been run yet.\")]),\n        task_call_module=func.__module__,\n        task_call_func=func.__name__,\n        # We don't have real values for these attributes but they cannot be null\n        start_timestamp=default_timer(),\n        finish_timestamp=default_timer(),\n        finish_time=datetime.now(),\n    ).save(86400)\n    LOGGER.debug(\"prefilled task\", task_name=func.__name__)\n    return func\n", "path": "authentik/events/monitored_tasks.py"}], "after_files": [{"content": "\"\"\"Monitored tasks\"\"\"\nfrom dataclasses import dataclass, field\nfrom datetime import datetime\nfrom enum import Enum\nfrom timeit import default_timer\nfrom typing import Any, Optional\n\nfrom celery import Task\nfrom django.core.cache import cache\nfrom django.utils.translation import gettext_lazy as _\nfrom structlog.stdlib import get_logger\n\nfrom authentik.events.apps import GAUGE_TASKS\nfrom authentik.events.models import Event, EventAction\nfrom authentik.lib.utils.errors import exception_to_string\n\nLOGGER = get_logger()\nCACHE_KEY_PREFIX = \"goauthentik.io/events/tasks/\"\n\n\nclass TaskResultStatus(Enum):\n    \"\"\"Possible states of tasks\"\"\"\n\n    SUCCESSFUL = 1\n    WARNING = 2\n    ERROR = 4\n    UNKNOWN = 8\n\n\n@dataclass\nclass TaskResult:\n    \"\"\"Result of a task run, this class is created by the task itself\n    and used by self.set_status\"\"\"\n\n    status: TaskResultStatus\n\n    messages: list[str] = field(default_factory=list)\n\n    # Optional UID used in cache for tasks that run in different instances\n    uid: Optional[str] = field(default=None)\n\n    def with_error(self, exc: Exception) -> \"TaskResult\":\n        \"\"\"Since errors might not always be pickle-able, set the traceback\"\"\"\n        self.messages.append(str(exc))\n        return self\n\n\n@dataclass\nclass TaskInfo:\n    \"\"\"Info about a task run\"\"\"\n\n    task_name: str\n    start_timestamp: float\n    finish_timestamp: float\n    finish_time: datetime\n\n    result: TaskResult\n\n    task_call_module: str\n    task_call_func: str\n    task_call_args: list[Any] = field(default_factory=list)\n    task_call_kwargs: dict[str, Any] = field(default_factory=dict)\n\n    task_description: Optional[str] = field(default=None)\n\n    @staticmethod\n    def all() -> dict[str, \"TaskInfo\"]:\n        \"\"\"Get all TaskInfo objects\"\"\"\n        return cache.get_many(cache.keys(CACHE_KEY_PREFIX + \"*\"))\n\n    @staticmethod\n    def by_name(name: str) -> Optional[\"TaskInfo\"]:\n        \"\"\"Get TaskInfo Object by name\"\"\"\n        return cache.get(CACHE_KEY_PREFIX + name, None)\n\n    def delete(self):\n        \"\"\"Delete task info from cache\"\"\"\n        return cache.delete(CACHE_KEY_PREFIX + self.task_name)\n\n    def update_metrics(self):\n        \"\"\"Update prometheus metrics\"\"\"\n        start = default_timer()\n        if hasattr(self, \"start_timestamp\"):\n            start = self.start_timestamp\n        try:\n            duration = max(self.finish_timestamp - start, 0)\n        except TypeError:\n            duration = 0\n        GAUGE_TASKS.labels(\n            task_name=self.task_name,\n            task_uid=self.result.uid or \"\",\n            status=self.result.status,\n        ).set(duration)\n\n    def save(self, timeout_hours=6):\n        \"\"\"Save task into cache\"\"\"\n        key = CACHE_KEY_PREFIX + self.task_name\n        if self.result.uid:\n            key += f\":{self.result.uid}\"\n            self.task_name += f\":{self.result.uid}\"\n        self.update_metrics()\n        cache.set(key, self, timeout=timeout_hours * 60 * 60)\n\n\nclass MonitoredTask(Task):\n    \"\"\"Task which can save its state to the cache\"\"\"\n\n    # For tasks that should only be listed if they failed, set this to False\n    save_on_success: bool\n\n    _result: Optional[TaskResult]\n\n    _uid: Optional[str]\n    start: Optional[float] = None\n\n    def __init__(self, *args, **kwargs) -> None:\n        super().__init__(*args, **kwargs)\n        self.save_on_success = True\n        self._uid = None\n        self._result = None\n        self.result_timeout_hours = 6\n\n    def set_uid(self, uid: str):\n        \"\"\"Set UID, so in the case of an unexpected error its saved correctly\"\"\"\n        self._uid = uid\n\n    def set_status(self, result: TaskResult):\n        \"\"\"Set result for current run, will overwrite previous result.\"\"\"\n        self._result = result\n\n    def before_start(self, task_id, args, kwargs):\n        self.start = default_timer()\n        return super().before_start(task_id, args, kwargs)\n\n    # pylint: disable=too-many-arguments\n    def after_return(self, status, retval, task_id, args: list[Any], kwargs: dict[str, Any], einfo):\n        super().after_return(status, retval, task_id, args, kwargs, einfo=einfo)\n        if not self._result:\n            return\n        if not self._result.uid:\n            self._result.uid = self._uid\n        info = TaskInfo(\n            task_name=self.__name__,\n            task_description=self.__doc__,\n            start_timestamp=self.start or default_timer(),\n            finish_timestamp=default_timer(),\n            finish_time=datetime.now(),\n            result=self._result,\n            task_call_module=self.__module__,\n            task_call_func=self.__name__,\n            task_call_args=args,\n            task_call_kwargs=kwargs,\n        )\n        if self._result.status == TaskResultStatus.SUCCESSFUL and not self.save_on_success:\n            info.delete()\n            return\n        info.save(self.result_timeout_hours)\n\n    # pylint: disable=too-many-arguments\n    def on_failure(self, exc, task_id, args, kwargs, einfo):\n        super().on_failure(exc, task_id, args, kwargs, einfo=einfo)\n        if not self._result:\n            self._result = TaskResult(status=TaskResultStatus.ERROR, messages=[str(exc)])\n        if not self._result.uid:\n            self._result.uid = self._uid\n        TaskInfo(\n            task_name=self.__name__,\n            task_description=self.__doc__,\n            start_timestamp=self.start or default_timer(),\n            finish_timestamp=default_timer(),\n            finish_time=datetime.now(),\n            result=self._result,\n            task_call_module=self.__module__,\n            task_call_func=self.__name__,\n            task_call_args=args,\n            task_call_kwargs=kwargs,\n        ).save(self.result_timeout_hours)\n        Event.new(\n            EventAction.SYSTEM_TASK_EXCEPTION,\n            message=f\"Task {self.__name__} encountered an error: {exception_to_string(exc)}\",\n        ).save()\n\n    def run(self, *args, **kwargs):\n        raise NotImplementedError\n\n\ndef prefill_task(func):\n    \"\"\"Ensure a task's details are always in cache, so it can always be triggered via API\"\"\"\n    status = TaskInfo.by_name(func.__name__)\n    if status:\n        return func\n    TaskInfo(\n        task_name=func.__name__,\n        task_description=func.__doc__,\n        result=TaskResult(TaskResultStatus.UNKNOWN, messages=[_(\"Task has not been run yet.\")]),\n        task_call_module=func.__module__,\n        task_call_func=func.__name__,\n        # We don't have real values for these attributes but they cannot be null\n        start_timestamp=default_timer(),\n        finish_timestamp=default_timer(),\n        finish_time=datetime.now(),\n    ).save(86400)\n    LOGGER.debug(\"prefilled task\", task_name=func.__name__)\n    return func\n", "path": "authentik/events/monitored_tasks.py"}]}