or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

authentication.mdcore-scheduler.mdindex.mdjob-management.mdrest-api.md
tile.json

tessl/pypi-flask-apscheduler

Flask extension that integrates APScheduler for job scheduling with REST API management and authentication support

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/flask-apscheduler@1.13.x

To install, run

npx @tessl/cli install tessl/pypi-flask-apscheduler@1.13.0

index.mddocs/

Flask-APScheduler

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.

Package Information

  • Package Name: Flask-APScheduler
  • Package Type: pypi
  • Language: Python
  • Installation: pip install Flask-APScheduler

Core Imports

from flask_apscheduler import APScheduler

For authentication:

from flask_apscheduler.auth import HTTPBasicAuth, Authorization

For scheduler states:

from flask_apscheduler import STATE_RUNNING, STATE_PAUSED, STATE_STOPPED

For utility functions:

from flask_apscheduler.utils import job_to_dict, trigger_to_dict, fix_job_def, bytes_to_wsgi, wsgi_to_bytes

For authentication utilities:

from flask_apscheduler.auth import get_authorization_header

For JSON serialization:

from flask_apscheduler.json import jsonify

Basic Usage

from flask import Flask
from flask_apscheduler import APScheduler

class Config:
    SCHEDULER_API_ENABLED = True

app = Flask(__name__)
app.config.from_object(Config())

scheduler = APScheduler()
scheduler.init_app(app)
scheduler.start()

# Add a job using decorator
@scheduler.task("interval", id="job1", seconds=30)
def scheduled_task():
    print("This task runs every 30 seconds")

# Add a job programmatically
def my_task():
    print("Task executed")

scheduler.add_job(
    id="job2",
    func=my_task,
    trigger="cron",
    hour=0,
    minute=0
)

app.run()

Architecture

Flask-APScheduler is built around these key components:

  • APScheduler Class: Main interface that wraps APScheduler functionality and integrates with Flask
  • REST API: Optional HTTP endpoints for runtime job management
  • Authentication System: Pluggable authentication handlers for API security
  • Configuration Integration: Flask config-driven setup for schedulers, job stores, and executors
  • Job Management: Full lifecycle management of scheduled tasks

Capabilities

Core Scheduler Management

Primary scheduler functionality including initialization, lifecycle management, job operations, and event handling. Provides the main APScheduler class that integrates with Flask applications.

class APScheduler:
    def __init__(self, scheduler=None, app=None): ...
    def init_app(self, app): ...
    def start(self, paused=False): ...
    def shutdown(self, wait=True): ...
    def pause(self): ...
    def resume(self): ...

Core Scheduler

Job Management

Comprehensive job management capabilities including adding, removing, modifying, and executing scheduled jobs. Supports programmatic job creation and decorator-based job definitions.

def add_job(self, id, func, **kwargs): ...
def remove_job(self, id, jobstore=None): ...
def get_job(self, id, jobstore=None): ...
def get_jobs(self, jobstore=None): ...
def modify_job(self, id, jobstore=None, **changes): ...

Job Management

REST API Management

HTTP REST API endpoints for runtime scheduler and job management. Provides programmatic access to all scheduler operations through HTTP requests with JSON responses.

def get_scheduler_info(): ...
def add_job(): ...
def get_jobs(): ...
def pause_scheduler(): ...
def start_scheduler(): ...

REST API

Authentication System

Authentication framework for securing REST API endpoints. Supports HTTP Basic authentication with customizable authentication callbacks.

class HTTPBasicAuth:
    def get_authorization(self): ...
    def get_authenticate_header(self): ...

class Authorization(dict):
    def __init__(self, auth_type, **kwargs): ...

def get_authorization_header(): ...

Authentication

Utility Functions

Helper functions for job serialization, trigger manipulation, and data conversion. These utilities support job management and API operations.

def job_to_dict(job): ...
def trigger_to_dict(trigger): ...
def fix_job_def(job_def): ...
def pop_trigger(data): ...
def extract_timedelta(delta): ...
def bytes_to_wsgi(data): ...
def wsgi_to_bytes(data): ...

JSON Serialization

Custom JSON serialization with support for datetime objects and APScheduler Job objects.

def jsonify(data, status=None): ...

Types

# Scheduler state constants
STATE_RUNNING: int
STATE_PAUSED: int
STATE_STOPPED: int

# Job trigger types
TRIGGER_TYPES = ["date", "interval", "cron"]

# Utility types
from collections import OrderedDict
from datetime import timedelta

Configuration Options

Flask configuration keys for customizing scheduler behavior:

# Flask configuration keys
SCHEDULER_JOBSTORES: dict          # Job store configuration
SCHEDULER_EXECUTORS: dict          # Executor configuration  
SCHEDULER_JOB_DEFAULTS: dict       # Default job options
SCHEDULER_TIMEZONE: str            # Scheduler timezone
SCHEDULER_AUTH: HTTPAuth           # Authentication handler
SCHEDULER_API_ENABLED: bool        # Enable REST API
SCHEDULER_API_PREFIX: str          # API URL prefix
SCHEDULER_ENDPOINT_PREFIX: str     # Flask endpoint prefix
SCHEDULER_ALLOWED_HOSTS: list      # Allowed hostnames
SCHEDULER_JOBS: list               # Job definitions to load
JOBS: list                         # Alternative job definitions key

Utility Functions

def job_to_dict(job):
    """
    Converts a job to an OrderedDict.
    
    Args:
        job: APScheduler Job object
        
    Returns:
        OrderedDict: Job data including id, name, func, args, kwargs, trigger info
    """

def trigger_to_dict(trigger):
    """
    Converts a trigger to an OrderedDict.
    
    Args:
        trigger: APScheduler trigger object (DateTrigger, IntervalTrigger, CronTrigger)
        
    Returns:
        OrderedDict: Trigger data with type and configuration
    """

def fix_job_def(job_def):
    """
    Replaces datetime strings with datetime objects in job definition.
    
    Args:
        job_def (dict): Job definition dictionary (modified in-place)
        
    Note:
        Parses 'start_date', 'end_date', 'run_date' strings to datetime objects
    """

def pop_trigger(data):
    """
    Pops trigger and trigger args from a given dict.
    
    Args:
        data (dict): Dictionary containing trigger configuration
        
    Returns:
        tuple: (trigger_name, trigger_args)
        
    Raises:
        Exception: If trigger type is not supported
    """

def extract_timedelta(delta):
    """
    Extract weeks, days, hours, minutes, seconds from timedelta.
    
    Args:
        delta (timedelta): Time delta to extract components from
        
    Returns:
        tuple: (weeks, days, hours, minutes, seconds)
    """

def bytes_to_wsgi(data):
    """
    Convert bytes to WSGI-compatible string.
    
    Args:
        data (bytes): Bytes data to convert
        
    Returns:
        str: WSGI-compatible string using latin1 encoding
    """

def wsgi_to_bytes(data):
    """
    Convert WSGI unicode represented bytes to real bytes.
    
    Args:
        data (str or bytes): Data to convert to bytes
        
    Returns:
        bytes: Converted bytes using latin1 encoding
    """

def get_authorization_header():
    """
    Return request's 'Authorization:' header as a two-tuple of (type, info).
    
    Returns:
        tuple or None: (auth_type, auth_info) if present, None otherwise
    """

def jsonify(data, status=None):
    """
    Create JSON response with custom serialization support.
    
    Args:
        data: Data to serialize to JSON
        status (int, optional): HTTP status code
        
    Returns:
        Response: Flask response with JSON content
        
    Note:
        Supports datetime objects and APScheduler Job objects
    """