Skip to main content

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()