### FastScheduler Quick Start Source: https://github.com/michielme/fastscheduler/blob/main/README.md Basic setup for FastScheduler, defining and starting interval and time-based tasks. ```python from fastscheduler import FastScheduler scheduler = FastScheduler(quiet=True) @scheduler.every(10).seconds def task(): print("Task executed") @scheduler.daily.at("14:30") async def daily_task(): print("Daily task at 2:30 PM") scheduler.start() ``` -------------------------------- ### Complete FastScheduler Example Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/QUICKSTART.md A comprehensive example demonstrating various scheduling methods including interval, daily, cron, weekly, and one-time jobs, along with starting, stopping, and status printing. ```python import asyncio import time from fastscheduler import FastScheduler scheduler = FastScheduler(quiet=True) # Interval job @scheduler.every(10).seconds def heartbeat(): print(f"[{time.time()}] ❤️") # Daily job @scheduler.daily.at("09:00", tz="America/New_York") async def morning(): print("Good morning!") # Cron job @scheduler.cron("*/5 * * * *") def frequent(): print("Every 5 minutes") # Weekly job @scheduler.weekly.monday.at("10:00") def standup(): print("Weekly standup") # One-time job @scheduler.once(60) def later(): print("60 seconds from start") # Start scheduler scheduler.start() try: while True: time.sleep(60) scheduler.print_status() except KeyboardInterrupt: scheduler.stop() ``` -------------------------------- ### Basic Scheduler Setup and Task Definition Source: https://github.com/michielme/fastscheduler/blob/main/README.md Includes setting up the scheduler, defining a recurring background task, and starting the scheduler. Access the dashboard at http://localhost:8000/scheduler/. ```python from fastscheduler import FastScheduler app = ... # Your FastAPI app instance scheduler = FastScheduler() app.include_router(create_scheduler_routes(scheduler)) @scheduler.every(30).seconds def background_task(): print("Background work") scheduler.start() ``` -------------------------------- ### Python Client Example for FastScheduler API Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/endpoints.md Demonstrates how to interact with the FastScheduler API using Python's requests library. Includes examples for getting jobs, pausing jobs, retrieving history, and clearing dead letters. ```python import requests base_url = "http://localhost:8000/scheduler/api" # Get all jobs jobs = requests.get(f"{base_url}/jobs").json() print(f"Active jobs: {len(jobs['jobs'])}") # Pause a job response = requests.post(f"{base_url}/jobs/job_0/pause") print(f"Paused: {response.json()['success']}") # Get history history = requests.get(f"{base_url}/history?limit=10").json() for entry in history['history']: print(f"{entry['func_name']}: {entry['status']}") # Clear dead letters response = requests.delete(f"{base_url}/dead-letters") print(f"Cleared {response.json()['cleared']} dead letters") ``` -------------------------------- ### Install FastScheduler with Database Support Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/configuration.md Installs FastScheduler with 'sqlmodel' for database storage capabilities. ```bash pip install fastscheduler[database] ``` -------------------------------- ### Configure Auto Start Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/configuration.md Sets the scheduler to automatically start upon initialization. The default is False, requiring manual start. ```python scheduler = FastScheduler(auto_start=True) ``` ```python # Equivalent to: scheduler = FastScheduler(auto_start=False) scheduler.start() ``` -------------------------------- ### Install FastScheduler Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/QUICKSTART.md Install the core library or with additional features like cron support, a web dashboard, or database storage. ```bash # Core pip install fastscheduler # With all features pip install fastscheduler[all] # Individual features pip install fastscheduler[cron] # Cron support pip install fastscheduler[fastapi] # Web dashboard pip install fastscheduler[database] # Database storage ``` -------------------------------- ### Basic File Handler Setup Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/configuration.md Configures a file handler for logging scheduler events. ```python handler = logging.FileHandler("scheduler.log") formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s") handler.setFormatter(formatter) logger.addHandler(handler) ``` -------------------------------- ### Install FastScheduler with All Features Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/configuration.md Installs FastScheduler with all available optional dependencies, including cron, database, and FastAPI support. ```bash pip install fastscheduler[all] ``` -------------------------------- ### Basic FastScheduler Usage Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/QUICKSTART.md Demonstrates how to create a scheduler instance, define jobs using decorators, and start the scheduler. ```python from fastscheduler import FastScheduler # Create scheduler scheduler = FastScheduler(quiet=True) # Define jobs @scheduler.every(30).seconds def heartbeat(): print("Every 30 seconds") @scheduler.daily.at("09:00") async def morning_task(): print("9 AM daily") # Start and run scheduler.start() ``` -------------------------------- ### Development Setup Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/configuration.md Configures the scheduler for development with detailed logging, a local state file, and limited workers. ```python import logging logging.basicConfig(level=logging.DEBUG) scheduler = FastScheduler( state_file="dev_scheduler.json", storage="json", quiet=False, # Show all logs max_history=1000, max_workers=2, # Single CPU max_dead_letters=100, ) ``` -------------------------------- ### Install FastScheduler with FastAPI Dashboard Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/configuration.md Installs FastScheduler along with 'fastapi' and 'starlette' for the web dashboard. ```bash pip install fastscheduler[fastapi] ``` -------------------------------- ### Install FastScheduler Source: https://github.com/michielme/fastscheduler/blob/main/README.md Install FastScheduler with optional features like the FastAPI dashboard, cron support, or database capabilities. ```bash pip install fastscheduler ``` ```bash pip install fastscheduler[fastapi] ``` ```bash pip install fastscheduler[cron] ``` ```bash pip install fastscheduler[database] ``` ```bash pip install fastscheduler[all] ``` -------------------------------- ### Install FastScheduler with Optional Features Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/README.md Install the FastScheduler package with optional dependencies for cron support, FastAPI integration, or database storage. Use 'all' for all available features. ```bash pip install fastscheduler pip install fastscheduler[cron] pip install fastscheduler[fastapi] pip install fastscheduler[database] pip install fastscheduler[all] ``` -------------------------------- ### start Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/api-reference/fastscheduler.md Starts the FastScheduler background thread, enabling it to begin executing scheduled jobs. ```APIDOC ## start() ### Description Start the scheduler background thread. ### Method `POST` (conceptual, as this is a Python method) ### Parameters None ### Example ```python scheduler.start() ``` ``` -------------------------------- ### Install FastScheduler with Cron Support Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/configuration.md Installs FastScheduler along with the 'croniter' library for cron-like scheduling. ```bash pip install fastscheduler[cron] ``` -------------------------------- ### JavaScript/Browser Example for FastScheduler API Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/endpoints.md Provides examples for interacting with the FastScheduler API from a browser environment using JavaScript. Includes SSE connection, pausing, resuming, running, and canceling jobs. ```javascript // Connect to SSE stream const eventSource = new EventSource("/scheduler/events"); eventSource.onmessage = (event) => { const data = JSON.parse(event.data); console.log("Running:", data.running); console.log("Jobs:", data.jobs); console.log("Stats:", data.stats); }; eventSource.onerror = () => { console.error("SSE connection error"); }; // Pause a job async function pauseJob(jobId) { const response = await fetch(`/scheduler/api/jobs/${jobId}/pause`, { method: "POST" }); return await response.json(); } // Resume a job async function resumeJob(jobId) { const response = await fetch(`/scheduler/api/jobs/${jobId}/resume`, { method: "POST" }); return await response.json(); } // Run immediately async function runJob(jobId) { const response = await fetch(`/scheduler/api/jobs/${jobId}/run`, { method: "POST" }); return await response.json(); } // Cancel a job async function cancelJob(jobId) { const response = await fetch(`/scheduler/api/jobs/${jobId}/cancel`, { method: "POST" }); return await response.json(); } ``` -------------------------------- ### Install Minimal FastScheduler Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/configuration.md Installs the core FastScheduler package without any optional extras. ```bash pip install fastscheduler ``` -------------------------------- ### Curl Example for GET /scheduler/api/dead-letters Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/endpoints.md Example of how to use curl to fetch dead letter entries with a specified limit. ```bash curl "http://localhost:8000/scheduler/api/dead-letters?limit=50" ``` -------------------------------- ### Get SQLModel Storage Backend Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/api-reference/storage.md Imports and initializes the SQLModelStorageBackend. This backend requires a database URL. ```python from fastscheduler.storage import get_sqlmodel_backend SQLModelStorageBackend = get_sqlmodel_backend() backend = SQLModelStorageBackend( database_url="sqlite:///scheduler.db" ) ``` -------------------------------- ### Implement Custom Storage Backend with Redis Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/configuration.md Example of implementing the StorageBackend interface using Redis for custom persistence. Requires the 'redis' library. ```python from fastscheduler.storage import StorageBackend from typing import Any, Dict, List, Optional class RedisStorageBackend(StorageBackend): """Example: Redis-backed storage""" def __init__(self, redis_url: str): import redis self.redis = redis.from_url(redis_url) def save_state( self, jobs: List[Dict[str, Any]], history: List[Dict[str, Any]], statistics: Dict[str, Any], job_counter: int, scheduler_running: bool, ) -> None: import json self.redis.set("scheduler:state", json.dumps({ "jobs": jobs, "history": history, "statistics": statistics, "job_counter": job_counter, })) def load_state(self) -> Optional[Dict[str, Any]]: import json data = self.redis.get("scheduler:state") return json.loads(data) if data else None def save_dead_letters( self, dead_letters: List[Dict[str, Any]], max_dead_letters: int ) -> None: import json self.redis.set("scheduler:dead_letters", json.dumps(dead_letters)) def load_dead_letters(self) -> List[Dict[str, Any]]: import json data = self.redis.get("scheduler:dead_letters") return json.loads(data) if data else [] def clear_dead_letters(self) -> int: count = len(self.load_dead_letters()) self.redis.delete("scheduler:dead_letters") return count def close(self) -> None: self.redis.close() # Usage scheduler = FastScheduler(storage=RedisStorageBackend("redis://localhost:6379")) ``` -------------------------------- ### FastAPI Integration Setup Source: https://github.com/michielme/fastscheduler/blob/main/README.md Integrate FastScheduler with a FastAPI application to enable the real-time dashboard. ```python from fastapi import FastAPI from fastscheduler import FastScheduler from fastscheduler.fastapi_integration import create_scheduler_routes app = FastAPI() scheduler = FastScheduler(quiet=True) ``` -------------------------------- ### Production Setup (Single Server) Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/configuration.md Optimizes the scheduler for a single production server with reduced logging, increased history, and more workers. ```python scheduler = FastScheduler( state_file="/var/lib/scheduler/state.json", storage="json", quiet=True, # Reduce logs max_history=50000, max_workers=16, # Multicore history_retention_days=30, max_dead_letters=1000, ) ``` -------------------------------- ### FastScheduler Initialization with Custom State File Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/api-reference/storage.md Example of initializing FastScheduler with a custom state file path. Ensure the data directory exists. ```python from fastscheduler import FastScheduler from pathlib import Path # Create data directory data_dir = Path("./data") data_dir.mkdir(exist_ok=True) # Use custom state file scheduler = FastScheduler( state_file=str(data_dir / "scheduler.json") ) scheduler.start() ``` -------------------------------- ### Custom FileSystemBackend Implementation Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/api-reference/storage.md Example of creating a custom storage backend that stores each job's state in a separate JSON file. Requires implementing all abstract methods from StorageBackend. ```python from fastscheduler.storage import StorageBackend from typing import Any, Dict, List, Optional import json from pathlib import Path class FileSystemBackend(StorageBackend): """Store state in separate files per job.""" def __init__(self, directory: str = "./scheduler_state"): self.directory = directory Path(directory).mkdir(exist_ok=True) def save_state( self, jobs: List[Dict[str, Any]], history: List[Dict[str, Any]], statistics: Dict[str, Any], job_counter: int, scheduler_running: bool, ) -> None: # Save jobs separately for job in jobs: path = Path(self.directory) / f"job_{job['job_id']}.json" with open(path, 'w') as f: json.dump(job, f) def load_state(self) -> Optional[Dict[str, Any]]: jobs = [] for file in Path(self.directory).glob("job_*.json"): with open(file) as f: jobs.append(json.load(f)) return {"jobs": jobs, "history": [], "statistics": {}, "_job_counter": 0} # ... implement other required methods ... # Usage scheduler = FastScheduler(storage=FileSystemBackend("./my_scheduler")) ``` -------------------------------- ### Correctly Start the Scheduler Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/ADVANCED.md Ensure that `scheduler.start()` is called after defining jobs. Forgetting to start the scheduler will prevent any defined jobs from running. ```python # ❌ Wrong scheduler = FastScheduler() @scheduler.every(1).hours def task(): pass # Jobs won't run! Missing scheduler.start() # ✅ Correct scheduler = FastScheduler() @scheduler.every(1).hours def task(): pass scheduler.start() ``` -------------------------------- ### Production Setup (Distributed) Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/configuration.md Configures the scheduler for a distributed environment using a shared SQL database, optimized for high throughput. ```python import os scheduler = FastScheduler( storage="sqlmodel", database_url=os.getenv("DATABASE_URL"), # Shared PostgreSQL quiet=True, max_history=100000, max_workers=20, history_retention_days=30, max_dead_letters=2000, ) ``` -------------------------------- ### Memory-Constrained Setup Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/configuration.md Configures the scheduler with minimal history, few workers, and short retention to conserve memory. ```python scheduler = FastScheduler( state_file="scheduler.json", storage="json", quiet=True, max_history=500, # Minimal history max_workers=4, # Few workers history_retention_days=1, # Short retention max_dead_letters=50, # Few dead letters ) ``` -------------------------------- ### SQLModelStorageBackend MySQL Database URL Format Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/api-reference/storage.md Example of a database URL for SQLModelStorageBackend using MySQL. ```python "mysql://user:pass@localhost/mydb" "mysql+pymysql://user:pass@localhost/mydb" ``` -------------------------------- ### SQLModelStorageBackend PostgreSQL Database URL Format Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/api-reference/storage.md Example of a database URL for SQLModelStorageBackend using PostgreSQL. ```python "postgresql://user:pass@localhost/mydb" "postgresql+psycopg2://user:pass@localhost/mydb" ``` -------------------------------- ### Lifecycle Methods Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/INDEX.md Methods to control the start and stop lifecycle of the scheduler. ```APIDOC ## Lifecycle Methods ### Description Methods to control the start and stop lifecycle of the scheduler. ### Methods - **start()** - Purpose: Start the scheduler and begin executing scheduled jobs. - **stop(wait, timeout)** - Purpose: Stop the scheduler. Optionally wait for running jobs to complete within a timeout. ``` -------------------------------- ### PostgreSQL Production Setup Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/api-reference/storage.md Configure FastScheduler to use PostgreSQL for production environments. Ensure the DATABASE_URL environment variable is set. ```python from fastscheduler import FastScheduler import os # PostgreSQL production setup scheduler = FastScheduler( storage="sqlmodel", database_url=os.getenv("DATABASE_URL", "postgresql://localhost/scheduler") ) scheduler.start() ``` -------------------------------- ### Time-based Scheduling Source: https://github.com/michielme/fastscheduler/blob/main/README.md Examples of scheduling tasks at specific times of day or week. ```python @scheduler.daily.at("09:00") # Daily at 9 AM @scheduler.hourly.at(":30") # Every hour at :30 @scheduler.weekly.monday.at("10:00") # Every Monday at 10 AM @scheduler.weekly.weekdays.at("09:00") # Weekdays at 9 AM @scheduler.weekly.weekends.at("12:00") # Weekends at noon ``` -------------------------------- ### Curl Example for DELETE /scheduler/api/dead-letters Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/endpoints.md Example of how to use curl to clear all dead letter entries. ```bash curl -X DELETE http://localhost:8000/scheduler/api/dead-letters ``` -------------------------------- ### Interval-based Scheduling Source: https://github.com/michielme/fastscheduler/blob/main/README.md Examples of scheduling tasks at fixed intervals using seconds, minutes, hours, or days. ```python @scheduler.every(10).seconds @scheduler.every(5).minutes @scheduler.every(2).hours @scheduler.every(1).days ``` -------------------------------- ### Start Scheduler Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/api-reference/fastscheduler.md Initiates the scheduler's background thread, allowing it to begin executing scheduled tasks. ```python scheduler.start() ``` -------------------------------- ### Handling State Persistence Errors Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/errors.md This example ensures the directory for the scheduler's state file exists before initialization. It also shows how to initialize the scheduler with `quiet=False` to observe any logging messages related to file permission errors or corrupted JSON state files. ```python from pathlib import Path # Ensure directory exists state_dir = Path("./scheduler_data") state_dir.mkdir(exist_ok=True) scheduler = FastScheduler( state_file=str(state_dir / "scheduler.json"), quiet=False # See error logs ) # Monitor logs for persistence errors import logging logging.basicConfig(level=logging.ERROR) ``` -------------------------------- ### Configuring FastScheduler Logging Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/QUICKSTART.md Set up logging for FastScheduler to monitor its activity. This example configures a file handler to log scheduler events. ```python import logging # Configure logger logger = logging.getLogger("fastscheduler") logger.setLevel(logging.DEBUG) handler = logging.FileHandler("scheduler.log") logger.addHandler(handler) # Create scheduler scheduler = FastScheduler(quiet=False) ``` -------------------------------- ### SQLModel Backend with PostgreSQL Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/api-reference/storage.md Configure the SQLModel backend for production using PostgreSQL. This setup is suitable for high-volume applications. ```python # Use PostgreSQL for production scheduler = FastScheduler( storage="sqlmodel", database_url="postgresql://host/mydb" ) # Add indexes for queries # (created automatically by SQLModel) ``` -------------------------------- ### Basic FastScheduler Usage with Decorators Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/README.md Demonstrates basic usage of FastScheduler by creating an instance, defining scheduled tasks using decorators for interval and daily execution, and starting the scheduler. Tasks can be synchronous or asynchronous. ```python from fastscheduler import FastScheduler scheduler = FastScheduler() @scheduler.every(30).seconds def task(): print("Every 30 seconds") @scheduler.daily.at("09:00") async def daily_task(): print("Daily at 9 AM") scheduler.start() ``` -------------------------------- ### File Organization Structure Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/README.md Illustrates the directory structure for the FastScheduler project, showing the location of markdown files for documentation, API references, and configuration guides. ```bash /workspace/home/output/ ├── README.md ← This file ├── INDEX.md ← Navigation guide ├── QUICKSTART.md ← 5-min start ├── ADVANCED.md ← Expert patterns ├── types.md ← Data types ├── endpoints.md ← HTTP API ├── configuration.md ← Setup guide ├── errors.md ← Error reference └── api-reference/ ├── fastscheduler.md ← Main class ├── schedulers.md ← Schedule types └── storage.md ← Storage API ``` -------------------------------- ### FastAPI Integration Example Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/INDEX.md Integrate FastScheduler routes into a FastAPI application to enable a web dashboard and API endpoints for managing scheduled jobs. ```python from fastscheduler.fastapi_integration import create_scheduler_routes router = create_scheduler_routes(scheduler, prefix="/scheduler", tags=["scheduler"]) app.include_router(router) ``` -------------------------------- ### Job get_schedule_description() Method Example Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/types.md Retrieves a human-readable description of the job's schedule. This method provides a user-friendly representation of scheduling details. ```python description = job.get_schedule_description() # Returns: "Every 30 seconds" # Returns: "Daily at 09:00" # Returns: "Every Monday at 10:00 (America/New_York)" # Returns: "Cron: 0 9 * * MON-FRI" ``` -------------------------------- ### SQLModelStorageBackend SQLite Database URL Formats Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/api-reference/storage.md Examples of database URLs for SQLModelStorageBackend using SQLite, including file-based and in-memory options. ```python # File-based "sqlite:///scheduler.db" # In-memory (testing) "sqlite:///:memory:" ``` -------------------------------- ### Job to_dict() Method Example Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/types.md Serializes a Job object into a dictionary for persistence. This method is useful for saving job configurations. ```python from fastscheduler import Job job_dict = job.to_dict() print(job_dict["job_id"]) print(job_dict["func_name"]) ``` -------------------------------- ### Job Modifiers Example Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/QUICKSTART.md Illustrates chaining job modifiers like timeout, retries, timezones, and skipping missed runs. ```python @scheduler.every(10).seconds.timeout(60).retries(3) def task(): pass @scheduler.daily.at("09:00").timeout(300) def backup(): pass @scheduler.weekly.monday.at("10:00").tz("America/New_York") def standup(): pass @scheduler.cron("0 * * * *").tz("Europe/London").timeout(600) def hourly_report(): pass @scheduler.every(1).hours.no_catch_up() def skip_missed(): pass ``` -------------------------------- ### JSON Response for GET /scheduler/api/dead-letters Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/endpoints.md Example JSON response structure for the GET /scheduler/api/dead-letters endpoint, showing details of failed jobs. ```json { "dead_letters": [ { "job_id": "job_1", "func_name": "flaky_task", "status": "failed", "timestamp": 1710516545.0, "timestamp_readable": "2024-03-15 14:29:05", "error": "Max retries exceeded: ConnectionError: Connection timed out", "run_count": 5, "retry_count": 3, "execution_time": 30.0 } ], "total": 3 } ``` -------------------------------- ### Catching ImportError for SQLModel Backend Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/errors.md Handles ImportError if SQLModel is not installed when using the SQLModel backend. Install with 'pip install fastscheduler[database]'. ```python from fastscheduler import FastScheduler try: scheduler = FastScheduler( storage="sqlmodel", database_url="sqlite:///scheduler.db" ) except ImportError as e: print(f"Database not available: {e}") # Install: pip install fastscheduler[database] ``` -------------------------------- ### PostgreSQL Database Configuration Source: https://github.com/michielme/fastscheduler/blob/main/README.md Configures FastScheduler to use PostgreSQL for storage, recommended for production environments. Requires `pip install fastscheduler[database]`. ```python scheduler = FastScheduler( storage="sqlmodel", database_url="postgresql://user:password@localhost:5432/mydb" ) ``` -------------------------------- ### Catching ImportError for FastAPI Integration Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/errors.md Handles ImportError if FastAPI is not installed when using FastAPI integration. Install with 'pip install fastscheduler[fastapi]'. ```python try: from fastscheduler.fastapi_integration import create_scheduler_routes except ImportError as e: print(f"FastAPI not available: {e}") # Install: pip install fastscheduler[fastapi] ``` -------------------------------- ### Catching ImportError for Cron Scheduling Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/errors.md Handles ImportError if croniter is not installed when using cron scheduling. Install with 'pip install fastscheduler[cron]'. ```python from fastscheduler import FastScheduler scheduler = FastScheduler() try: @scheduler.cron("0 9 * * MON-FRI") def market_open(): pass except ImportError as e: print(f"Cron not available: {e}") # Install: pip install fastscheduler[cron] ``` -------------------------------- ### Custom Storage Backend Implementation Source: https://github.com/michielme/fastscheduler/blob/main/README.md Demonstrates how to implement a custom storage backend by subclassing `StorageBackend` and providing custom `save_state` and `load_state` methods. ```python from fastscheduler.storage import StorageBackend class MyCustomBackend(StorageBackend): def save_state(self, jobs, history, statistics, job_counter, scheduler_running): # Your implementation ... def load_state(self): # Your implementation ... # Implement other required methods... scheduler = FastScheduler(storage=MyCustomBackend()) ``` -------------------------------- ### JobStatus Usage Example Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/types.md Demonstrates how to check the status of a job using the JobStatus enum. Access the string value of a status using the .value attribute. ```python from fastscheduler import JobStatus if job.status == JobStatus.SCHEDULED: print("Job is waiting") elif job.status == JobStatus.COMPLETED: print("Job finished") # Access string value status_str = JobStatus.COMPLETED.value # "completed" ``` -------------------------------- ### Async File Operations Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/ADVANCED.md Perform asynchronous file read and write operations using aiofiles. This example shows how to read content from a file, process it asynchronously, and write the results back to another file. ```python import asyncio import aiofiles @scheduler.daily.at("02:00") async def batch_process(): async with aiofiles.open("data.txt") as f: content = await f.read() # Process asynchronously results = await process_async(content) async with aiofiles.open("results.txt", "w") as f: await f.write(results) ``` -------------------------------- ### SQLModelStorageBackend Constructor Parameters Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/api-reference/storage.md Illustrates the optional parameters for the SQLModelStorageBackend constructor, including database URL, echo, and quiet settings. ```python backend = SQLModelStorageBackend( database_url="sqlite:///scheduler.db", echo=False, quiet=False ) ``` -------------------------------- ### Abstract load_state Method Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/api-reference/storage.md Abstract method to load the complete scheduler state. This method is called once at scheduler initialization. It should return None for a fresh start and handle missing or corrupted state gracefully. ```python @abstractmethod def load_state(self) -> Optional[Dict[str, Any]] ``` ```python def load_state(self) -> Optional[Dict[str, Any]]: try: # Load and return state... return state_dict except FileNotFoundError: return None # Fresh start except Exception as e: logger.error(f"Load error: {e}") return None # Fallback to fresh start ``` -------------------------------- ### Environment Variable Configuration Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/configuration.md Demonstrates how to configure the scheduler using environment variables for database URL, state file, and quiet mode. ```python import os from fastscheduler import FastScheduler # Database URL from env database_url = os.getenv( "FASTSCHEDULER_DB_URL", "sqlite:///scheduler.db" ) # State file from env state_file = os.getenv( "FASTSCHEDULER_STATE_FILE", "fastscheduler_state.json" ) # Quiet mode from env quiet = os.getenv("FASTSCHEDULER_QUIET", "false").lower() == "true" # Create scheduler scheduler = FastScheduler( state_file=state_file, database_url=database_url, quiet=quiet, ) ``` -------------------------------- ### Scheduler as Context Manager Source: https://github.com/michielme/fastscheduler/blob/main/README.md Manages the scheduler's lifecycle using a context manager. The scheduler starts automatically upon entering the 'with' block and stops automatically upon exiting. ```python with FastScheduler(quiet=True) as scheduler: @scheduler.every(5).seconds def task(): print("Running") # Scheduler starts automatically time.sleep(30) # Scheduler stops automatically on exit ``` -------------------------------- ### Test Scheduler Components with Pytest Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/ADVANCED.md Write unit tests for your scheduler setup using pytest. This includes creating a scheduler fixture and testing job scheduling and execution logic to ensure correctness. ```python import pytest from fastscheduler import FastScheduler @pytest.fixture def scheduler(): return FastScheduler(storage="json", quiet=True) def test_job_scheduling(scheduler): @scheduler.every(1).seconds def test_task(): return "success" jobs = scheduler.get_jobs() assert len(jobs) == 1 assert jobs[0]["func_name"] == "test_task" def test_job_execution(scheduler): executed = [] @scheduler.once(0).seconds def quick_task(): executed.append(True) scheduler.start() time.sleep(1) scheduler.stop() assert executed == [True] ``` -------------------------------- ### Initialize FastScheduler with SQLModel Backend Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/errors.md Demonstrates how to initialize FastScheduler using the SQLModel backend with a PostgreSQL database URL. Includes a try-except block to catch potential database connection errors and fall back to a JSON storage if an exception occurs. ```python from fastscheduler import FastScheduler try: scheduler = FastScheduler( storage="sqlmodel", database_url="postgresql://user:pass@localhost/mydb" ) except Exception as e: print(f"Database error: {e}") # Fallback to JSON scheduler = FastScheduler(storage="json") ``` -------------------------------- ### Get Specific Job Source: https://github.com/michielme/fastscheduler/blob/main/README.md Retrieve details for a specific job by its ID. ```APIDOC ## GET /scheduler/api/jobs/{job_id} ### Description Fetches detailed information about a single job identified by its unique ID. ### Method GET ### Endpoint /scheduler/api/jobs/{job_id} ### Parameters #### Path Parameters - **job_id** (string) - Required - The unique identifier of the job. ``` -------------------------------- ### Basic Scheduling Patterns Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/INDEX.md Demonstrates basic scheduling of tasks using decorators for different time intervals and specific times. ```python @scheduler.every(30).seconds def heartbeat(): pass @scheduler.daily.at("09:00") def daily(): pass @scheduler.cron("0 9 * * MON-FRI") def weekdays(): pass ``` -------------------------------- ### cron(expression) Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/api-reference/fastscheduler.md Schedules a task using cron expressions. Requires the 'croniter' library to be installed. ```APIDOC ## cron(expression) ### Description Schedules a task using cron expressions. Requires the 'croniter' library to be installed. ### Method Not applicable (decorator/method call) ### Endpoint Not applicable ### Parameters #### Path Parameters None #### Query Parameters None #### Request Body None ### Parameters - **expression** (str) - Required - Standard cron expression (minute hour day month weekday) ### Request Example ```python @scheduler.cron("0 9 * * MON-FRI") # 9 AM weekdays def market_open(): print("Market open") @scheduler.cron("*/15 * * * *") # Every 15 minutes def frequent(): print("Every 15 minutes") ``` ### Response #### Success Response (200) Returns a CronScheduler object decorated with the cron schedule. #### Response Example None ### Raises ImportError if croniter not installed ``` -------------------------------- ### FastScheduler Configuration Options Source: https://github.com/michielme/fastscheduler/blob/main/README.md Demonstrates various configuration parameters for FastScheduler, including state file, storage backend, database URL, logging verbosity, auto-start, history limits, worker count, and dead letter queue size. ```python scheduler = FastScheduler( state_file="scheduler.json", # Persistence file for JSON backend (default: fastscheduler_state.json) storage="json", # Storage backend: "json" (default) or "sqlmodel" database_url=None, # Database URL for sqlmodel backend quiet=True, # Suppress log messages (default: False) auto_start=False, # Start immediately (default: False) max_history=5000, # Max history entries to keep (default: 10000) max_workers=20, # Concurrent job threads (default: 10) history_retention_days=8, # Delete history older than X days (default: 7) max_dead_letters=500, # Max failed jobs in dead letter queue (default: 500) ) ``` -------------------------------- ### Get All Scheduled Jobs Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/INDEX.md Retrieve a list of all currently scheduled jobs using the `get_jobs` method. ```python jobs = scheduler.get_jobs() ``` -------------------------------- ### GET /scheduler/api/jobs/{job_id} Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/endpoints.md Retrieves detailed information about a specific job using its unique identifier. ```APIDOC ## GET /scheduler/api/jobs/{job_id} ### Description Get specific job details. ### Method GET ### Endpoint /scheduler/api/jobs/{job_id} ### Parameters #### Path Parameters - **job_id** (str) - Yes - Unique job identifier (e.g., "job_0") ### Response #### Success Response (200) - **job** (object) - Details of the job including ID, function name, status, schedule, next run time, run count, retry count, pause status, timeout, and last run time. ### Response Example ```json { "job": { "job_id": "job_0", "func_name": "my_task", "status": "scheduled", "schedule": "Every 30 seconds", "next_run": "2024-03-15 14:30:45", "next_run_in": 15.5, "run_count": 42, "retry_count": 0, "paused": false, "timeout": 60.0, "last_run": "2024-03-15 14:30:15" } } ``` #### Error Response (200) - **error** (string) - Indicates that the job was not found. - **job_id** (string) - The identifier of the job that was not found. ### Response Example ```json { "error": "Job not found", "job_id": "job_999" } ``` ``` -------------------------------- ### Implement Custom Storage Backend Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/INDEX.md For custom persistence needs, subclass the `StorageBackend` class and implement the required methods to manage job data. ```python Subclass StorageBackend ``` -------------------------------- ### Initialize FastScheduler with Custom Storage Backend Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/api-reference/fastscheduler.md Initializes the scheduler with a custom storage backend instance. Ensure the custom backend class is imported. ```python from fastscheduler.storage import StorageBackend scheduler = FastScheduler(storage=MyCustomBackend()) ``` -------------------------------- ### Get Specific Job Details Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/INDEX.md Fetch details for a specific job by its ID using the `get_job` method. ```python job_details = scheduler.get_job(job_id='some_job_id') ``` -------------------------------- ### JSON Storage Configuration Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/configuration.md Configure FastScheduler to use JSON file storage. This is suitable for single-server setups. ```python scheduler = FastScheduler( state_file="scheduler.json" ) # Equivalent to: scheduler = FastScheduler( storage="json", state_file="scheduler.json" ) ``` ```python from pathlib import Path data_dir = Path("./data") data_dir.mkdir(exist_ok=True) scheduler = FastScheduler( state_file=str(data_dir / "scheduler_state.json") ) scheduler.start() ``` -------------------------------- ### Initialize FastAPI App with Scheduler Routes Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/endpoints.md This snippet demonstrates how to integrate FastScheduler into a FastAPI application. It initializes the FastAPI app and the FastScheduler instance, then includes the scheduler's routes using `create_scheduler_routes`. ```python from fastapi import FastAPI from fastscheduler import FastScheduler from fastscheduler.fastapi_integration import create_scheduler_routes app = FastAPI() scheduler = FastScheduler() # Include all scheduler routes app.include_router(create_scheduler_routes(scheduler)) ``` -------------------------------- ### Configure Database Storage Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/INDEX.md Enable persistent storage for job schedules and history by passing `storage="sqlmodel"` and providing a `database_url` during scheduler initialization. ```python storage="sqlmodel" database_url="..." ``` -------------------------------- ### Scheduler Lifecycle Control Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/QUICKSTART.md Control the scheduler's operational state by starting, stopping, and printing its current status. ```python scheduler.start() # Start scheduler.stop() # Stop scheduler.print_status() # Print status ``` -------------------------------- ### Configure Scheduler with Environment Variables Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/ADVANCED.md Initialize the FastScheduler using environment variables for flexible configuration. This allows you to manage settings like state file, storage backend, database URL, quiet mode, and worker count without modifying the code. ```python import os scheduler = FastScheduler( state_file=os.getenv("SCHEDULER_STATE_FILE", "scheduler.json"), storage=os.getenv("SCHEDULER_STORAGE", "json"), database_url=os.getenv("DATABASE_URL"), quiet=os.getenv("SCHEDULER_QUIET", "false").lower() == "true", max_workers=int(os.getenv("SCHEDULER_WORKERS", "10")), ) ``` -------------------------------- ### FastScheduler JSON State File Structure Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/api-reference/storage.md Example structure of the fastscheduler_state.json file used for storing scheduler state. ```json { "version": "1.0", "metadata": { "last_save": 1710516645.123, "last_save_readable": "2024-03-15 14:30:45", "scheduler_running": true }, "jobs": [ { "job_id": "job_0", "func_name": "my_task", "next_run": 1710516645.0, "interval": 30.0, ... } ], "history": [ { "job_id": "job_0", "func_name": "my_task", "status": "completed", "timestamp": 1710516645.123, ... } ], "statistics": { "total_runs": 1250, "total_failures": 3, "total_retries": 2, "start_time": 1710512000.0 }, "_job_counter": 5 } ``` -------------------------------- ### FastScheduler Constructor Options Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/INDEX.md Initialize the FastScheduler with various configuration parameters to control state file, storage, auto-start behavior, logging verbosity, worker limits, and retention policies. ```python FastScheduler( state_file="fastscheduler_state.json", storage=None, database_url=None, auto_start=False, quiet=False, max_history=10000, max_workers=10, history_retention_days=7, max_dead_letters=500, ) ``` -------------------------------- ### Get Execution History Source: https://github.com/michielme/fastscheduler/blob/main/README.md Retrieves the execution history of jobs. You can filter by function name and limit the number of results. ```python # Get execution history history = scheduler.get_history(limit=100) history = scheduler.get_history(func_name="my_task", limit=50) ``` -------------------------------- ### Get Specific Job Source: https://github.com/michielme/fastscheduler/blob/main/README.md Retrieves a specific job by its unique identifier. Use this to inspect or manage individual jobs. ```python # Get specific job job = scheduler.get_job("job_0") ``` -------------------------------- ### Get Scheduler Statistics Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/INDEX.md Obtain runtime statistics for the scheduler, such as active jobs and execution times, using `get_statistics`. ```python stats = scheduler.get_statistics() ``` -------------------------------- ### Lazy Loading Jobs from Configuration Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/ADVANCED.md Implement dynamic loading of job configurations from a JSON file at application startup. This approach is suitable for applications with a large number of jobs, allowing for flexible management of schedules and job parameters. ```python def load_jobs_from_config(): import json with open("jobs_config.json") as f: config = json.load(f) for job_config in config["jobs"]: schedule_type = job_config["type"] func_name = job_config["function"] if schedule_type == "interval": sched = scheduler.every(job_config["interval"]) elif schedule_type == "daily": sched = scheduler.daily.at(job_config["time"]) elif schedule_type == "cron": sched = scheduler.cron(job_config["expression"]) # Set modifiers if "timeout" in job_config: sched = sched.timeout(job_config["timeout"]) if "retries" in job_config: sched = sched.retries(job_config["retries"]) # Register function func = import_function(func_name) sched(func) # Load once at startup load_jobs_from_config() ``` -------------------------------- ### GET /scheduler/api/history Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/endpoints.md Retrieves the execution history of jobs, with optional filtering by function name and a limit on the number of entries. ```APIDOC ## GET /scheduler/api/history ### Description Get job execution history. ### Method GET ### Endpoint /scheduler/api/history ### Parameters #### Query Parameters - **func_name** (str) - No - Filter by function name - **limit** (int) - No - Maximum entries to return (default: 50) ### Response #### Success Response (200) - **history** (array) - A list of job execution records, each containing job ID, function name, status, timestamp, readable timestamp, error (if any), run count, retry count, and execution time. ### Response Example ```json { "history": [ { "job_id": "job_0", "func_name": "my_task", "status": "completed", "timestamp": 1710516645.123, "timestamp_readable": "2024-03-15 14:30:45", "error": null, "run_count": 42, "retry_count": 0, "execution_time": 1.23 } ] } ``` ``` -------------------------------- ### FastScheduler Constructor Options Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/configuration.md Initialize FastScheduler with various configuration parameters. ```python from fastscheduler import FastScheduler scheduler = FastScheduler( state_file="fastscheduler_state.json", storage=None, database_url=None, auto_start=False, quiet=False, max_history=10000, max_workers=10, history_retention_days=7, max_dead_letters=500, ) ``` -------------------------------- ### FastScheduler JSON Dead Letters File Structure Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/api-reference/storage.md Example structure of the fastscheduler_state_dead_letters.json file for storing failed job entries. ```json { "dead_letters": [ { "job_id": "job_1", "func_name": "flaky_task", "status": "failed", "timestamp": 1710516545.0, "error": "Max retries exceeded: ConnectionError", ... } ], "max_dead_letters": 500 } ``` -------------------------------- ### Run Tests with Coverage Source: https://github.com/michielme/fastscheduler/blob/main/tests/README.md Execute all tests and generate an HTML report for code coverage analysis. ```bash uv run pytest tests/ --cov=fastscheduler --cov-report=html ``` -------------------------------- ### Monitor Job Execution and Status Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/INDEX.md Retrieve logs of all job executions using `scheduler.get_history()` and get summary statistics with `scheduler.get_statistics()`. ```python scheduler.get_history() scheduler.get_statistics() ``` -------------------------------- ### Initialize FastScheduler with SQLModel Database Storage Source: https://github.com/michielme/fastscheduler/blob/main/CHANGELOG.md Use this snippet to initialize FastScheduler with database storage using SQLModel. Specify the storage type and the database connection URL. ```python scheduler = FastScheduler(storage="sqlmodel", database_url="sqlite:///scheduler.db") ``` -------------------------------- ### GET /scheduler/api/dead-letters Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/endpoints.md Retrieves a list of failed job entries from the dead-letter queue. Supports limiting the number of returned entries. ```APIDOC ## GET /scheduler/api/dead-letters ### Description Get failed job entries. ### Method GET ### Endpoint /scheduler/api/dead-letters #### Query Parameters - **limit** (int) - Optional - Maximum entries to return. Defaults to 100. ### Response #### Success Response (200) - **dead_letters** (array) - A list of dead letter entries. - **job_id** (string) - The ID of the job. - **func_name** (string) - The name of the function that failed. - **status** (string) - The status of the job run (e.g., "failed"). - **timestamp** (float) - The Unix epoch timestamp of the event. - **timestamp_readable** (string) - The human-readable local datetime string. - **error** (string) - The error message associated with the failure. - **run_count** (int) - The number of times the job has run. - **retry_count** (int) - The number of times the job has been retried. - **execution_time** (float) - The execution time of the job in seconds. - **total** (int) - The total number of dead letter entries available. ### Request Example ```bash curl "http://localhost:8000/scheduler/api/dead-letters?limit=50" ``` ### Response Example ```json { "dead_letters": [ { "job_id": "job_1", "func_name": "flaky_task", "status": "failed", "timestamp": 1710516545.0, "timestamp_readable": "2024-03-15 14:29:05", "error": "Max retries exceeded: ConnectionError: Connection timed out", "run_count": 5, "retry_count": 3, "execution_time": 30.0 } ], "total": 3 } ``` ``` -------------------------------- ### MySQL Storage Backend Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/QUICKSTART.md Configure the scheduler to use MySQL for state persistence via SQLModel. ```python scheduler = FastScheduler( storage="sqlmodel", database_url="mysql://user:pass@host/db" ) ``` -------------------------------- ### GET /scheduler/api/dead-letters Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/endpoints.md Retrieves failed job entries from the dead-letter queue. You can specify a limit for the maximum number of entries to return. ```http GET /scheduler/api/dead-letters ``` -------------------------------- ### GET /scheduler/api/status Source: https://github.com/michielme/fastscheduler/blob/main/_autodocs/endpoints.md Retrieves the current status and detailed statistics of the scheduler, including uptime, run counts, and job-specific metrics. ```APIDOC ## GET /scheduler/api/status ### Description Get scheduler status and statistics. ### Method GET ### Endpoint /scheduler/api/status ### Response #### Success Response (200) - Content-Type: application/json - Body: JSON object containing scheduler status and statistics ### Response Body Example ```json { "running": true, "statistics": { "total_runs": 1250, "total_failures": 3, "total_retries": 2, "start_time": 1710512000.0, "uptime_seconds": 4800.5, "uptime_readable": "1:20:00", "per_job": { "task_name": { "completed": 100, "failed": 0, "total_runs": 100 } }, "active_jobs": 5 } } ``` ### Request Example ```bash curl http://localhost:8000/scheduler/api/status ``` ```