Scheduler Module
The Scheduler module provides deployment scheduling capabilities with a provider-agnostic design.
The root-level praisonai.scheduler module is deprecated. Use from praisonai.scheduler import ... instead.
Import
from praisonai.scheduler import ScheduleParser, DeploymentScheduler
Quick Example
from praisonai.scheduler import DeploymentScheduler
# Create scheduler
scheduler = DeploymentScheduler()
# Schedule deployment every hour
scheduler.schedule(interval_minutes=60)
# Start the scheduler
scheduler.start()
Classes
DeploymentScheduler
Minimal deployment scheduler with provider-agnostic design.
Features:
- Simple interval-based scheduling
- Thread-safe operation
- Extensible deployer factory pattern
from praisonai.scheduler import DeploymentScheduler
scheduler = DeploymentScheduler()
scheduler.schedule(interval_minutes=30)
scheduler.start()
# Later, stop the scheduler
scheduler.stop()
ScheduleParser
Parses schedule expressions into scheduling parameters.
from praisonai.scheduler import ScheduleParser
# Parse cron-like expressions
schedule = ScheduleParser.parse("every 30 minutes")
schedule = ScheduleParser.parse("daily at 09:00")
DeployerInterface
Abstract interface for deployers to ensure provider compatibility.
from praisonai.scheduler import DeployerInterface
class MyDeployer(DeployerInterface):
def deploy(self) -> bool:
"""Execute deployment. Returns True on success."""
# Your deployment logic
return True
CloudDeployerAdapter
Adapter for existing CloudDeployer to match interface.
from praisonai.scheduler import CloudDeployerAdapter
adapter = CloudDeployerAdapter()
success = adapter.deploy()
Methods
DeploymentScheduler.schedule(interval_minutes)
Schedule deployments at a fixed interval.
Parameters:
interval_minutes (int): Minutes between deployments
DeploymentScheduler.start()
Start the scheduler in a background thread.
DeploymentScheduler.stop()
Stop the scheduler.
DeploymentScheduler.is_running()
Check if the scheduler is currently running.
Returns: bool
Example: Custom Deployer
from praisonai.scheduler import DeploymentScheduler, DeployerInterface
class DockerDeployer(DeployerInterface):
def __init__(self, image_name: str):
self.image_name = image_name
def deploy(self) -> bool:
import subprocess
try:
subprocess.run(
["docker", "push", self.image_name],
check=True
)
return True
except subprocess.CalledProcessError:
return False
# Use custom deployer
scheduler = DeploymentScheduler(deployer=DockerDeployer("myapp:latest"))
scheduler.schedule(interval_minutes=60)
scheduler.start()