Skip to content

scheduled_tasks

Module with functions used to execute scheduled tasks.

See https://panel.holoviz.org/how_to/callbacks/schedule.html for details.

Classes:

Name Description
CleanFilesDB

Task action for cleaning temporary tables and files.
ResetGuestPassword

Task action for resetting guest user password.
Task

Generic task object.
TaskAction

Generic task action object.
TaskManager

Task manager object.
UploadDBToGCP

Task action for uploading database to Google Cloud Storage.

Attributes:

Name Type Description
log Logger

Module logger.

log module-attribute 

log: Logger = getLogger(__name__)

Module logger.

CleanFilesDB

Bases: TaskAction

Task action for cleaning temporary tables and files.

Its scope is to build a callable that will be executed when the task is triggered.

Methods:

Name Description
build_callable

Build and return the scheduled callable that cleans temporary
Source code in dlunch/scheduled_tasks.py 
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
class CleanFilesDB(TaskAction):
    """Task action for cleaning temporary tables and files.

    Its scope is to build a callable that will be executed when the task is
    triggered.
    """

    def build_callable(self, config: DictConfig) -> callable:
        """Build and return the scheduled callable that cleans temporary
        tables and files.

        Args:
            config (DictConfig): Hydra configuration dictionary.
        """

        # Set waiter
        waiter = core.Waiter(config=config)

        async def action_callable() -> None:
            """Scheduled callable that cleans temporary tables and files."""
            log.info(
                f"clean task (files and db) executed at {dt.datetime.now()}"
            )
            waiter.delete_files()
            waiter.clean_tables()

        return action_callable

build_callable 

build_callable(config: DictConfig) -> callable

Build and return the scheduled callable that cleans temporary tables and files.

Parameters:

Name Type Description Default
config DictConfig

Hydra configuration dictionary.

 required
Source code in dlunch/scheduled_tasks.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def build_callable(self, config: DictConfig) -> callable:
    """Build and return the scheduled callable that cleans temporary
    tables and files.

    Args:
        config (DictConfig): Hydra configuration dictionary.
    """

    # Set waiter
    waiter = core.Waiter(config=config)

    async def action_callable() -> None:
        """Scheduled callable that cleans temporary tables and files."""
        log.info(
            f"clean task (files and db) executed at {dt.datetime.now()}"
        )
        waiter.delete_files()
        waiter.clean_tables()

    return action_callable

ResetGuestPassword

Bases: TaskAction

Task action for resetting guest user password.

Its scope is to build a callable that will be executed when the task is triggered.

Methods:

Name Description
build_callable

Build and return the scheduled callable that resets the guest
Source code in dlunch/scheduled_tasks.py 
 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
103
104
class ResetGuestPassword(TaskAction):
    """Task action for resetting guest user password.

    Its scope is to build a callable that will be executed when the task is
    triggered.
    """

    def build_callable(self, config: DictConfig) -> callable:
        """Build and return the scheduled callable that resets the guest
        user password.

        Args:
            config (DictConfig): Hydra configuration dictionary.
        """

        # Set auth configurations
        auth_context = auth.AuthContext(config=config)

        async def action_callable() -> None:
            """Scheduled callable that resets the guest user password."""
            log.info(
                f"reset guest user password executed at {dt.datetime.now()}"
            )
            auth_context.database_connector.set_flag(
                id="reset_guest_user_password", value=True
            )
            auth_context.set_guest_user_password()

        return action_callable

build_callable 

build_callable(config: DictConfig) -> callable

Build and return the scheduled callable that resets the guest user password.

Parameters:

Name Type Description Default
config DictConfig

Hydra configuration dictionary.

 required
Source code in dlunch/scheduled_tasks.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def build_callable(self, config: DictConfig) -> callable:
    """Build and return the scheduled callable that resets the guest
    user password.

    Args:
        config (DictConfig): Hydra configuration dictionary.
    """

    # Set auth configurations
    auth_context = auth.AuthContext(config=config)

    async def action_callable() -> None:
        """Scheduled callable that resets the guest user password."""
        log.info(
            f"reset guest user password executed at {dt.datetime.now()}"
        )
        auth_context.database_connector.set_flag(
            id="reset_guest_user_password", value=True
        )
        auth_context.set_guest_user_password()

    return action_callable

Task

Generic task object.

Parameters:

Name Type Description Default
name str

Task name (used for logs).

 required
enabled bool

Flag that marks a task as enabled.

 required
hour int

Start hour (used only if also minute is not None).

 required
minute int

Start minute (used only if also hour is not None).

 required
period str

The period between executions. May be expressed as a timedelta or a string.

  • Week: '1w'
  • Day: '1d'
  • Hour: '1h'
  • Minute: '1m'
  • Second: '1s'
 required
actions list[TaskAction]

List of actions to be executed.

 required

Methods:

Name Description
__init__
build_callable

Build and return a callable that executes all actions
schedule_task

Schedule a task execution using Panel.

Attributes:

Name Type Description
actions list[TaskAction]

List of actions to be executed.
enabled bool

Flag that marks a task as enabled.
hour int

Start hour (used only if also minute is not None).
minute int

Start minute (used only if also hour is not None).
name str

Task name (used for logs).
period str

The period between executions.
Source code in dlunch/scheduled_tasks.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
class Task:
    """Generic task object.

    Args:
        name (str): Task name (used for logs).
        enabled (bool): Flag that marks a task as enabled.
        hour (int): Start hour (used only if also minute is not None).
        minute (int): Start minute (used only if also hour is not None).
        period (str): The period between executions.
            May be expressed as a timedelta or a string.

            * Week: `'1w'`
            * Day: `'1d'`
            * Hour: `'1h'`
            * Minute: `'1m'`
            * Second: `'1s'`
        actions (list[TaskAction]): List of actions to be executed.
    """

    def __init__(
        self,
        name: str,
        enabled: bool,
        hour: int | None,
        minute: int | None,
        period: str,
        actions: list[TaskAction],
    ) -> None:
        self.name: str = name
        """Task name (used for logs)."""
        self.enabled: bool = enabled
        """Flag that marks a task as enabled."""
        self.hour: int = hour
        """Start hour (used only if also minute is not None)."""
        self.minute: int = minute
        """Start minute (used only if also hour is not None)."""
        self.period: str = period
        """The period between executions.
            May be expressed as a timedelta or a string.

        * Week: `'1w'`
        * Day: `'1d'`
        * Hour: `'1h'`
        * Minute: `'1m'`
        * Second: `'1s'`
        """
        self.actions: list[TaskAction] = actions
        """List of actions to be executed."""

    def build_callable(self, config: DictConfig) -> callable:
        """Build and return a callable that executes all actions
        in the task.

        Args:
            config (DictConfig): Hydra configuration dictionary.
        """

        task_callables = [
            action.build_callable(config=config) for action in self.actions
        ]

        async def task_callable() -> None:
            for callable in task_callables:
                await callable()

        return task_callable

    def schedule_task(self, config: DictConfig) -> None:
        """Schedule a task execution using Panel.

        Args:
            config (DictConfig): Hydra configuration dictionary.

        If start time is not defined, the task will start immediately.
        If hour and minute are defined, the task will start at the specified
        time. If the time already passed, the task will start the next day.
        """
        if self.enabled:
            log.info(f"starting task '{self.name}'")
            if (self.hour is not None) and (self.minute is not None):
                start_time = dt.datetime.today().replace(
                    hour=self.hour,
                    minute=self.minute,
                )
                if start_time < dt.datetime.now():
                    start_time = start_time + dt.timedelta(days=1)
                log.info(
                    f"starting time: {start_time.strftime('%Y-%m-%d %H:%M')} - period: {self.period}"
                )
            else:
                start_time = None
                log.info(f"starting time: now - period: {self.period}")

            # Schedule task
            pn.state.schedule_task(
                name=self.name,
                callback=self.build_callable(config),
                period=self.period,
                at=start_time,
            )

actions instance-attribute 

actions: list[TaskAction] = actions

List of actions to be executed.

enabled instance-attribute 

enabled: bool = enabled

Flag that marks a task as enabled.

hour instance-attribute 

hour: int = hour

Start hour (used only if also minute is not None).

minute instance-attribute 

minute: int = minute

Start minute (used only if also hour is not None).

name instance-attribute 

name: str = name

Task name (used for logs).

period instance-attribute 

period: str = period

The period between executions. May be expressed as a timedelta or a string.

  • Week: '1w'
  • Day: '1d'
  • Hour: '1h'
  • Minute: '1m'
  • Second: '1s'

__init__ 

__init__(
    name: str,
    enabled: bool,
    hour: int | None,
    minute: int | None,
    period: str,
    actions: list[TaskAction],
) -> None
Source code in dlunch/scheduled_tasks.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
def __init__(
    self,
    name: str,
    enabled: bool,
    hour: int | None,
    minute: int | None,
    period: str,
    actions: list[TaskAction],
) -> None:
    self.name: str = name
    """Task name (used for logs)."""
    self.enabled: bool = enabled
    """Flag that marks a task as enabled."""
    self.hour: int = hour
    """Start hour (used only if also minute is not None)."""
    self.minute: int = minute
    """Start minute (used only if also hour is not None)."""
    self.period: str = period
    """The period between executions.
        May be expressed as a timedelta or a string.

    * Week: `'1w'`
    * Day: `'1d'`
    * Hour: `'1h'`
    * Minute: `'1m'`
    * Second: `'1s'`
    """
    self.actions: list[TaskAction] = actions
    """List of actions to be executed."""

build_callable 

build_callable(config: DictConfig) -> callable

Build and return a callable that executes all actions in the task.

Parameters:

Name Type Description Default
config DictConfig

Hydra configuration dictionary.

 required
Source code in dlunch/scheduled_tasks.py 
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
def build_callable(self, config: DictConfig) -> callable:
    """Build and return a callable that executes all actions
    in the task.

    Args:
        config (DictConfig): Hydra configuration dictionary.
    """

    task_callables = [
        action.build_callable(config=config) for action in self.actions
    ]

    async def task_callable() -> None:
        for callable in task_callables:
            await callable()

    return task_callable

schedule_task 

schedule_task(config: DictConfig) -> None

Schedule a task execution using Panel.

Parameters:

Name Type Description Default
config DictConfig

Hydra configuration dictionary.

 required

If start time is not defined, the task will start immediately. If hour and minute are defined, the task will start at the specified time. If the time already passed, the task will start the next day.

Source code in dlunch/scheduled_tasks.py 
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
def schedule_task(self, config: DictConfig) -> None:
    """Schedule a task execution using Panel.

    Args:
        config (DictConfig): Hydra configuration dictionary.

    If start time is not defined, the task will start immediately.
    If hour and minute are defined, the task will start at the specified
    time. If the time already passed, the task will start the next day.
    """
    if self.enabled:
        log.info(f"starting task '{self.name}'")
        if (self.hour is not None) and (self.minute is not None):
            start_time = dt.datetime.today().replace(
                hour=self.hour,
                minute=self.minute,
            )
            if start_time < dt.datetime.now():
                start_time = start_time + dt.timedelta(days=1)
            log.info(
                f"starting time: {start_time.strftime('%Y-%m-%d %H:%M')} - period: {self.period}"
            )
        else:
            start_time = None
            log.info(f"starting time: now - period: {self.period}")

        # Schedule task
        pn.state.schedule_task(
            name=self.name,
            callback=self.build_callable(config),
            period=self.period,
            at=start_time,
        )

TaskAction

Generic task action object.

Its scope is to build a callable that will be executed when the task is triggered.

Methods:

Name Description
build_callable

Build and return the scheduled callable that executes a dummy
Source code in dlunch/scheduled_tasks.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
class TaskAction:
    """Generic task action object.

    Its scope is to build a callable that will be executed when the task is
    triggered.
    """

    def build_callable(self, config: DictConfig) -> callable:
        """Build and return the scheduled callable that executes a dummy
        task action.

        It just logs a message.

        Args:
            config (DictConfig): Hydra configuration dictionary.
        """

        async def action_callable() -> None:
            """Scheduled callable that executes a dummy task action."""
            log.info(
                f"dummy task executed for {config.panel.gui.title.lower()}"
            )

        return action_callable

build_callable 

build_callable(config: DictConfig) -> callable

Build and return the scheduled callable that executes a dummy task action.

It just logs a message.

Parameters:

Name Type Description Default
config DictConfig

Hydra configuration dictionary.

 required
Source code in dlunch/scheduled_tasks.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
def build_callable(self, config: DictConfig) -> callable:
    """Build and return the scheduled callable that executes a dummy
    task action.

    It just logs a message.

    Args:
        config (DictConfig): Hydra configuration dictionary.
    """

    async def action_callable() -> None:
        """Scheduled callable that executes a dummy task action."""
        log.info(
            f"dummy task executed for {config.panel.gui.title.lower()}"
        )

    return action_callable

TaskManager

Task manager object.

Used to schedule all tasks given a list of tasks.

Parameters:

Name Type Description Default
config DictConfig

Hydra configuration dictionary.

 required
tasks list[Task]

List of tasks to be scheduled.

 required

Methods:

Name Description
__init__
log_tasks

Log all tasks defined in the collection.
schedule_all

Schedule all tasks in the collection.

Attributes:

Name Type Description
config DictConfig

Hydra configuration dictionary.
tasks list[Task]

List of tasks to be scheduled.
Source code in dlunch/scheduled_tasks.py 
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
class TaskManager:
    """Task manager object.

    Used to schedule all tasks given a list of tasks.

    Args:
        config (DictConfig): Hydra configuration dictionary.
        tasks (list[Task]): List of tasks to be scheduled.
    """

    def __init__(self, config: DictConfig, tasks: list[Task]) -> None:
        self.config: DictConfig = config
        """Hydra configuration dictionary."""
        self.tasks: list[Task] = tasks
        """List of tasks to be scheduled."""

    def log_tasks(self, enabled_only: bool = False) -> None:
        """Log all tasks defined in the collection.

        Args:
            enabled_only (bool): If True, only enabled tasks will be listed.
        """
        message = "listing tasks"
        if enabled_only:
            message += " (enabled only)"

        # Loop over tasks and actions
        for task in self.tasks:
            if enabled_only and not task.enabled:
                continue
            message += f"\ntask: {task.name}"
            message += (
                "\n"
                if enabled_only
                else f" ({'enabled' if task.enabled else 'disabled'})\n"
            )
            message += "  actions:"
            for action in task.actions:
                message += f"\n    {action.__class__.__name__}"

        log.info(message)

    def schedule_all(self) -> None:
        """Schedule all tasks in the collection."""
        for task in self.tasks:
            task.schedule_task(self.config)

config instance-attribute 

config: DictConfig = config

Hydra configuration dictionary.

tasks instance-attribute 

tasks: list[Task] = tasks

List of tasks to be scheduled.

__init__ 

__init__(config: DictConfig, tasks: list[Task]) -> None
Source code in dlunch/scheduled_tasks.py 
250
251
252
253
254
def __init__(self, config: DictConfig, tasks: list[Task]) -> None:
    self.config: DictConfig = config
    """Hydra configuration dictionary."""
    self.tasks: list[Task] = tasks
    """List of tasks to be scheduled."""

log_tasks 

log_tasks(enabled_only: bool = False) -> None

Log all tasks defined in the collection.

Parameters:

Name Type Description Default
enabled_only bool

If True, only enabled tasks will be listed.

 False
Source code in dlunch/scheduled_tasks.py 
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
def log_tasks(self, enabled_only: bool = False) -> None:
    """Log all tasks defined in the collection.

    Args:
        enabled_only (bool): If True, only enabled tasks will be listed.
    """
    message = "listing tasks"
    if enabled_only:
        message += " (enabled only)"

    # Loop over tasks and actions
    for task in self.tasks:
        if enabled_only and not task.enabled:
            continue
        message += f"\ntask: {task.name}"
        message += (
            "\n"
            if enabled_only
            else f" ({'enabled' if task.enabled else 'disabled'})\n"
        )
        message += "  actions:"
        for action in task.actions:
            message += f"\n    {action.__class__.__name__}"

    log.info(message)

schedule_all 

schedule_all() -> None

Schedule all tasks in the collection.

Source code in dlunch/scheduled_tasks.py 
282
283
284
285
def schedule_all(self) -> None:
    """Schedule all tasks in the collection."""
    for task in self.tasks:
        task.schedule_task(self.config)

UploadDBToGCP

Bases: TaskAction

Task action for uploading database to Google Cloud Storage.

Its scope is to build a callable that will be executed when the task is triggered.

Parameters:

Name Type Description Default
kwargs dict

Keyword arguments for the cloud.upload_to_gcloud function.

 {}

Methods:

Name Description
__init__
build_callable

Build and return the scheduled callable that uploads the database to

Attributes:

Name Type Description
gcp_kwargs
Source code in dlunch/scheduled_tasks.py 
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
class UploadDBToGCP(TaskAction):
    """Task action for uploading database to Google Cloud Storage.

    Its scope is to build a callable that will be executed when the task is
    triggered.

    Args:
        kwargs (dict): Keyword arguments for the cloud.upload_to_gcloud function.
    """

    def __init__(self, **kwargs) -> None:
        self.gcp_kwargs = kwargs

    def build_callable(self, config: DictConfig) -> callable:
        """Build and return the scheduled callable that uploads the database to
        Google Cloud Storage.

        Args:
            config (DictConfig): Hydra configuration dictionary.
        """

        async def action_callable() -> None:
            """Scheduled callable that uploads the database to Google Cloud Storage."""
            log.info(
                f"upload database to gcp storage executed at {dt.datetime.now()}"
            )
            cloud.upload_to_gcloud(**self.gcp_kwargs)

        return action_callable

gcp_kwargs instance-attribute 

gcp_kwargs = kwargs

__init__ 

__init__(**kwargs) -> None
Source code in dlunch/scheduled_tasks.py 
117
118
def __init__(self, **kwargs) -> None:
    self.gcp_kwargs = kwargs

build_callable 

build_callable(config: DictConfig) -> callable

Build and return the scheduled callable that uploads the database to Google Cloud Storage.

Parameters:

Name Type Description Default
config DictConfig

Hydra configuration dictionary.

 required
Source code in dlunch/scheduled_tasks.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def build_callable(self, config: DictConfig) -> callable:
    """Build and return the scheduled callable that uploads the database to
    Google Cloud Storage.

    Args:
        config (DictConfig): Hydra configuration dictionary.
    """

    async def action_callable() -> None:
        """Scheduled callable that uploads the database to Google Cloud Storage."""
        log.info(
            f"upload database to gcp storage executed at {dt.datetime.now()}"
        )
        cloud.upload_to_gcloud(**self.gcp_kwargs)

    return action_callable