Skip to main content

Loops and Conditions

Paral provides powerful control flow mechanisms through loops and conditions. These features allow you to create dynamic workflows that adapt to data, environment conditions, and runtime states.

Conditional Execution with @if​

The @if directive allows you to conditionally execute tasks based on variable values, comparisons, and logical operations.

Basic Conditional Syntax​

debug_mode = true
environment = "production"

@if debug_mode
task enable_logging {
  -> echo "Debug logging enabled"
  -> export LOG_LEVEL=debug
}

@if @is(environment, "production")
task production_setup {
  -> kubectl apply -f k8s/production/
}

Comparison Functions​

Paral provides comparison functions using @is and @isnot for use with @if:

user_count = 1000
error_rate = 5.2
server_status = "healthy"

@if @is(user_count, "gt", 500)
task scale_up {
  -> kubectl scale deployment app --replicas=10
}

@if @is(error_rate, "lt", 1.0)
task reduce_monitoring {
  -> @printf("Error rate acceptable: %.2f%%", error_rate)
}

@if @is(server_status, "healthy")
task proceed_deployment {
  -> sh scripts/deploy.sh
}

@if @isnot(server_status, "healthy")
task alert_ops_team {
  -> slack-cli send "Server unhealthy, deployment blocked" --channel=#ops
}

Available Comparison Functions:

  • @is(a, b) - Equals
  • @isnot(a, b) - Not equals
  • @is(a, "gt", b) - Greater than
  • @is(a, "lt", b) - Less than
  • @is(a, "gte", b) - Greater than or equal
  • @is(a, "lte", b) - Less than or equal

Logical Operations​

is_weekend = false
maintenance_window = true
critical_issue = false

@if @and(is_weekend, maintenance_window)
task scheduled_maintenance {
  -> sh scripts/maintenance.sh
}

@if @or(critical_issue, @is(environment, "emergency"))
task emergency_response {
  -> python3 scripts/emergency_deploy.py
}

@if @not(critical_issue)
task normal_operations {
  -> python3 scripts/routine_tasks.py
}

Logical Functions:

  • @and(condition1, condition2, ...) - All conditions must be true
  • @or(condition1, condition2, ...) - At least one condition must be true
  • @not(condition) - Negates the condition

Complex Conditional Logic​

cpu_usage = 85.5
memory_usage = 70.2
disk_usage = 45.0

@if @and(@is(cpu_usage, "gt", 80), @is(memory_usage, "gt", 75))
task scale_resources {
  -> @printf("High resource usage: CPU %.1f%%, Memory %.1f%%", cpu_usage, memory_usage)
  -> kubectl patch deployment app -p '{"spec":{"replicas":5}}'
}

@if @or(@is(cpu_usage, "gt", 90), @is(memory_usage, "gt", 90), @is(disk_usage, "gt", 85))
task critical_alert {
  -> @printf("Critical resource usage detected")
  -> python3 scripts/emergency_scale.py
}

Loops with @for​

The @for directive enables iteration over lists and arrays, executing the task for each item.

Basic Loop Syntax​

servers = ["web1", "web2", "web3"]
ports = [8080, 8081, 8082]

@for servers
task restart_server {
  -> @printf("Restarting server: %s", @value)
  -> ssh @value "sudo systemctl restart nginx"
}

@for ports
task check_port {
  -> @printf("Checking port %d (index %d)", @value, @key)
  -> netstat -tuln | grep @value
}

Loop Variables:

  • @value - The current item being processed
  • @key - The current index (0-based position in the list)

Iterating Over Different Data Types​

Simple Lists​

databases = ["users", "products", "analytics", "logs"]

@for databases
task backup_database {
  -> @printf("Backing up database %d: %s", @key, @value)
  -> pg_dump @value > @sprintf("backup-%s-%s.sql", @value, $(date +%Y%m%d))
  -> @stash[@sprintf("backup_%s", @value)] << @sprintf("backup-%s-%s.sql", @value, $(date +%Y%m%d))
}

Numeric Ranges​

worker_count = 5

@for @range(worker_count)
task start_worker {
  -> @printf("Starting worker %d", @value)
  -> python3 worker.py @sprintf("--id=%d", @value) &
}

File Lists​

@for @glob("*.csv")
task process_csv_file {
  -> @printf("Processing file %d: %s", @key, @value)
  -> python3 scripts/process_csv.py --input=@value --output=@sprintf("processed_%s", @value)
}

Advanced Loop Patterns​

Parallel Processing with Coordination​

image_files = ["image1.jpg", "image2.jpg", "image3.jpg", "image4.jpg"]

@for image_files
task process_image {
  -> @printf("Processing image %d: %s", @key, @value)
  -> convert @value -resize 800x600 @sprintf("resized_%s", @value)
  -> @buf[@sprintf("processed_%d", @key)] << @sprintf("resized_%s", @value)
}

@wait @eq(@len(@buf), @len(image_files))
task create_gallery {
  -> @printf("All %d images processed, creating gallery", @len(image_files))
  -> python3 scripts/create_gallery.py --images=@buf
}

Configuration Deployment​

environments = ["dev", "staging", "prod"]
services = ["api", "web", "worker"]

@for environments
task setup_namespace {
  -> @printf("Setting up namespace: %s", @value)
  -> kubectl create namespace @value
  -> @buf[@sprintf("ns_%s", @value)] << @value
}

@for services
@wait @eq(@len(@buf), @len(environments))
task deploy_service_to_all_envs {
  -> @printf("Deploying service: %s", @value)
  -> @for environments {
       kubectl apply -f @sprintf("k8s/%s/%s.yaml", @value, @outer_value) -n @value
     }
}

Nested Loops and Conditions​

Combining @for and @if​

servers = ["web1", "web2", "web3", "api1", "api2"]
maintenance_servers = ["web2", "api1"]

@for servers
@if @not(@contains(maintenance_servers, @value))
task health_check_active_servers {
  -> @printf("Checking active server: %s", @value)
  -> curl -f @sprintf("http://%s:8080/health", @value)
  -> @buf[@sprintf("health_%s", @value)] << echo "healthy"
}

Multi-Level Iteration​

regions = ["us-east", "us-west", "eu-central"]
instance_types = ["t3.micro", "t3.small", "t3.medium"]

@for regions
task setup_region {
  -> @printf("Setting up region: %s", @value)
  -> @stash[@sprintf("region_%d", @key)] << @value
  
  # Nested iteration over instance types
  -> @for instance_types {
       @printf("Creating instance type %s in region %s", @value, @stash(@sprintf("region_%d", @outer_key))
       aws ec2 describe-instance-types --instance-types @value --region @stash(@sprintf("region_%d", @outer_key))
     }
}

Real-World Examples​

Database Migration Pipeline​

migration_files = @glob("migrations/*.sql")
target_databases = ["app_db", "analytics_db", "audit_db"]

@for migration_files
task validate_migration {
  -> @printf("Validating migration %d: %s", @key, @value)
  -> sqlfluff lint @value
  -> @buf[@sprintf("migration_%d_valid", @key)] << echo "true"
}

@for target_databases
@wait @eq(@len(@buf), @len(migration_files))
task apply_migrations {
  -> @printf("Applying migrations to database: %s", @value)
  -> @for migration_files {
       @if @eq(@buf(@sprintf("migration_%d_valid", @key)), "true") {
         @printf("Applying %s to %s", @value, @outer_value)
         psql @outer_value -f @value
       }
     }
}

Service Health Monitoring​

services = ["web", "api", "database", "redis", "elasticsearch"]
health_threshold = 3

@schedule "*/30 * * * * *"  # Every 30 seconds
@for services
task monitor_service_health {
  -> @printf("Monitoring service: %s", @value)
  -> health_status=$(curl -s -o /dev/null -w "%{http_code}" @sprintf("http://%s:8080/health", @value))
  -> @buf[@sprintf("%s_health", @value)] << echo $health_status
  
  -> @if @isnot(@buf(@sprintf("%s_health", @value)), "200") {
       @buf[@sprintf("%s_failures", @value)] << @add(@buf(@sprintf("%s_failures", @value)), 1)
     }
  
  -> @if @is(@buf(@sprintf("%s_failures", @value)), "gt", health_threshold) {
       @buf[@sprintf("%s_alert_needed", @value)] << echo "true"
     }
}

@wait @contains(@buf, "_alert_needed")
task send_health_alerts {
  -> @for services {
       @if @eq(@buf(@sprintf("%s_alert_needed", @value)), "true") {
         @printf("ALERT: Service %s has failed health checks", @value)
         slack-cli send @sprintf("🚨 Service %s is unhealthy", @value) --channel=#alerts
       }
     }
}

Feature Flag Deployment​

feature_flags = ["new_ui", "advanced_analytics", "beta_api"]
target_environments = ["staging", "canary", "production"]
rollout_percentage = 10

@for feature_flags
task prepare_feature {
  -> @printf("Preparing feature flag: %s", @value)
  -> @stash[@sprintf("feature_%s_config", @value)] << cat @sprintf("features/%s.json", @value)
}

@for target_environments
@depend prepare_feature
task deploy_features {
  -> @printf("Deploying features to: %s", @value)
  -> @for feature_flags {
       @if @is(@outer_value, "production") {
         @if @is(rollout_percentage, "gte", 50) {
           @printf("Deploying %s to production with %d%% rollout", @value, rollout_percentage)
           kubectl apply -f @sprintf("features/%s-prod.yaml", @value)
         }
       } else {
         @printf("Deploying %s to %s", @value, @outer_value)
         kubectl apply -f @sprintf("features/%s-%s.yaml", @value, @outer_value)
       }
     }
}

Dynamic Resource Scaling​

load_metrics = [45.2, 78.9, 92.1, 65.4, 38.7]
scale_threshold = 75.0
min_replicas = 2
max_replicas = 20

@for load_metrics
task evaluate_scaling {
  -> @printf("Evaluating load metric %d: %.1f%%", @key, @value)
  
  -> @if @is(@value, "gt", scale_threshold) {
       current_replicas=$(@buf("current_replicas"))
       new_replicas=$(($current_replicas + 2))
       @if @is($new_replicas, "lte", max_replicas) {
         @printf("Scaling up to %d replicas", $new_replicas)
         kubectl scale deployment app --replicas=$new_replicas
         @buf["current_replicas"] << echo $new_replicas
       }
     }
  
  -> @if @and(@is(@value, "lt", 30.0), @is(@buf("current_replicas"), "gt", min_replicas)) {
       new_replicas=$((@buf("current_replicas") - 1))
       @printf("Scaling down to %d replicas", $new_replicas)
       kubectl scale deployment app --replicas=$new_replicas
       @buf["current_replicas"] << echo $new_replicas
     }
}

Loop Control and Optimization​

Breaking Out of Loops​

servers = ["web1", "web2", "web3", "web4", "web5"]

@for servers
task find_healthy_server {
  -> @printf("Testing server: %s", @value)
  -> response=$(curl -s -o /dev/null -w "%{http_code}" @sprintf("http://%s:8080/health", @value))
  
  -> @if @is($response, "200") {
       @buf["healthy_server"] << @value
       @printf("Found healthy server: %s", @value)
       @break  # Exit the loop early
     }
}

Conditional Loop Execution​

backup_files = @glob("backups/*.sql")
restore_needed = true

@if restore_needed
@for backup_files
task restore_from_backup {
  -> @printf("Attempting restore from: %s", @value)
  -> backup_date=$(echo @value | grep -o '[0-9]\{8\}')
  
  -> @if @is($backup_date, "gt", $(date -d "7 days ago" +%Y%m%d)) {
       @printf("Using recent backup: %s", @value)
       psql mydb < @value
       @buf["restore_completed"] << echo "true"
       @break
     }
}