Flask extension that integrates APScheduler for job scheduling with REST API management and authentication support
—
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
Authentication framework for securing REST API endpoints. Provides pluggable authentication handlers with support for HTTP Basic authentication and custom authentication callbacks.
Core authentication interfaces and data structures.
class HTTPAuth:
"""
A base class from which all authentication classes should inherit.
"""
def get_authorization(self):
"""
Get the authorization header.
Returns:
Authorization: The authentication data or None if not present or invalid
Raises:
NotImplemented: Must be implemented by subclasses
"""
def get_authenticate_header(self):
"""
Return the value of WWW-Authenticate header in a 401 Unauthenticated response.
Returns:
str: WWW-Authenticate header value
"""
class Authorization(dict):
"""
A class to hold the authorization data.
Args:
auth_type (str): The authorization type (e.g., "basic", "bearer")
**kwargs: Additional authorization data
Attributes:
auth_type (str): The authorization type
"""
def __init__(self, auth_type, **kwargs):
"""
Initialize Authorization object.
Args:
auth_type (str): Authorization type ("basic", "bearer", etc.)
**kwargs: Authorization-specific data
"""Implementation of HTTP Basic authentication for API endpoints.
class HTTPBasicAuth(HTTPAuth):
"""
HTTP Basic authentication implementation.
Attributes:
www_authenticate_realm (str): Authentication realm for WWW-Authenticate header
"""
www_authenticate_realm: str = "Authentication Required"
def get_authorization(self):
"""
Get the username and password for Basic authentication header.
Returns:
Authorization: Authorization object with username and password, or None
Authorization object contains:
auth_type (str): "basic"
username (str): Decoded username
password (str): Decoded password
"""
def get_authenticate_header(self):
"""
Return the value of WWW-Authenticate header for Basic auth.
Returns:
str: Basic realm="[realm]" header value
"""Utility function for parsing HTTP Authorization headers.
def get_authorization_header():
"""
Return request's 'Authorization:' header as a two-tuple of (type, info).
Returns:
tuple: (auth_type, auth_info) or None if header missing or invalid
Examples:
("basic", b"dXNlcjpwYXNz") # For "Basic dXNlcjpwYXNz"
("bearer", b"token123") # For "Bearer token123"
"""from flask import Flask
from flask_apscheduler import APScheduler
from flask_apscheduler.auth import HTTPBasicAuth
class Config:
SCHEDULER_API_ENABLED = True
SCHEDULER_AUTH = HTTPBasicAuth()
app = Flask(__name__)
app.config.from_object(Config())
scheduler = APScheduler()
scheduler.init_app(app)
# Define authentication callback
@scheduler.authenticate
def authenticate(auth):
"""
Authenticate users for API access.
Args:
auth (Authorization): Authorization object with username/password
Returns:
bool: True if authentication successful, False otherwise
"""
return auth["username"] == "admin" and auth["password"] == "secret"
scheduler.start()from flask_apscheduler.auth import HTTPBasicAuth
# Create auth handler with custom realm
auth_handler = HTTPBasicAuth()
auth_handler.www_authenticate_realm = "Scheduler API"
class Config:
SCHEDULER_API_ENABLED = True
SCHEDULER_AUTH = auth_handler
app.config.from_object(Config())from flask_apscheduler import APScheduler
from flask_apscheduler.auth import HTTPBasicAuth
scheduler = APScheduler()
scheduler.auth = HTTPBasicAuth() # Set directly on scheduler instance
@scheduler.authenticate
def authenticate(auth):
# Check against database, LDAP, etc.
user = get_user_from_database(auth["username"])
return user and user.check_password(auth["password"])import hashlib
import hmac
@scheduler.authenticate
def authenticate(auth):
"""
Advanced authentication with multiple user support and secure password checking.
"""
# Multiple valid users
users = {
"admin": "hashed_admin_password",
"operator": "hashed_operator_password",
"readonly": "hashed_readonly_password"
}
username = auth.get("username")
password = auth.get("password")
if not username or not password:
return False
stored_hash = users.get(username)
if not stored_hash:
return False
# Secure password comparison
expected_hash = hashlib.pbkdf2_hmac('sha256', password.encode('utf-8'), b'salt', 100000)
return hmac.compare_digest(stored_hash.encode(), expected_hash.hex().encode())from flask import request
from flask_apscheduler.auth import HTTPAuth, Authorization
class APIKeyAuth(HTTPAuth):
"""Custom API key authentication."""
def __init__(self, valid_keys):
self.valid_keys = valid_keys
def get_authorization(self):
"""Extract API key from X-API-Key header."""
api_key = request.headers.get('X-API-Key')
if not api_key:
return None
if api_key in self.valid_keys:
return Authorization("apikey", key=api_key)
return None
def get_authenticate_header(self):
"""Return custom auth challenge."""
return 'API-Key realm="Scheduler API"'
# Use custom authentication
api_keys = ["key1", "key2", "key3"]
scheduler.auth = APIKeyAuth(api_keys)
@scheduler.authenticate
def authenticate(auth):
# API key already validated in get_authorization
return auth["key"] in api_keysclass Config:
SCHEDULER_API_ENABLED = True
# SCHEDULER_AUTH not set - no authentication required
# Or explicitly disable
scheduler.auth = None# Using curl with Basic auth
curl -u admin:secret http://localhost:5000/scheduler
# Using Authorization header directly
curl -H "Authorization: Basic YWRtaW46c2VjcmV0" \
http://localhost:5000/scheduler/jobs
# Add job with authentication
curl -u admin:secret \
-X POST http://localhost:5000/scheduler/jobs \
-H "Content-Type: application/json" \
-d '{"id": "test", "func": "tasks:test_task", "trigger": "interval", "seconds": 30}'import requests
from requests.auth import HTTPBasicAuth
# Get scheduler info
response = requests.get(
'http://localhost:5000/scheduler',
auth=HTTPBasicAuth('admin', 'secret')
)
# Add job
job_data = {
"id": "python_job",
"func": "tasks:my_task",
"trigger": "interval",
"seconds": 60
}
response = requests.post(
'http://localhost:5000/scheduler/jobs',
json=job_data,
auth=HTTPBasicAuth('admin', 'secret')
)# Authentication configuration
SCHEDULER_AUTH: HTTPAuth = None # Authentication handler instance
# Built-in options
HTTPBasicAuth.www_authenticate_realm: str = "Authentication Required"HTTP/1.1 401 Unauthorized
WWW-Authenticate: Basic realm="Authentication Required"
Content-Type: text/plain
Access DeniedSame as unauthorized - returns 401 with WWW-Authenticate header to prompt for credentials.
Same response as missing credentials to avoid revealing valid usernames.
Install with Tessl CLI
npx tessl i tessl/pypi-flask-apscheduler