This Python script provides functionality to schedule and execute scripts based on cron expressions. It allows executing both bash and Python scripts and provides options to add script parameters and skip conditions.
Project description
Orchestrator
This Python script provides functionality to schedule and execute scripts based on cron expressions. It allows executing both bash and Python scripts and provides options to add script parameters and skip conditions.
Dependencies
croniter
: A Python library to parse cron expressions and calculate the next execution time.
Usage
- Import the required modules:
import logging
import subprocess
import signal
import pytz
import time
import os
from croniter import croniter
from datetime import datetime, timedelta
- Configure logging settings:
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
- Create a
Task
object to define a script execution task:
task = Task(cron_expression, script_type, interval=1)
cron_expression
(str): A cron expression specifying the schedule of the task.script_type
(str): The type of script to execute, either 'bash' or 'python'.interval
(int): The interval in seconds between task executions (default is 1 second).
- Add scripts to the task using the
add_script
method:
task.add_script(script_path, script_parameters=None)
script_path
(str): The path to the script file.script_parameters
(dict): Optional dictionary of script parameters.
- Add skip conditions to the task using the
add_skip
method:
task.add_skip(skip_condition)
skip_condition
(bool): A condition that determines whether the task should be skipped.
- Execute a script using the
execute_script
method:
task.execute_script(script_path)
script_path
(str): The path to the script file.
- Repeat script execution with different parameters using the
repeat
method:
task.repeat(script_path, parameters_list)
script_path
(str): The path to the script file.parameters_list
(list): A list of dictionaries containing different sets of script parameters.
- Check if the task should be executed using the
should_execute
method:
if task.should_execute():
# Execute the task
- Create an
Orchestrator
object to manage multiple tasks:
orchestrator = Orchestrator()
- Add tasks to the orchestrator using the
add_task
method:
task = orchestrator.add_task(cron_expression, script_type)
cron_expression
(str): A cron expression specifying the schedule of the task.script_type
(str): The type of script to execute, either 'bash' or 'python'.
- Execute all tasks using the
execute_tasks
method:
orchestrator.execute_tasks()
Note: The script will run indefinitely, executing tasks based on their schedules and intervals.
Example
Here's an example usage of the script:
# Create a Task
task = Task("*/5 * * * *", "bash", interval=5)
# Add scripts to the task
task.add_script("script.sh")
task.add_script("another_script.sh", {"param1": "value1", "param2": "value2"})
# Add skip conditions
task.add_skip(False) # Do not skip this task
# Create a Orchestrator
orchestrator = Orchestrator()
# Add the task to the orchestrator
orchestrator.add_task("0 0 * * *", "python")
# Execute all tasks
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
orchestration-1.0.3.tar.gz
(3.4 kB
view hashes)
Built Distribution
Close
Hashes for orchestration-1.0.3-py2-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | f6d1df929b8dbe2d9da833c0ed490cd8318fc685dff06d9510726bb339c6ff73 |
|
MD5 | 6de6339eddc1a678e3698d53218666e8 |
|
BLAKE2b-256 | 698df6120acf072f89cbffee166b9a4e524c2fcdd8260c1c179c015e6f57d622 |