0
# Flask-APScheduler
1
2
A Flask extension that integrates the Advanced Python Scheduler (APScheduler) with Flask applications, providing comprehensive job scheduling capabilities for web applications. It enables developers to schedule background tasks, periodic jobs, and delayed executions directly within Flask applications with support for multiple scheduler backends, REST API management, and authentication.
3
4
## Package Information
5
6
- **Package Name**: Flask-APScheduler
7
- **Package Type**: pypi
8
- **Language**: Python
9
- **Installation**: `pip install Flask-APScheduler`
10
11
## Core Imports
12
13
```python
14
from flask_apscheduler import APScheduler
15
```
16
17
For authentication:
18
19
```python
20
from flask_apscheduler.auth import HTTPBasicAuth, Authorization
21
```
22
23
For scheduler states:
24
25
```python
26
from flask_apscheduler import STATE_RUNNING, STATE_PAUSED, STATE_STOPPED
27
```
28
29
For utility functions:
30
31
```python
32
from flask_apscheduler.utils import job_to_dict, trigger_to_dict, fix_job_def, bytes_to_wsgi, wsgi_to_bytes
33
```
34
35
For authentication utilities:
36
37
```python
38
from flask_apscheduler.auth import get_authorization_header
39
```
40
41
For JSON serialization:
42
43
```python
44
from flask_apscheduler.json import jsonify
45
```
46
47
## Basic Usage
48
49
```python
50
from flask import Flask
51
from flask_apscheduler import APScheduler
52
53
class Config:
54
SCHEDULER_API_ENABLED = True
55
56
app = Flask(__name__)
57
app.config.from_object(Config())
58
59
scheduler = APScheduler()
60
scheduler.init_app(app)
61
scheduler.start()
62
63
# Add a job using decorator
64
@scheduler.task("interval", id="job1", seconds=30)
65
def scheduled_task():
66
print("This task runs every 30 seconds")
67
68
# Add a job programmatically
69
def my_task():
70
print("Task executed")
71
72
scheduler.add_job(
73
id="job2",
74
func=my_task,
75
trigger="cron",
76
hour=0,
77
minute=0
78
)
79
80
app.run()
81
```
82
83
## Architecture
84
85
Flask-APScheduler is built around these key components:
86
87
- **APScheduler Class**: Main interface that wraps APScheduler functionality and integrates with Flask
88
- **REST API**: Optional HTTP endpoints for runtime job management
89
- **Authentication System**: Pluggable authentication handlers for API security
90
- **Configuration Integration**: Flask config-driven setup for schedulers, job stores, and executors
91
- **Job Management**: Full lifecycle management of scheduled tasks
92
93
## Capabilities
94
95
### Core Scheduler Management
96
97
Primary scheduler functionality including initialization, lifecycle management, job operations, and event handling. Provides the main APScheduler class that integrates with Flask applications.
98
99
```python { .api }
100
class APScheduler:
101
def __init__(self, scheduler=None, app=None): ...
102
def init_app(self, app): ...
103
def start(self, paused=False): ...
104
def shutdown(self, wait=True): ...
105
def pause(self): ...
106
def resume(self): ...
107
```
108
109
[Core Scheduler](./core-scheduler.md)
110
111
### Job Management
112
113
Comprehensive job management capabilities including adding, removing, modifying, and executing scheduled jobs. Supports programmatic job creation and decorator-based job definitions.
114
115
```python { .api }
116
def add_job(self, id, func, **kwargs): ...
117
def remove_job(self, id, jobstore=None): ...
118
def get_job(self, id, jobstore=None): ...
119
def get_jobs(self, jobstore=None): ...
120
def modify_job(self, id, jobstore=None, **changes): ...
121
```
122
123
[Job Management](./job-management.md)
124
125
### REST API Management
126
127
HTTP REST API endpoints for runtime scheduler and job management. Provides programmatic access to all scheduler operations through HTTP requests with JSON responses.
128
129
```python { .api }
130
def get_scheduler_info(): ...
131
def add_job(): ...
132
def get_jobs(): ...
133
def pause_scheduler(): ...
134
def start_scheduler(): ...
135
```
136
137
[REST API](./rest-api.md)
138
139
### Authentication System
140
141
Authentication framework for securing REST API endpoints. Supports HTTP Basic authentication with customizable authentication callbacks.
142
143
```python { .api }
144
class HTTPBasicAuth:
145
def get_authorization(self): ...
146
def get_authenticate_header(self): ...
147
148
class Authorization(dict):
149
def __init__(self, auth_type, **kwargs): ...
150
151
def get_authorization_header(): ...
152
```
153
154
[Authentication](./authentication.md)
155
156
### Utility Functions
157
158
Helper functions for job serialization, trigger manipulation, and data conversion. These utilities support job management and API operations.
159
160
```python { .api }
161
def job_to_dict(job): ...
162
def trigger_to_dict(trigger): ...
163
def fix_job_def(job_def): ...
164
def pop_trigger(data): ...
165
def extract_timedelta(delta): ...
166
def bytes_to_wsgi(data): ...
167
def wsgi_to_bytes(data): ...
168
```
169
170
### JSON Serialization
171
172
Custom JSON serialization with support for datetime objects and APScheduler Job objects.
173
174
```python { .api }
175
def jsonify(data, status=None): ...
176
```
177
178
## Types
179
180
```python { .api }
181
# Scheduler state constants
182
STATE_RUNNING: int
183
STATE_PAUSED: int
184
STATE_STOPPED: int
185
186
# Job trigger types
187
TRIGGER_TYPES = ["date", "interval", "cron"]
188
189
# Utility types
190
from collections import OrderedDict
191
from datetime import timedelta
192
```
193
194
## Configuration Options
195
196
Flask configuration keys for customizing scheduler behavior:
197
198
```python { .api }
199
# Flask configuration keys
200
SCHEDULER_JOBSTORES: dict # Job store configuration
201
SCHEDULER_EXECUTORS: dict # Executor configuration
202
SCHEDULER_JOB_DEFAULTS: dict # Default job options
203
SCHEDULER_TIMEZONE: str # Scheduler timezone
204
SCHEDULER_AUTH: HTTPAuth # Authentication handler
205
SCHEDULER_API_ENABLED: bool # Enable REST API
206
SCHEDULER_API_PREFIX: str # API URL prefix
207
SCHEDULER_ENDPOINT_PREFIX: str # Flask endpoint prefix
208
SCHEDULER_ALLOWED_HOSTS: list # Allowed hostnames
209
SCHEDULER_JOBS: list # Job definitions to load
210
JOBS: list # Alternative job definitions key
211
```
212
213
## Utility Functions
214
215
```python { .api }
216
def job_to_dict(job):
217
"""
218
Converts a job to an OrderedDict.
219
220
Args:
221
job: APScheduler Job object
222
223
Returns:
224
OrderedDict: Job data including id, name, func, args, kwargs, trigger info
225
"""
226
227
def trigger_to_dict(trigger):
228
"""
229
Converts a trigger to an OrderedDict.
230
231
Args:
232
trigger: APScheduler trigger object (DateTrigger, IntervalTrigger, CronTrigger)
233
234
Returns:
235
OrderedDict: Trigger data with type and configuration
236
"""
237
238
def fix_job_def(job_def):
239
"""
240
Replaces datetime strings with datetime objects in job definition.
241
242
Args:
243
job_def (dict): Job definition dictionary (modified in-place)
244
245
Note:
246
Parses 'start_date', 'end_date', 'run_date' strings to datetime objects
247
"""
248
249
def pop_trigger(data):
250
"""
251
Pops trigger and trigger args from a given dict.
252
253
Args:
254
data (dict): Dictionary containing trigger configuration
255
256
Returns:
257
tuple: (trigger_name, trigger_args)
258
259
Raises:
260
Exception: If trigger type is not supported
261
"""
262
263
def extract_timedelta(delta):
264
"""
265
Extract weeks, days, hours, minutes, seconds from timedelta.
266
267
Args:
268
delta (timedelta): Time delta to extract components from
269
270
Returns:
271
tuple: (weeks, days, hours, minutes, seconds)
272
"""
273
274
def bytes_to_wsgi(data):
275
"""
276
Convert bytes to WSGI-compatible string.
277
278
Args:
279
data (bytes): Bytes data to convert
280
281
Returns:
282
str: WSGI-compatible string using latin1 encoding
283
"""
284
285
def wsgi_to_bytes(data):
286
"""
287
Convert WSGI unicode represented bytes to real bytes.
288
289
Args:
290
data (str or bytes): Data to convert to bytes
291
292
Returns:
293
bytes: Converted bytes using latin1 encoding
294
"""
295
296
def get_authorization_header():
297
"""
298
Return request's 'Authorization:' header as a two-tuple of (type, info).
299
300
Returns:
301
tuple or None: (auth_type, auth_info) if present, None otherwise
302
"""
303
304
def jsonify(data, status=None):
305
"""
306
Create JSON response with custom serialization support.
307
308
Args:
309
data: Data to serialize to JSON
310
status (int, optional): HTTP status code
311
312
Returns:
313
Response: Flask response with JSON content
314
315
Note:
316
Supports datetime objects and APScheduler Job objects
317
"""
318
```