55 - Simple Scheduling with Python

# File: de-onboarding/interval_schedule.py
from apscheduler.schedulers.blocking import BlockingScheduler
from typing import NoReturn
from datetime import datetime

def print_time() -> None:
    """Print current time."""
    print(f"Task executed at: {datetime.now()}")  # Log time

def run_scheduler() -> NoReturn:
    """Run interval-based scheduler."""
    scheduler = BlockingScheduler()  # Create scheduler
    scheduler.add_job(
        print_time,  # Task to run
        "interval",  # Trigger type
        seconds=5,  # Run every 5 seconds
        id="time_task"  # Name job for identification
        # Example with parameters: add_job(print_message, "interval", seconds=5, args=["Hello"], id="message_task")
    )
    print("Starting scheduler...")  # Log start
    print(f"Scheduled {len(scheduler.get_jobs())} job(s)")  # Log job count
    scheduler.start()  # Start scheduler (blocks execution)
    print("Scheduler stopped")  # Log shutdown (unreachable due to blocking)

if __name__ == "__main__":
    run_scheduler()

# Expected Output (runs indefinitely):
# Starting scheduler...
# Scheduled 1 job(s)
# Task executed at: 2025-04-25 10:00:00.123456
# Task executed at: 2025-04-25 10:00:05.123456
# (continues every 5 seconds)

# File: de-onboarding/cron_schedule.py
from apscheduler.schedulers.blocking import BlockingScheduler
from typing import NoReturn
from datetime import datetime

def print_time() -> None:
    """Print current time."""
    print(f"Task executed at: {datetime.now()}")  # Log time

def run_scheduler() -> NoReturn:
    """Run cron-based scheduler."""
    scheduler = BlockingScheduler()  # Create scheduler
    scheduler.add_job(
        print_time,  # Task to run
        "cron",  # Trigger type
        second="*/10",  # Run every 10 seconds
        id="cron_time_task"  # Name job
        # Example for daily at 8 AM: hour=8, minute=0 (per APScheduler docs, runs daily at 8:00 AM)
    )
    print("Starting scheduler...")  # Log start
    print(f"Scheduled {len(scheduler.get_jobs())} job(s)")  # Log job count
    scheduler.start()  # Start scheduler
    print("Scheduler stopped")  # Log shutdown (unreachable due to blocking)

if __name__ == "__main__":
    run_scheduler()

# Expected Output (runs indefinitely):
# Starting scheduler...
# Scheduled 1 job(s)
# Task executed at: 2025-04 aktywności:25 10:00:00.123456
# Task executed at: 2025-04-25 10:00:10.123456
# (continues every 10 seconds)

min_price: 10.0
max_quantity: 100
required_fields:
  - product
  - price
  - quantity
product_prefix: 'Halal'
max_decimals: 2

# File: de-onboarding/utils.py (updated from Chapter 54)
from typing import Any, Dict, Union

def is_numeric(s: str, max_decimals: int = 2) -> bool:
    """Check if string is a decimal number with up to max_decimals."""
    parts = s.split(".")
    if len(parts) != 2 or not parts[0].replace("-", "").isdigit() or not parts[1].isdigit():
        return False
    return len(parts[1]) <= max_decimals

def clean_string(s: Union[str, Any]) -> str:
    """Strip whitespace from string."""
    return str(s).strip()

def is_numeric_value(x: Any) -> bool:
    """Check if value is numeric."""
    return isinstance(x, (int, float))

def has_valid_decimals(x: Any, max_decimals: int) -> bool:
    """Check if value has valid decimal places."""
    return is_numeric(str(x), max_decimals)

def apply_valid_decimals(x: Any, max_decimals: int) -> bool:
    """Apply decimal validation."""
    return has_valid_decimals(x, max_decimals)

def is_integer(x: Any) -> bool:
    """Check if value is an integer."""
    return str(x).isdigit()

def validate_sale(sale: Dict[str, str], config: Dict[str, Any]) -> bool:
    """Validate sale based on config rules."""
    required_fields = config["required_fields"]
    min_price = config["min_price"]
    max_quantity = config["max_quantity"]
    prefix = config["product_prefix"]
    max_decimals = config["max_decimals"]

    print(f"Validating sale: {sale}")
    for field in required_fields:
        if not sale[field] or sale[field].strip() == "":
            print(f"Invalid sale: missing {field}: {sale}")
            return False

    product = clean_string(sale["product"])
    if not product.startswith(prefix):
        print(f"Invalid sale: product lacks '{prefix}' prefix: {sale}")
        return False

    price = clean_string(sale["price"])
    if not is_numeric(price, max_decimals) or float(price) < min_price or float(price) <= 0:
        print(f"Invalid sale: invalid price: {sale}")
        return False

    quantity = clean_string(sale["quantity"])
    if not quantity.isdigit() or int(quantity) > max_quantity:
        print(f"Invalid sale: invalid quantity: {sale}")
        return False

    return True

# File: de-onboarding/etl_processor.py
from apscheduler.schedulers.blocking import BlockingScheduler
from typing import Dict, Tuple, NoReturn
import pandas as pd
import yaml
import json
import os
import utils

def read_config(config_path: str) -> Dict[str, Any]:
    """Read YAML configuration (uses PyYAML from Chapter 2)."""
    print(f"Opening config: {config_path}")
    with open(config_path, "r") as file:
        config = yaml.safe_load(file)
    print(f"Loaded config: {config}")
    return config

def load_and_validate_sales(csv_path: str, config: Dict[str, Any]) -> Tuple[pd.DataFrame, int, int]:
    """Load and validate sales CSV (uses Pandas from Chapter 3)."""
    print(f"Loading CSV: {csv_path}")
    df = pd.read_csv(csv_path)  # Load CSV into DataFrame
    print("Initial DataFrame:")
    print(df.head())

    required_fields = config["required_fields"]
    missing_fields = [f for f in required_fields if f not in df.columns]
    if missing_fields:
        print(f"Missing columns: {missing_fields}")
        return pd.DataFrame(), 0, len(df)

    df = df.dropna(subset=["product"])
    df = df[df["product"].str.startswith(config["product_prefix"])]
    df = df[df["quantity"].apply(utils.is_integer)]
    df["quantity"] = df["quantity"].astype(int)
    df = df[df["quantity"] <= config["max_quantity"]]
    df = df[df["price"].apply(utils.is_numeric_value)]
    df = df[df["price"] > 0]
    df = df[df["price"] >= config["min_price"]]
    df = df[df["price"].apply(lambda x: utils.apply_valid_decimals(x, config["max_decimals"]))]

    total_records = len(df)
    print("Validated DataFrame:")
    print(df)
    return df, len(df), total_records

def process_sales(df: pd.DataFrame, config: Dict[str, Any]) -> Tuple[Dict[str, Any], int]:
    """Process sales data (uses Pandas from Chapter 3)."""
    if df.empty:
        print("No valid sales data")
        return {"total_sales": 0.0, "unique_products": [], "top_products": {}}, 0

    df["amount"] = df["price"] * df["quantity"]
    print("DataFrame with Amount:")
    print(df)

    total_sales = df["amount"].sum()
    unique_products = df["product"].unique().tolist()
    sales_by_product = df.groupby("product")["amount"].sum()
    top_products = sales_by_product.sort_values(ascending=False).head(3).to_dict()

    valid_sales = len(df)
    print(f"Valid sales: {valid_sales} records")

    return {
        "total_sales": float(total_sales),
        "unique_products": unique_products,
        "top_products": top_products
    }, valid_sales

def export_results(results: Dict[str, Any], json_path: str) -> None:
    """Export results to JSON (uses file handling from Chapter 2)."""
    print(f"Writing to: {json_path}")
    print(f"Results: {results}")
    with open(json_path, "w") as file:
        json.dump(results, file, indent=2)
    print(f"Exported results to {json_path}")

def etl_task(csv_path: str, config_path: str, json_path: str) -> None:
    """ETL task to process sales data."""
    print(f"Running ETL task at: {pd.Timestamp.now()}")
    config = read_config(config_path)
    df, valid_sales, total_records = load_and_validate_sales(csv_path, config)
    results, valid_sales = process_sales(df, config)
    export_results(results, json_path)

    print("\nSales Report:")
    print(f"Total Records Processed: {total_records}")
    print(f"Valid Sales: {valid_sales}")
    print(f"Invalid Sales: {total_records - valid_sales}")
    print(f"Total Sales: ${round(results['total_sales'], 2)}")
    print(f"Unique Products: {results['unique_products']}")
    print(f"Top Products: {results['top_products']}")

def run_scheduler(csv_path: str, config_path: str, json_path: str) -> NoReturn:
    """Run scheduler for ETL task."""
    scheduler = BlockingScheduler()
    scheduler.add_job(
        etl_task,  # Task to run
        "interval",  # Trigger type
        seconds=5,  # Run every 5 seconds
        args=[csv_path, config_path, json_path],  # Pass arguments
        id="etl_task"  # Name job
    )
    print("Starting scheduler...")
    print(f"Scheduled {len(scheduler.get_jobs())} job(s)")  # Log job count
    scheduler.start()
    print("Scheduler stopped")  # Log shutdown (unreachable due to blocking)

def main() -> None:
    """Main function."""
    csv_path = "data/sales.csv"
    config_path = "data/config.yaml"
    json_path = "data/sales_report.json"
    run_scheduler(csv_path, config_path, json_path)

if __name__ == "__main__":
    main()

# File: de-onboarding/test_etl_processor.py
import pytest
import pandas as pd
import os
import json
import time
from typing import Dict, Any
from apscheduler.schedulers.blocking import BlockingScheduler
from etl_processor import read_config, load_and_validate_sales, process_sales, export_results, etl_task, run_scheduler

@pytest.fixture
def config() -> Dict[str, Any]:
    """Fixture for config."""
    return read_config("data/config.yaml")

def test_load_and_validate_sales(config: Dict[str, Any]) -> None:
    """Test loading and validating sales data."""
    df, valid_sales, total_records = load_and_validate_sales("data/sales.csv", config)
    assert valid_sales == 3
    assert total_records == 3
    assert len(df) == 3
    assert set(df["product"]) == {"Halal Laptop", "Halal Mouse", "Halal Keyboard"}

def test_load_empty_csv(config: Dict[str, Any]) -> None:
    """Test loading empty CSV."""
    df, valid_sales, total_records = load_and_validate_sales("data/empty.csv", config)
    assert df.empty
    assert valid_sales == 0
    assert total_records == 0

def test_process_sales(config: Dict[str, Any]) -> None:
    """Test processing sales data."""
    df, _, _ = load_and_validate_sales("data/sales.csv", config)
    results, valid_sales = process_sales(df, config)
    assert valid_sales == 3
    assert results["total_sales"] == 2499.83
    assert len(results["unique_products"]) == 3
    assert results["top_products"]["Halal Laptop"] == 1999.98

def test_export_results(tmp_path: Any) -> None:
    """Test exporting results."""
    json_path = tmp_path / "test.json"
    results = {
        "total_sales": 2499.83,
        "unique_products": ["Halal Laptop", "Halal Mouse"],
        "top_products": {"Halal Laptop": 1999.98}
    }
    export_results(results, str(json_path))
    assert os.path.exists(json_path)
    with open(json_path, "r") as file:
        saved = json.load(file)
    assert saved == results

def test_etl_task(tmp_path: Any, config: Dict[str, Any]) -> None:
    """Test ETL task."""
    json_path = tmp_path / "sales_report.json"
    etl_task("data/sales.csv", "data/config.yaml", str(json_path))
    assert os.path.exists(json_path)
    with open(json_path, "r") as file:
        results = json.load(file)
    assert results["total_sales"] == 2499.83

def test_invalid_yaml(tmp_path: Any) -> None:
    """Test handling invalid YAML."""
    invalid_yaml = tmp_path / "invalid.yaml"
    with open(invalid_yaml, "w") as file:
        file.write("min_price: 10.0\n  max_quantity: 100")  # Invalid indentation
    df, valid_sales, total_records = load_and_validate_sales("data/sales.csv", read_config(str(invalid_yaml)))
    assert df.empty
    assert valid_sales == 0

def test_partial_data(tmp_path: Any, config: Dict[str, Any]) -> None:
    """Test processing partial valid data (simulates CSV with mixed valid/invalid rows)."""
    partial_csv = tmp_path / "partial.csv"
    with open(partial_csv, "w") as file:
        file.write("product,price,quantity\nHalal Laptop,999.99,2\nMonitor,invalid,2\n")
    df, valid_sales, total_records = load_and_validate_sales(str(partial_csv), config)
    assert valid_sales == 1
    assert len(df) == 1
    assert df["product"].iloc[0] == "Halal Laptop"

def test_scheduler_job_count() -> None:
    """Test scheduler schedules exactly one job."""
    scheduler = BlockingScheduler()
    scheduler.add_job(
        lambda: None,  # Mock task
        "interval",
        seconds=5,
        id="test_task"
    )
    assert len(scheduler.get_jobs()) == 1
    scheduler.shutdown()

def test_scheduler_execution(tmp_path: Any) -> None:
    """Test scheduler executes at least once. Note: time.sleep(2) assumes a 1-second interval; increase to 3 seconds for slower systems."""
    timestamp_file = tmp_path / "timestamp.txt"
    def mock_task() -> None:
        with open(timestamp_file, "w") as f:
            f.write(str(pd.Timestamp.now()))

    scheduler = BlockingScheduler()
    scheduler.add_job(mock_task, "interval", seconds=1, id="mock_task")
    scheduler.start(paused=True)  # Start paused
    scheduler.resume()
    time.sleep(2)  # Wait for execution
    scheduler.shutdown()
    assert os.path.exists(timestamp_file)

{
  "total_sales": 2499.83,
  "unique_products": ["Halal Laptop", "Halal Mouse", "Halal Keyboard"],
  "top_products": {
    "Halal Laptop": 1999.98,
    "Halal Mouse": 249.9,
    "Halal Keyboard": 249.95
  }
}

{
  "total_sales": 2249.88,
  "unique_products": ["Halal Laptop", "Halal Mouse"],
  "top_products": {
    "Halal Laptop": 1999.98,
    "Halal Mouse": 249.9
  }
}

from apscheduler.schedulers.blocking import BlockingScheduler
def print_time():
    print("Task executed")
scheduler = BlockingScheduler()
scheduler.add_job(print_time, "interval", seconds=1)  # Bug: Too frequent
scheduler.start()

from apscheduler.schedulers.blocking import BlockingScheduler
scheduler = BlockingScheduler()
scheduler.add_job(nonexistent_task, "interval", seconds=5)  # Bug: Missing function
scheduler.start()

from apscheduler.schedulers.blocking import BlockingScheduler
from typing import NoReturn
from datetime import datetime

def print_time() -> None:
    """Print current time."""
    print(f"Task executed at: {datetime.now()}")

def run_scheduler() -> NoReturn:
    """Run interval scheduler."""
    scheduler = BlockingScheduler()
    scheduler.add_job(print_time, "interval", seconds=3, id="interval_time_task")
    print("Starting scheduler...")
    print(f"Scheduled {len(scheduler.get_jobs())} job(s)")
    scheduler.start()
    print("Scheduler stopped")

if __name__ == "__main__":
    run_scheduler()

from apscheduler.schedulers.blocking import BlockingScheduler
from typing import NoReturn
from datetime import datetime

def print_time() -> None:
    """Print current time."""
    print(f"Task executed at: {datetime.now()}")

def run_scheduler() -> NoReturn:
    """Run cron scheduler."""
    scheduler = BlockingScheduler()
    scheduler.add_job(print_time, "cron", second="*/10", id="cron_time_task")
    print("Starting scheduler...")
    print(f"Scheduled {len(scheduler.get_jobs())} job(s)")
    scheduler.start()
    print("Scheduler stopped")

if __name__ == "__main__":
    run_scheduler()

from apscheduler.schedulers.blocking import BlockingScheduler
from typing import Dict, Tuple, NoReturn
import pandas as pd
import yaml
import json
import utils

def read_config(config_path: str) -> Dict[str, Any]:
    """Read YAML configuration."""
    with open(config_path, "r") as file:
        config = yaml.safe_load(file)
    return config

def load_and_validate_sales(csv_path: str, config: Dict[str, Any]) -> Tuple[pd.DataFrame, int, int]:
    """Load and validate sales CSV."""
    df = pd.read_csv(csv_path)
    df = df.dropna(subset=["product"])
    df = df[df["product"].str.startswith(config["product_prefix"])]
    df = df[df["quantity"].apply(utils.is_integer)]
    df["quantity"] = df["quantity"].astype(int)
    df = df[df["quantity"] <= config["max_quantity"]]
    df = df[df["price"].apply(utils.is_numeric_value)]
    df = df[df["price"] > 0]
    df = df[df["price"] >= config["min_price"]]
    return df, len(df), len(df)

def process_sales(df: pd.DataFrame, config: Dict[str, Any]) -> Tuple[Dict[str, Any], int]:
    """Process sales data."""
    if df.empty:
        return {"total_sales": 0.0, "unique_products": [], "top_products": {}}, 0
    df["amount"] = df["price"] * df["quantity"]
    total_sales = df["amount"].sum()
    unique_products = df["product"].unique().tolist()
    sales_by_product = df.groupby("product")["amount"].sum()
    top_products = sales_by_product.sort_values(ascending=False).head(3).to_dict()
    return {
        "total_sales": float(total_sales),
        "unique_products": unique_products,
        "top_products": top_products
    }, len(df)

def export_results(results: Dict[str, Any], json_path: str) -> None:
    """Export results to JSON."""
    with open(json_path, "w") as file:
        json.dump(results, file, indent=2)

def etl_task(csv_path: str, config_path: str, json_path: str) -> None:
    """ETL task."""
    config = read_config(config_path)
    df, valid_sales, total_records = load_and_validate_sales(csv_path, config)
    results, valid_sales = process_sales(df, config)
    export_results(results, json_path)

def run_scheduler(csv_path: str, config_path: str, json_path: str) -> NoReturn:
    """Run scheduler."""
    scheduler = BlockingScheduler()
    scheduler.add_job(
        etl_task,
        "interval",
        seconds=5,
        args=[csv_path, config_path, json_path],
        id="etl_task"
    )
    print("Starting scheduler...")
    print(f"Scheduled {len(scheduler.get_jobs())} job(s)")
    scheduler.start()
    print("Scheduler stopped")

def main() -> None:
    """Main function."""
    run_scheduler("data/sample.csv", "data/config.yaml", "data/sample_report.json")

if __name__ == "__main__":
    main()

from apscheduler.schedulers.blocking import BlockingScheduler
from typing import NoReturn

def print_time() -> None:
    """Print task execution."""
    print("Task executed")

def run_scheduler() -> NoReturn:
    """Run scheduler."""
    scheduler = BlockingScheduler()
    scheduler.add_job(print_time, "interval", seconds=5, id="print_task")  # Fix: 5 seconds
    print("Starting scheduler...")
    print(f"Scheduled {len(scheduler.get_jobs())} job(s)")
    scheduler.start()
    print("Scheduler stopped")

if __name__ == "__main__":
    run_scheduler()

import pytest
import json
import os
from typing import Dict, Any
from ex3_etl import etl_task, read_config

@pytest.fixture
def config() -> Dict[str, Any]:
    """Fixture for config."""
    return read_config("data/config.yaml")

def test_etl_task(tmp_path: Any, config: Dict[str, Any]) -> None:
    """Test ETL task."""
    json_path = tmp_path / "sample_report.json"
    etl_task("data/sample.csv", "data/config.yaml", str(json_path))
    assert os.path.exists(json_path)
    with open(json_path, "r") as file:
        results = json.load(file)
    assert results["total_sales"] == 2249.88

from apscheduler.schedulers.blocking import BlockingScheduler
from typing import NoReturn

def nonexistent_task() -> None:
    """Print task execution."""
    print("Task executed")

def run_scheduler() -> NoReturn:
    """Run scheduler."""
    scheduler = BlockingScheduler()
    scheduler.add_job(nonexistent_task, "interval", seconds=5, id="fixed_task")  # Fix: Define function
    print("Starting scheduler...")
    print(f"Scheduled {len(scheduler.get_jobs())} job(s)")
    scheduler.start()
    print("Scheduler stopped")

if __name__ == "__main__":
    run_scheduler()