Automated Task Scheduler

A Python-based task scheduling system with advanced design patterns, task bundling, dependency management, and a Flask web interface.

Features

Core Capabilities

• Singleton Scheduler - Thread-safe, single instance scheduler
• Priority Queue - Time-based task scheduling with automatic ordering
• Command Pattern - Extensible task types with unified interface
• Chain of Responsibility - Pipeline-based task processing
• Observer Pattern - Event-driven architecture with real-time notifications
• Dependency Management - Tasks can depend on other tasks
• Task Bundling - Group multiple tasks into chains with different execution modes

Task Types

Type	Description
Email	Send emails via SMTP (Gmail supported)
Cleanup	Delete files/folders with wildcard patterns
Alarm	Display notifications with optional sound
Note	Write timestamped notes to files
File Copy	Copy files for backup purposes

Execution Modes

• Sequential - Tasks run in order, each depending on the previous
• Parallel - All tasks run simultaneously
• Conditional - Chain stops if the first task fails

Installation

# Clone the repository
git clone https://github.com/yourusername/AutomatedTaskSchedular.git
cd AutomatedTaskSchedular

# Create virtual environment (optional)
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

# Install dependencies
pip install flask python-dotenv

Configuration

Email Settings (Optional)

Create a .env file in the project root:

EMAIL_USER=your-email@gmail.com
EMAIL_PASSWORD=your-app-password

For Gmail, you need to create an App Password:

1. Enable 2-Factor Authentication
2. Go to https://myaccount.google.com/apppasswords
3. Generate a new app password

Usage

Web Interface

python run.py

Open http://localhost:5000 in your browser.

Available Pages:

Route	Description
/tasks	View all tasks with live log
/tasks/new	Create a single task
/bundles/new	Create a task chain (bundle)

Console Mode

# Basic demo
python main.py

# Bundle demo
python quick_test_bundle.py

Programmatic Usage

Single Task

from app.core.scheduler import Scheduler
from app.core.commands import AlarmTask
from app.core.models import ScheduledTask

scheduler = Scheduler.get_instance()
scheduler.start()

# Schedule an alarm 5 seconds from now
alarm = AlarmTask("alarm_1", "Time is up!")
task = ScheduledTask.schedule_after(5, alarm, name="Test Alarm")
scheduler.schedule(task)

Task with Dependency

from app.core.commands import EmailTask, CleanupTask

# First task
cleanup = CleanupTask("cleanup_1", "/tmp/cache", file_pattern="*.tmp")
task1 = ScheduledTask.schedule_after(2, cleanup)
scheduler.schedule(task1)

# Second task depends on first
email = EmailTask("email_1", to="admin@example.com", 
                  subject="Cleanup Done", body="Cache cleared.")
task2 = ScheduledTask.schedule_after(5, email)
task2.dependencies.append(task1.task_id)
scheduler.schedule(task2)

Task Bundle

from app.core.task_bundle import TaskBundle

bundle = TaskBundle(name="Backup Routine")
bundle.add_task(CleanupTask("clean", "/backup/old"), delay=0)
bundle.add_task(FileCopyTask("copy", "/data/db.sqlite", "/backup/db.bak"), delay=2)
bundle.add_task(EmailTask("notify", "admin@example.com", "Backup Done", "Success"), delay=1)

scheduler.schedule_bundle(bundle)

Chain Builder

from app.core.task_bundle import TaskChainBuilder

chain = TaskChainBuilder("deploy-pipeline") \
    .then(AlarmTask("start", "Deployment started"), delay=0) \
    .wait(2) \
    .then(AlarmTask("build", "Building..."), delay=0) \
    .wait(3) \
    .then(AlarmTask("done", "Deployment complete"), delay=0) \
    .build()

scheduler.schedule_bundle(chain)

REST API

Create a bundle via JSON:

curl -X POST http://localhost:5000/bundles/quick \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Quick Bundle",
    "mode": "sequential",
    "start_delay": 5,
    "tasks": [
      {"type": "alarm", "message": "Task 1", "delay": 0},
      {"type": "note", "text": "Task completed", "file": "log.txt", "delay": 2}
    ]
  }'

Project Structure

AutomatedTaskSchedular/
├── app/
│   ├── __init__.py           # Flask app factory
│   ├── core/                  # Business logic
│   │   ├── commands.py        # Task command classes
│   │   ├── command_factory.py # Factory for creating commands
│   │   ├── events.py          # Observer pattern implementation
│   │   ├── handlers.py        # Chain of Responsibility handlers
│   │   ├── models.py          # Data models (ScheduledTask, TaskStatus)
│   │   ├── scheduler.py       # Singleton scheduler
│   │   ├── task_bundle.py     # Bundle and chain builder
│   │   └── task_registry.py   # Task type registry
│   └── web/                   # Flask web interface
│       ├── routes.py          # HTTP endpoints
│       └── templates/         # HTML templates
├── main.py                    # Console demo
├── run.py                     # Flask server
├── demo_bundle.py             # Bundle demonstrations
└── quick_test_bundle.py       # Quick bundle test

Design Patterns

Pattern	Implementation	Purpose
Singleton	Scheduler	Single scheduler instance
Command	ITaskCommand	Encapsulate task operations
Chain of Responsibility	TaskHandler	Pipeline processing
Observer	EventManager	Event notifications
Factory	CommandFactory	Create commands from form data
Registry	TaskRegistry	Register task types
Composite	TaskBundle	Group multiple tasks
Builder	TaskChainBuilder	Fluent task chain creation
Strategy	Execution modes	Different scheduling strategies

Task States

CREATED -> SCHEDULED -> RUNNING -> COMPLETED
                              \-> FAILED
                              \-> CANCELED

API Reference

Scheduler

scheduler = Scheduler.get_instance()
scheduler.start()
scheduler.schedule(task)
scheduler.schedule_bundle(bundle, start_time=None)
scheduler.cancel_task(task_id)
scheduler.list_tasks()
scheduler.stop()

ScheduledTask

task = ScheduledTask(
    scheduled_time=timestamp,
    command=command,
    max_retries=0,
    name="Task Name"
)
task.dependencies.append(other_task_id)

# Helper method
task = ScheduledTask.schedule_after(delay_seconds, command, name="Name")

TaskBundle

bundle = TaskBundle(name="Bundle Name")
bundle.add_task(command, delay=0, name=None, max_retries=0)
bundle.set_execution_mode("sequential")  # or "parallel", "conditional"
bundle.get_tasks()
bundle.get_execution_mode()

TaskChainBuilder

chain = TaskChainBuilder("chain-name") \
    .then(command, delay=0, name=None) \
    .wait(seconds) \
    .parallel(cmd1, cmd2, cmd3) \
    .build()

Web Interface Features

• Task List - View all scheduled and completed tasks
• Live Log - Real-time event stream via Server-Sent Events
• Single Task Form - Create individual tasks with all options
• Bundle Form - Create task chains with individual scheduling
• Task Cancellation - Cancel pending or running tasks

Troubleshooting

Tasks not running

Ensure the scheduler is started:

scheduler = Scheduler.get_instance()
scheduler.start()  # Don't forget this

Email task fails

1. Check .env file exists with correct credentials
2. For Gmail, use an App Password (not your regular password)
3. Ensure 2FA is enabled on your Google account

Module not found error

Install required dependencies:

pip install flask python-dotenv

Tasks not respecting dependencies

Make sure you're using sequential mode for dependent tasks:

bundle.set_execution_mode("sequential")

Known Limitations

• Task data is stored in memory (lost on restart)
• No user authentication
• Retry mechanism defined but not fully implemented
• Bundle cancellation cancels individual tasks only

Future Improvements

• Database persistence (SQLite/PostgreSQL)
• User authentication and authorization
• Cron expression support for recurring tasks
• Bundle templates
• Visual workflow editor
• Comprehensive API documentation
• Unit test coverage
• Docker support

License

MIT License

Contributing

1. Fork the repository
2. Create a feature branch
3. Commit your changes
4. Push to the branch
5. Open a Pull Request