CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-flask-apscheduler

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

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview
Eval results
Files

rest-api.mddocs/

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. API endpoints are automatically registered when SCHEDULER_API_ENABLED is True.

Capabilities

Scheduler Management API

HTTP endpoints for controlling scheduler lifecycle and getting status information.

def get_scheduler_info():
    """
    Gets the scheduler info.
    
    HTTP: GET /scheduler
    
    Returns:
        JSON response with:
            current_host (str): Current hostname
            allowed_hosts (list): List of allowed hostnames  
            running (bool): Whether scheduler is running
    """

def pause_scheduler():
    """
    Pauses job processing in the scheduler.
    
    HTTP: POST /scheduler/pause
    
    Returns:
        HTTP 204: Success
        HTTP 500: Error with {"error_message": str}
    """

def resume_scheduler():
    """
    Resumes job processing in the scheduler.
    
    HTTP: POST /scheduler/resume
    
    Returns:
        HTTP 204: Success
        HTTP 500: Error with {"error_message": str}
    """

def start_scheduler():
    """
    Starts the scheduler.
    
    HTTP: POST /scheduler/start
    
    Returns:
        HTTP 204: Success
        HTTP 400: SchedulerAlreadyRunningError
        HTTP 500: Other errors with {"error_message": str}
    """

def shutdown_scheduler():
    """
    Shuts down the scheduler.
    
    HTTP: POST /scheduler/shutdown
    
    Request Body (JSON, optional):
        wait (bool): Wait for running jobs to finish (default: true)
    
    Returns:
        HTTP 204: Success
        HTTP 400: SchedulerNotRunningError
        HTTP 500: Other errors with {"error_message": str}
    """

Job Management API

HTTP endpoints for managing individual jobs and job collections.

def add_job():
    """
    Adds a new job.
    
    HTTP: POST /scheduler/jobs
    
    Request Body (JSON):
        id (str): Job identifier
        func (str): Function reference (e.g., "module:function")
        trigger (str): Trigger type ("date", "interval", "cron")
        [trigger parameters]: Based on trigger type
        [job options]: name, misfire_grace_time, max_instances, etc.
    
    Returns:
        HTTP 200: Job object as JSON
        HTTP 409: ConflictingIdError if job ID already exists
        HTTP 500: Other errors with {"error_message": str}
    """

def get_jobs():
    """
    Gets all scheduled jobs.
    
    HTTP: GET /scheduler/jobs
    
    Returns:
        JSON array of job objects with:
            id, name, func, args, kwargs, trigger info, next_run_time, etc.
    """

def get_job(job_id):
    """
    Gets a specific job.
    
    HTTP: GET /scheduler/jobs/{job_id}
    
    Args:
        job_id (str): Job identifier from URL path
        
    Returns:
        HTTP 200: Job object as JSON
        HTTP 404: Job not found with {"error_message": str}
    """

def delete_job(job_id):
    """
    Deletes a job.
    
    HTTP: DELETE /scheduler/jobs/{job_id}
    
    Args:
        job_id (str): Job identifier from URL path
        
    Returns:
        HTTP 204: Success
        HTTP 404: Job not found with {"error_message": str}
        HTTP 500: Other errors with {"error_message": str}
    """

def update_job(job_id):
    """
    Updates a job.
    
    HTTP: PATCH /scheduler/jobs/{job_id}
    
    Args:
        job_id (str): Job identifier from URL path
        
    Request Body (JSON):
        [job properties]: Any job properties to modify
        
    Returns:
        HTTP 200: Updated job object as JSON
        HTTP 404: Job not found with {"error_message": str}
        HTTP 500: Other errors with {"error_message": str}
    """

Job Control API

HTTP endpoints for controlling individual job execution.

def pause_job(job_id):
    """
    Pauses a job.
    
    HTTP: POST /scheduler/jobs/{job_id}/pause
    
    Args:
        job_id (str): Job identifier from URL path
        
    Returns:
        HTTP 200: Updated job object as JSON
        HTTP 404: Job not found with {"error_message": str}
        HTTP 500: Other errors with {"error_message": str}
    """

def resume_job(job_id):
    """
    Resumes a job.
    
    HTTP: POST /scheduler/jobs/{job_id}/resume
    
    Args:
        job_id (str): Job identifier from URL path
        
    Returns:
        HTTP 200: Updated job object as JSON
        HTTP 404: Job not found with {"error_message": str}
        HTTP 500: Other errors with {"error_message": str}
    """

def run_job(job_id):
    """
    Executes a job immediately.
    
    HTTP: POST /scheduler/jobs/{job_id}/run
    
    Args:
        job_id (str): Job identifier from URL path
        
    Returns:
        HTTP 200: Job object as JSON
        HTTP 404: Job not found with {"error_message": str}
        HTTP 500: Other errors with {"error_message": str}
    """

API Configuration

# Flask configuration for API
SCHEDULER_API_ENABLED: bool = False      # Enable REST API
SCHEDULER_API_PREFIX: str = "/scheduler" # API URL prefix  
SCHEDULER_ENDPOINT_PREFIX: str = "scheduler." # Flask endpoint prefix

Usage Examples

Enable REST API

from flask import Flask
from flask_apscheduler import APScheduler

class Config:
    SCHEDULER_API_ENABLED = True
    SCHEDULER_API_PREFIX = "/api/scheduler"

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

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

Scheduler Management Requests

# Get scheduler info
curl http://localhost:5000/scheduler

# Start scheduler
curl -X POST http://localhost:5000/scheduler/start

# Pause scheduler
curl -X POST http://localhost:5000/scheduler/pause

# Resume scheduler  
curl -X POST http://localhost:5000/scheduler/resume

# Shutdown scheduler (wait for jobs)
curl -X POST http://localhost:5000/scheduler/shutdown \
  -H "Content-Type: application/json" \
  -d '{"wait": true}'

Job Management Requests

# Add a new job
curl -X POST http://localhost:5000/scheduler/jobs \
  -H "Content-Type: application/json" \
  -d '{
    "id": "test_job",
    "func": "mymodule:my_function", 
    "trigger": "interval",
    "seconds": 30,
    "name": "Test Job"
  }'

# Get all jobs
curl http://localhost:5000/scheduler/jobs

# Get specific job
curl http://localhost:5000/scheduler/jobs/test_job

# Update job (change interval)
curl -X PATCH http://localhost:5000/scheduler/jobs/test_job \
  -H "Content-Type: application/json" \
  -d '{"seconds": 60}'

# Delete job
curl -X DELETE http://localhost:5000/scheduler/jobs/test_job

Job Control Requests

# Pause job
curl -X POST http://localhost:5000/scheduler/jobs/test_job/pause

# Resume job
curl -X POST http://localhost:5000/scheduler/jobs/test_job/resume

# Run job immediately
curl -X POST http://localhost:5000/scheduler/jobs/test_job/run

Advanced Job Creation

# Cron job
curl -X POST http://localhost:5000/scheduler/jobs \
  -H "Content-Type: application/json" \
  -d '{
    "id": "daily_task",
    "func": "tasks:daily_cleanup",
    "trigger": "cron",
    "hour": 2,
    "minute": 0,
    "name": "Daily Cleanup Task",
    "misfire_grace_time": 300
  }'

# Date job (one-time execution)
curl -X POST http://localhost:5000/scheduler/jobs \
  -H "Content-Type: application/json" \
  -d '{
    "id": "delayed_task",
    "func": "tasks:process_data",
    "trigger": "date", 
    "run_date": "2024-12-25T10:00:00",
    "args": ["data.json"],
    "kwargs": {"format": "json"}
  }'

Response Formats

Job Object Structure

{
  "id": "job_id",
  "name": "Job Name", 
  "func": "module:function",
  "args": [1, 2, 3],
  "kwargs": {"key": "value"},
  "trigger": "interval",
  "seconds": 30,
  "next_run_time": "2024-01-15T10:30:00.123456",
  "misfire_grace_time": 60,
  "max_instances": 1
}

Scheduler Info Structure

{
  "current_host": "hostname.local",
  "allowed_hosts": ["*"],
  "running": true
}

Error Response Structure

{
  "error_message": "Detailed error description"
}

Install with Tessl CLI

npx tessl i tessl/pypi-flask-apscheduler

docs

authentication.md

core-scheduler.md

index.md

job-management.md

rest-api.md

tile.json