Best Practices for RPA Processes

Designing effective RPA processes is both an art and a science. While automation technology has advanced tremendously, the difference between a process that simply works and one that’s maintainable, scalable, and reliable lies in following proven best practices. This comprehensive guide covers everything you need to know to build world-class RPA processes in Heptora.

Core Philosophy

Great RPA processes are simple, reusable, well-documented, and resilient to errors. Heptora’s platform is designed from the ground up to help you achieve these goals with built-in features that encourage best practices.

Why Best Practices Matter

Before diving into specific practices, it’s important to understand why they matter:

Without Best Practices	With Best Practices
Processes break frequently	Robust, reliable execution
Difficult to maintain and update	Easy to modify and extend
Knowledge locked in individual developers	Transferable, documented knowledge
Long development cycles	Rapid development and deployment
High technical debt	Clean, maintainable code
Difficult to scale	Easily replicable patterns

The business impact:

• Reduced maintenance costs: 60-70% less time spent fixing broken processes
• Faster development: 40-50% reduction in development time through reuse
• Higher success rates: 95%+ execution success vs 70-80% without best practices
• Easier scaling: Deploy new automations 3-4x faster
• Better compliance: Clearer audit trails and documentation

Investment in Quality

Time spent following best practices upfront pays dividends throughout the automation lifecycle. A well-designed process is a long-term asset, while a poorly designed one becomes technical debt.

Fundamental Principles of RPA Process Design

The KISS Principle: Keep It Simple, Stupid

Simplicity is the foundation of maintainable automation. Complex processes are:

• Harder to debug when they fail
• More difficult for others to understand
• More fragile and prone to breaking
• Slower to execute
• Costly to maintain

How Heptora enforces simplicity:

• Visual workflow designer: See your entire process at a glance
• Pre-built automation blocks: No need to reinvent common operations
• Process templates: Start with proven patterns
• Automatic validation: Warns when processes become too complex

Example: Simple vs Complex

❌ Overly complex approach:

1. Open Excel
2. Read cell A1
3. If A1 > 100 then
4.   Read cell B1
5.   Open browser
6.   Navigate to URL in B1
7.   Wait 5 seconds
8.   Find element by XPath
9.   If element exists then
10.    Click element
11.    Wait for page load
12.    Read response
13.    If response contains "success" then
14.      Close browser
15.      Write "OK" to Excel cell C1
16.    Else
17.      Handle error scenario A
18.  Else
19.    Handle error scenario B
20. Else
21.   Handle error scenario C

✅ Simple, modular approach:

1. Use "Read Excel Data" block → Store in variable
2. Use "Process Web Transaction" automation block
3. Use "Update Excel Result" block

By leveraging Heptora’s pre-built blocks, you reduce 21 steps to 3, making the process easier to understand, maintain, and debug.

Single Responsibility Principle

Each process or block should do one thing and do it well.

Benefits:

• Easier testing: Test individual components in isolation
• Better reusability: Small, focused components can be used in many processes
• Simpler debugging: Problems are easier to locate
• Clearer documentation: Each component has a clear, single purpose

How to apply in Heptora:

❌ Process doing too much: “Complete Order Management”

• Reads orders from Excel
• Validates customer data
• Checks inventory
• Processes payment
• Updates ERP
• Sends confirmation email
• Generates reports

✅ Separate, focused processes:

• “Read Order Data” (reads and validates input)
• “Validate Customer” (checks customer information)
• “Check Inventory” (verifies product availability)
• “Process Payment” (handles payment transaction)
• “Update ERP System” (posts to ERP)
• “Send Notification” (sends emails)
• “Generate Report” (creates summary report)

Each can be tested, maintained, and reused independently.

Design for Failure

Assume things will go wrong and plan accordingly. Systems crash, websites change, data is malformed, networks fail. Your processes must be resilient.

Failure scenarios to consider:

• Connectivity failures: Network drops, API timeouts
• Data quality issues: Missing fields, wrong formats, invalid values
• System availability: Target applications down for maintenance
• UI changes: Websites or applications update their interface
• Resource constraints: Memory, CPU, disk space limitations
• Authentication failures: Expired credentials, locked accounts

Heptora’s built-in resilience features:

• Automatic retries: Configure retry logic for transient failures
• Exception handling: Built-in try-catch mechanisms
• Result system: OK, KO, ERROR states for granular failure tracking
• Logging and monitoring: Detailed execution logs for debugging
• Rollback capabilities: Undo operations when processes fail
• Alerts and notifications: Immediate notification of failures

Defensive Programming

Always ask: “What could go wrong here?” and build appropriate handling into your process.

Keeping Blocks Simple

Why Simplicity in Blocks Matters

Individual blocks are the building units of your processes. Keeping them simple:

• Makes processes easier to understand at a glance
• Reduces cognitive load for developers
• Minimizes potential points of failure
• Speeds up execution
• Simplifies testing and validation

How Heptora Maintains Simple Blocks

1. Pre-built Automation Blocks

Heptora provides a comprehensive library of pre-tested automation blocks for common operations:

Category	Examples	Benefit
Communication	Send Email, Send Slack Message, Webhook	No need to write SMTP or API code
File Operations	Read File, Write File, Move File, ZIP/Unzip	Handles errors and edge cases automatically
Excel Management	Read Excel, Write Excel, Filter Data	Built-in validation and formatting
Database	SQL Query, Insert Record, Update Record	Connection pooling and transaction management
Web Automation	Navigate, Click, Type, Extract Data	Robust element detection and waiting

Example: Sending an email

❌ Without automation blocks:

import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.mime.base import MIMEBase
from email import encoders

# Configure SMTP connection
smtp_server = "smtp.gmail.com"
smtp_port = 587
sender_email = heptora.data.get("email_user")
sender_password = heptora.data.get("email_password")

# Create message
message = MIMEMultipart()
message["From"] = sender_email
message["To"] = recipient_email
message["Subject"] = subject

# Add body
message.attach(MIMEText(body, "plain"))

# Add attachments
if attachment_path:
    with open(attachment_path, "rb") as attachment:
        part = MIMEBase("application", "octet-stream")
        part.set_payload(attachment.read())
        encoders.encode_base64(part)
        part.add_header("Content-Disposition", f"attachment; filename= {filename}")
        message.attach(part)

# Send email
try:
    server = smtplib.SMTP(smtp_server, smtp_port)
    server.starttls()
    server.login(sender_email, sender_password)
    server.send_message(message)
    server.quit()
    heptora.log.info("Email sent successfully")
except Exception as e:
    heptora.log.result("error", f"Failed to send email: {str(e)}")

✅ With Heptora automation block:

Use "Send Email" block:
- To: {{ recipient_email }}
- Subject: {{ subject }}
- Body: {{ body }}
- Attachment: {{ attachment_path }}

The automation block handles all the complexity, error management, and edge cases.

2. Visual Workflow Builder

Heptora’s visual interface helps maintain simplicity:

• Drag-and-drop design: Build processes visually without writing code
• Clear flow visualization: See the entire process structure at a glance
• Automatic validation: Real-time feedback on process design
• Complexity warnings: Alerts when a process becomes too complex

3. Configuration Over Code

Instead of writing code for every scenario, use configuration:

❌ Code-heavy approach:

# Read Excel with custom parsing
import openpyxl
workbook = openpyxl.load_workbook("data.xlsx")
sheet = workbook["Sheet1"]

data = []
for row in sheet.iter_rows(min_row=2, values_only=True):
    if row[0] is not None:  # Check for empty rows
        record = {
            "id": row[0],
            "name": row[1],
            "amount": float(row[2]) if row[2] else 0,
            "date": row[3].strftime("%Y-%m-%d") if row[3] else None
        }
        data.append(record)

heptora.data.set("records", data)

✅ Configuration-driven approach:

Use "Read Excel" block:
- File: data.xlsx
- Sheet: Sheet1
- Columns: ["id", "name", "amount", "date"]
- Has Header: Yes
- Skip Empty Rows: Yes
→ Store in variable: records

Block Simplicity Checklist

Use this checklist when designing blocks:

• Single purpose: Does this block do exactly one thing?
• Clear naming: Is it obvious what this block does from its name?
• Minimal configuration: Does it require only essential parameters?
• No nested complexity: Is it free from deeply nested conditionals?
• Reusable: Could this block be used in other processes?
• Well-documented: Are parameters and outputs clearly described?
• Error handling: Does it handle expected errors gracefully?
• Testable: Can it be tested independently?

Anti-Pattern: The “God Block”

Avoid creating blocks that try to do everything:

❌ God Block Anti-Pattern:

Block: "Process Everything"
- Reads data from multiple sources (Excel, API, Database)
- Validates all data types
- Applies complex business rules
- Updates multiple systems
- Sends notifications
- Generates reports
- Archives results
- 500+ lines of code
- 30+ configuration parameters

This is a maintenance nightmare. When it breaks, it’s difficult to determine why. When requirements change, it’s risky to modify.

✅ Proper decomposition:

Pipeline of focused blocks:
1. "Read Data Sources" → outputs: raw_data
2. "Validate Data" → outputs: valid_data, invalid_data
3. "Apply Business Rules" → outputs: processed_data
4. "Update Systems" → outputs: update_results
5. "Send Notifications" → outputs: notification_status
6. "Generate Report" → outputs: report_file
7. "Archive Results" → outputs: archive_path

Each block is simple, testable, and maintainable.

Reusability: Don’t Repeat Yourself

The Importance of Reusability

Reusability is one of the most powerful principles in RPA development:

Benefits:

• Faster development: Build new processes in hours instead of days
• Consistency: Same operation performed the same way everywhere
• Easier maintenance: Fix once, improve everywhere
• Knowledge sharing: Capture best practices in reusable components
• Quality improvement: Reused components are battle-tested
• Cost reduction: Avoid duplicate development effort

Cost of not reusing:

• Developers reinvent the wheel constantly
• Same functionality implemented differently across processes
• Bug fixes must be applied in multiple places
• Inconsistent behavior across automations
• Higher maintenance burden

Creating Reusable Components in Heptora

1. Automation Blocks

Automation blocks are Heptora’s primary reusability mechanism. Any sequence of steps that you use more than once should be converted into an automation block.

When to create an automation block:

• ✅ You use the same sequence in 2+ processes
• ✅ The operation is conceptually distinct and reusable
• ✅ You want to standardize how something is done
• ✅ The logic is complex and you want to abstract it
• ✅ You anticipate future reuse

Example: Login to Corporate Portal

Instead of repeating login logic in every process that needs it:

Create automation block: "Login to Corporate Portal"
Inputs:
  - username (from secrets)
  - password (from secrets)
Outputs:
  - login_successful (boolean)
  - session_token (string)

Steps:
1. Open browser
2. Navigate to portal URL
3. Wait for login form
4. Enter username
5. Enter password
6. Click login button
7. Wait for dashboard
8. Verify successful login
9. Extract session token
10. Return results

Now, any process can simply use:

Call block: "Login to Corporate Portal"
→ Outputs: login_successful, session_token

This block can be:

• Used in dozens of processes
• Updated once when the portal changes
• Tested independently
• Documented centrally
• Monitored for performance

2. Process Templates

Heptora’s process templates provide pre-built, production-ready automations for common scenarios.

Using process templates:

• Speed: Deploy in less than 1 hour vs weeks of development
• Best practices: Built by experts, following all best practices
• Automatic updates: Maintained by Heptora when target systems change
• Customizable: Adapt to your specific needs
• Proven: Already tested in production environments

Available template categories:

• Public administration (FACe, official registries)
• Financial services (bank reconciliation, payments)
• Insurance (claim processing, invoice submission)
• Human resources (payroll portals, time tracking)
• E-commerce (inventory sync, order management)

Example: FACe Electronic Invoicing

Instead of building from scratch:

Development time: 2-6 weeks
Testing time: 1-2 weeks
Maintenance: Your responsibility
Cost: $5,000-$15,000
Risk: High (unproven)

Use the FACe template:

Setup time: Less than 1 hour
Testing: Already validated
Maintenance: Automatic updates by Heptora
Cost: Included in subscription
Risk: Low (production-proven)

How templates promote reusability:

• Start with a working foundation
• Customize only what you need
• Leverage expert implementations
• Benefit from continuous improvements
• Share configurations across teams

3. Variable and Data Reuse

Use Heptora’s variable system to pass data between blocks:

# Store data for reuse
heptora.data.set("customer_id", customer_id)
heptora.data.set("order_total", total_amount)
heptora.data.set("invoice_data", invoice_dict)

# Retrieve later in another block
customer_id = heptora.data.get("customer_id")
order_total = heptora.data.get("order_total")
invoice_data = heptora.data.get("invoice_data")

Best practices for variables:

• Use descriptive names: customer_id not x1
• Namespace related variables: order_id, order_date, order_total
• Document expected data types and structures
• Clear variables when no longer needed
• Validate data when retrieving

4. Code Libraries and Helpers

For Python-based automations, create reusable functions:

invoice_helpers.py

def validate_invoice(invoice):
    """
    Validates invoice data according to business rules.

    Args:
        invoice (dict): Invoice data to validate

    Returns:
        tuple: (is_valid, error_message)
    """
    if not invoice.get("number"):
        return False, "Missing invoice number"

    if not invoice.get("amount") or float(invoice["amount"]) <= 0:
        return False, "Invalid amount"

    if not validate_tax_id(invoice.get("tax_id")):
        return False, "Invalid tax ID format"

    return True, ""

def format_currency(amount, currency="EUR"):
    """Formats amount as currency string."""
    return f"{amount:.2f} {currency}"

def calculate_tax(base_amount, tax_rate=0.21):
    """Calculates tax amount."""
    return base_amount * tax_rate

# Use in multiple processes
from invoice_helpers import validate_invoice, calculate_tax

is_valid, error = validate_invoice(invoice)
if not is_valid:
    heptora.log.force_ko(f"Validation failed: {error}")

tax = calculate_tax(base_amount)

5. Configuration Files

Store reusable configuration in centralized files:

# config.py - Shared configuration

PORTAL_URLS = {
    "development": "https://dev.portal.com",
    "staging": "https://staging.portal.com",
    "production": "https://portal.com"
}

RETRY_CONFIG = {
    "max_attempts": 3,
    "backoff_seconds": 5,
    "timeout_seconds": 30
}

VALIDATION_RULES = {
    "invoice": {
        "max_amount": 10000,
        "required_fields": ["number", "date", "amount", "customer_id"]
    }
}

# Use in processes
from config import PORTAL_URLS, RETRY_CONFIG

url = PORTAL_URLS["production"]
page.goto(url, timeout=RETRY_CONFIG["timeout_seconds"] * 1000)

Reusability Patterns

Pattern 1: The Login Block

Nearly every automation needs to log into something. Create standard login blocks:

Automation Block: "Login to [System Name]"
- Handles authentication
- Manages session tokens
- Detects and reports failures
- Implements retry logic
- Returns authenticated connection

Use cases:
- Web portal logins
- Database connections
- API authentication
- Desktop application access

Pattern 2: The Data Validator

Data validation is repeated across processes. Standardize it:

Automation Block: "Validate Business Data"
Inputs:
  - data_type (customer, invoice, order, etc.)
  - data_object (the data to validate)
Outputs:
  - is_valid (boolean)
  - errors (list of error messages)

Validates according to:
- Business rules
- Data type constraints
- Required fields
- Format requirements

Pattern 3: The Notification Dispatcher

Sending notifications happens everywhere. Centralize it:

Automation Block: "Send Notification"
Inputs:
  - notification_type (success, error, warning, info)
  - message
  - recipients
  - channels (email, Slack, Teams, webhook)

Handles:
- Multiple notification channels
- Template application
- Retry on failures
- Rate limiting

Pattern 4: The Report Generator

Many processes generate similar reports. Template it:

Automation Block: "Generate Excel Report"
Inputs:
  - data (array of records)
  - template_name
  - report_title
  - output_path

Features:
- Consistent formatting
- Company branding
- Automatic charts
- Summary statistics
- PDF conversion option

Reusability Anti-Patterns

❌ Copy-Paste Development

Process A: Contains login logic (100 lines)
Process B: Copy-pasted same login logic
Process C: Copy-pasted again with slight modification
Problem: Portal changes → must update 3+ places

❌ Over-Generic Components

Block: "Do Everything Configurably"
- 50+ parameters
- Tries to handle every possible scenario
- Configuration is more complex than custom code
Problem: Too flexible → nobody knows how to use it

❌ Tight Coupling

Block: "Process Order and Send Email and Update Database"
Problem: Can't reuse parts independently

✅ Proper Reusability

Clear, focused, loosely coupled components:
- "Validate Order Data"
- "Submit Order to API"
- "Send Confirmation Email"
- "Update Order Database"

Each can be used independently and recombined as needed.

Exception Handling: Building Resilient Processes

Understanding Exceptions in RPA

Exceptions are inevitable in automation. The question isn’t if they’ll occur, but when and how you’ll handle them.

Common exception categories:

Category	Examples	Typical Cause
Data Quality	Missing fields, invalid formats	Poor input data
System Availability	Application not responding, API down	Infrastructure issues
Network	Timeout, connection refused	Connectivity problems
Authentication	Invalid credentials, expired token	Security/access issues
Business Logic	Rule violations, state conflicts	Business rule enforcement
UI Changes	Element not found, unexpected popup	Application updates
Resource	Out of memory, disk full	Capacity constraints

Heptora’s Result System: OK, KO, ERROR

Heptora provides a sophisticated three-state result system that elegantly handles different types of failures.

The three states explained:

✅ OK (Success)

Meaning: The operation completed successfully
When to use: Everything worked as expected
Example: Invoice submitted and confirmation received
Action required: None

⚠️ KO (Controlled Error)

Meaning: The operation failed for a known, expected reason
When to use: Business rule violations, data validation failures
Example: Invoice already paid, missing required field, duplicate record
Action required: Human review and correction of input data

❌ ERROR (Uncontrolled Error)

Meaning: The operation failed for an unexpected, technical reason
When to use: System failures, connectivity issues, unexpected exceptions
Example: Website didn't load, API timeout, database connection failed
Action required: Technical intervention or retry

Why this matters:

Traditional automation tools often treat all failures the same way. Heptora’s three-state system allows you to:

• Different handling: KOs don’t trigger error alerts but ERRORs do
• Proper routing: KOs go to business teams, ERRORs to technical teams
• Accurate metrics: Distinguish data quality issues from technical failures
• Better reporting: Stakeholders see actionable information
• Intelligent retries: Retry ERRORs but not KOs

Implementing Exception Handling in Heptora

1. Using log.result()

# Process an invoice
try:
    # Validate data first
    if not invoice.get("number"):
        heptora.log.result("ko", "Invoice missing required number field")
        continue

    if invoice.get("status") == "Paid":
        heptora.log.result("ko", "Invoice already paid, no action needed")
        continue

    # Attempt to submit
    result = submit_to_portal(invoice)

    if result["status"] == "success":
        heptora.log.result("ok", f"Invoice {invoice['number']} submitted successfully")
    else:
        heptora.log.result("ko", f"Portal rejected invoice: {result['reason']}")

except ConnectionError as e:
    # Network issue - technical error
    heptora.log.result("error", f"Connection failed: {str(e)}")

except Exception as e:
    # Unexpected error - technical error
    heptora.log.result("error", f"Unexpected error: {str(e)}")

2. Using log.force_ko()

For immediate termination with a KO status:

def process_customer(customer):
    # Validate required fields
    required_fields = ["name", "email", "tax_id"]

    for field in required_fields:
        if not customer.get(field):
            heptora.log.force_ko(f"Customer missing required field: {field}")
            # Execution stops here, status set to KO

    # Validate email format
    if not validate_email(customer["email"]):
        heptora.log.force_ko(f"Invalid email format: {customer['email']}")

    # If we reach here, data is valid
    # Continue with processing...

3. Try-Except-Finally Pattern

Python’s exception handling integrates seamlessly with Heptora:

import heptora
from typing import Dict, Any

def process_order(order: Dict[str, Any]) -> bool:
    """
    Processes an order with comprehensive error handling.

    Returns:
        bool: True if successful, False otherwise
    """
    order_number = order.get("number", "UNKNOWN")
    browser_opened = False

    try:
        # Step 1: Validate data (KO if fails)
        heptora.log.info(f"Validating order {order_number}")
        validate_order(order)  # Raises ValueError if invalid

        # Step 2: Connect to system (ERROR if fails)
        heptora.log.info("Opening browser")
        page = heptora.browser.connect()
        browser_opened = True

        # Step 3: Navigate and submit
        heptora.log.info(f"Submitting order {order_number}")
        page.goto("https://portal.com/orders")
        page.wait_for_selector("#order-form")

        # Fill form
        page.locator("#order-number").fill(order["number"])
        page.locator("#customer").fill(order["customer"])
        page.locator("#amount").fill(str(order["amount"]))
        page.locator("button#submit").click()

        # Wait for confirmation
        page.wait_for_selector(".success-message", timeout=10000)

        heptora.log.result("ok", f"Order {order_number} processed successfully")
        return True

    except ValueError as e:
        # Data validation error - KO
        heptora.log.result("ko", f"Order {order_number} validation failed: {str(e)}")
        return False

    except TimeoutError:
        # UI element didn't appear - ERROR
        heptora.log.result("error", f"Order {order_number}: Timeout waiting for UI element")
        return False

    except ConnectionError as e:
        # Network issue - ERROR
        heptora.log.result("error", f"Order {order_number}: Connection error: {str(e)}")
        return False

    except Exception as e:
        # Unexpected error - ERROR
        heptora.log.result("error", f"Order {order_number}: Unexpected error: {str(e)}")
        return False

    finally:
        # Cleanup always happens
        if browser_opened:
            heptora.log.info("Closing browser")
            # Browser cleanup handled by Heptora

        heptora.log.info(f"Finished processing order {order_number}")

4. Retry Logic for Transient Failures

Some errors are temporary and retrying can help:

def process_with_retry(item, max_attempts=3, backoff_seconds=5):
    """
    Processes an item with retry logic for transient failures.

    Args:
        item: The item to process
        max_attempts: Maximum number of retry attempts
        backoff_seconds: Seconds to wait between retries

    Returns:
        bool: True if successful, False otherwise
    """
    for attempt in range(1, max_attempts + 1):
        try:
            heptora.log.info(f"Attempt {attempt}/{max_attempts} for item {item['id']}")

            # Attempt the operation
            result = submit_to_api(item)

            # Success!
            heptora.log.result("ok", f"Item {item['id']} processed successfully")
            return True

        except ConnectionError as e:
            # Transient network error
            if attempt < max_attempts:
                heptora.log.info(
                    f"Connection error on attempt {attempt}. "
                    f"Retrying in {backoff_seconds} seconds..."
                )
                heptora.time.pause(backoff_seconds)
            else:
                # Max attempts exhausted
                heptora.log.result(
                    "error",
                    f"Item {item['id']}: Connection failed after {max_attempts} attempts"
                )
                return False

        except ValueError as e:
            # Data error - don't retry
            heptora.log.result("ko", f"Item {item['id']}: Data validation error: {str(e)}")
            return False

        except Exception as e:
            # Unexpected error - don't retry
            heptora.log.result("error", f"Item {item['id']}: Unexpected error: {str(e)}")
            return False

    return False

5. Graceful Degradation

When part of a process fails, continue with what you can:

def process_invoice_batch(invoices):
    """
    Processes a batch of invoices with graceful degradation.
    """
    successful = []
    failed_ko = []
    failed_error = []

    for invoice in invoices:
        number = invoice.get("number", "UNKNOWN")

        try:
            # Validate
            if not validate_invoice(invoice):
                failed_ko.append(invoice)
                heptora.log.result("ko", f"Invoice {number}: Validation failed")
                continue

            # Process
            submit_invoice(invoice)
            successful.append(invoice)
            heptora.log.result("ok", f"Invoice {number}: Success")

        except Exception as e:
            # Technical error
            failed_error.append(invoice)
            heptora.log.result("error", f"Invoice {number}: {str(e)}")
            continue  # Don't let one failure stop the batch

    # Summary
    heptora.log.info(f"Batch complete: {len(successful)} OK, "
                     f"{len(failed_ko)} KO, {len(failed_error)} ERROR")

    # Generate reports for failures
    if failed_ko:
        generate_ko_report(failed_ko)
    if failed_error:
        generate_error_report(failed_error)

    return successful, failed_ko, failed_error

Logging Strategies

Effective logging is crucial for troubleshooting and improvement.

Logging levels in Heptora:

# Info: General progress updates
heptora.log.info("Starting invoice processing")
heptora.log.info(f"Processing invoice {invoice_number}")
heptora.log.info("Validation completed")

# Results: Final outcome of operations
heptora.log.result("ok", "Invoice submitted successfully")
heptora.log.result("ko", "Invoice data validation failed")
heptora.log.result("error", "Portal connection timeout")

# Force KO: Immediate termination with controlled error
heptora.log.force_ko("Critical validation failure - stopping process")

What to log:

✅ Do log:

• Process start and completion
• Key milestones and state transitions
• Validation results
• External system interactions
• Data transformations
• Decision points (if/else paths taken)
• Performance metrics (processing times)
• Warnings about unusual conditions

❌ Don’t log:

• Sensitive data (passwords, credit cards, personal info)
• Excessive detail that clutters logs
• Redundant information
• Non-actionable information

Structured logging example:

def process_customer_order(order):
    """Process order with comprehensive logging."""
    order_id = order["id"]

    # Log entry with context
    heptora.log.info(f"=== Processing Order {order_id} ===")
    heptora.log.info(f"Customer: {order['customer_name']}")
    heptora.log.info(f"Amount: {order['amount']}")
    heptora.log.info(f"Items: {len(order['items'])}")

    # Log validation
    heptora.log.info("Step 1: Validating order data")
    is_valid, error = validate_order(order)
    if not is_valid:
        heptora.log.result("ko", f"Order {order_id}: Validation failed - {error}")
        return False
    heptora.log.info("✓ Validation passed")

    # Log external calls
    heptora.log.info("Step 2: Checking inventory")
    try:
        inventory = check_inventory(order["items"])
        heptora.log.info(f"✓ Inventory check complete: {inventory['available']}/{inventory['requested']}")
    except Exception as e:
        heptora.log.result("error", f"Order {order_id}: Inventory check failed - {str(e)}")
        return False

    # Log processing
    heptora.log.info("Step 3: Submitting order")
    try:
        result = submit_order(order)
        heptora.log.info(f"✓ Order submitted - Confirmation: {result['confirmation_number']}")
    except Exception as e:
        heptora.log.result("error", f"Order {order_id}: Submission failed - {str(e)}")
        return False

    # Log success
    heptora.log.result("ok", f"Order {order_id} processed successfully")
    heptora.log.info(f"=== Completed Order {order_id} ===")
    return True

Recovery Mechanisms

Build processes that can recover from failures:

1. Transaction Rollback

def process_with_rollback(record):
    """Process a record with transaction support."""
    changes_made = []

    try:
        # Operation 1: Update database
        heptora.log.info("Updating database")
        db_result = update_database(record)
        changes_made.append(("database", db_result["id"]))

        # Operation 2: Call API
        heptora.log.info("Calling external API")
        api_result = call_external_api(record)
        changes_made.append(("api", api_result["transaction_id"]))

        # Operation 3: Update file
        heptora.log.info("Updating file")
        file_result = update_file(record)
        changes_made.append(("file", file_result["path"]))

        # All succeeded
        heptora.log.result("ok", "Record processed successfully")
        return True

    except Exception as e:
        # Something failed - rollback all changes
        heptora.log.info(f"Error occurred: {str(e)}")
        heptora.log.info("Rolling back changes...")

        for change_type, change_id in reversed(changes_made):
            try:
                if change_type == "database":
                    rollback_database(change_id)
                    heptora.log.info(f"✓ Rolled back database change {change_id}")
                elif change_type == "api":
                    cancel_api_transaction(change_id)
                    heptora.log.info(f"✓ Rolled back API transaction {change_id}")
                elif change_type == "file":
                    restore_file(change_id)
                    heptora.log.info(f"✓ Rolled back file change {change_id}")
            except Exception as rollback_error:
                heptora.log.info(f"✗ Failed to rollback {change_type}: {str(rollback_error)}")

        heptora.log.result("error", f"Record processing failed and rolled back: {str(e)}")
        return False

2. Checkpoint and Resume

def process_large_batch(items):
    """Process large batch with checkpoint capability."""
    checkpoint_file = "process_checkpoint.json"

    # Load checkpoint if exists
    processed_ids = load_checkpoint(checkpoint_file)
    heptora.log.info(f"Resuming from checkpoint: {len(processed_ids)} already processed")

    for i, item in enumerate(items):
        # Skip already processed items
        if item["id"] in processed_ids:
            continue

        try:
            # Process item
            process_item(item)
            heptora.log.result("ok", f"Item {item['id']} processed")

            # Save checkpoint after each success
            processed_ids.add(item["id"])
            save_checkpoint(checkpoint_file, processed_ids)

        except Exception as e:
            heptora.log.result("error", f"Item {item['id']} failed: {str(e)}")
            # Don't stop - continue with next item
            continue

    # Clean up checkpoint when complete
    remove_checkpoint(checkpoint_file)
    heptora.log.info("Batch processing complete")

3. Quarantine and Retry Queue

def process_with_quarantine(items):
    """Process items with quarantine for problematic records."""
    retry_queue = []
    quarantine = []

    for item in items:
        try:
            process_item(item)
            heptora.log.result("ok", f"Item {item['id']} processed")

        except TransientError as e:
            # Temporary error - add to retry queue
            retry_queue.append(item)
            heptora.log.info(f"Item {item['id']} added to retry queue")

        except DataError as e:
            # Data problem - quarantine for manual review
            item["_error"] = str(e)
            quarantine.append(item)
            heptora.log.result("ko", f"Item {item['id']} quarantined: {str(e)}")

        except Exception as e:
            # Unknown error - quarantine
            item["_error"] = str(e)
            quarantine.append(item)
            heptora.log.result("error", f"Item {item['id']} quarantined: {str(e)}")

    # Retry failed items
    if retry_queue:
        heptora.log.info(f"Retrying {len(retry_queue)} items")
        heptora.time.pause(5)  # Brief pause before retry

        for item in retry_queue:
            try:
                process_item(item)
                heptora.log.result("ok", f"Item {item['id']} processed on retry")
            except Exception as e:
                item["_error"] = str(e)
                quarantine.append(item)
                heptora.log.result("error", f"Item {item['id']} failed on retry")

    # Generate quarantine report
    if quarantine:
        heptora.log.info(f"Generating quarantine report for {len(quarantine)} items")
        generate_quarantine_report(quarantine)

Exception Handling Best Practices Checklist

• Distinguish KO from ERROR: Use appropriate result types
• Specific exception types: Catch specific exceptions, not just generic Exception
• Meaningful error messages: Include context and actionable information
• Log before failing: Capture state before raising exceptions
• Clean up resources: Use finally blocks for cleanup
• Fail gracefully: Don’t let one item failure stop entire batch
• Retry transient failures: Implement retry logic with backoff
• Don’t retry permanent failures: Don’t waste time on unrecoverable errors
• Quarantine bad data: Separate for manual review
• Monitor and alert: Set up notifications for critical failures
• Document error scenarios: Help users understand what went wrong
• Test error paths: Ensure error handling actually works

Process Design Patterns

Design patterns are proven solutions to common problems. Here are essential patterns for RPA processes.

Pattern 1: The Pipeline Pattern

Break complex processes into sequential stages, where each stage:

• Performs one transformation or operation
• Receives input from previous stage
• Produces output for next stage
• Can be tested independently

Example: Invoice Processing Pipeline

Stage 1: Data Acquisition
→ Read invoices from Excel
→ Output: raw_invoice_data

Stage 2: Validation
→ Validate invoice data
→ Output: valid_invoices, invalid_invoices

Stage 3: Enrichment
→ Add customer details from CRM
→ Output: enriched_invoices

Stage 4: Submission
→ Submit to FACe portal
→ Output: submission_results

Stage 5: Confirmation
→ Download confirmations
→ Output: confirmation_documents

Stage 6: Reporting
→ Generate summary report
→ Output: final_report

Advantages:

• Clear flow and dependencies
• Easy to add/remove stages
• Each stage independently testable
• Easy to identify bottlenecks
• Can parallelize independent stages

Pattern 2: The Dispatcher-Performer Pattern

Separate coordination logic from execution logic:

Dispatcher Process:
- Reads work queue
- Checks worker availability
- Assigns work to performers
- Monitors progress
- Aggregates results

Performer Process:
- Receives work item
- Executes business logic
- Reports results back
- Handles errors locally

Example: Order Processing
Dispatcher: "Order Queue Manager"
- Monitors order queue
- Assigns orders to available robots
- Tracks completion
- Handles failures

Performers: "Order Processor" (multiple instances)
- Process assigned order
- Update inventory
- Generate invoice
- Send confirmation
- Report completion

Advantages:

• Scalability: Add more performers as needed
• Isolation: Performer failure doesn’t affect others
• Load balancing: Distribute work optimally
• Monitoring: Central visibility of all work

Pattern 3: The State Machine Pattern

For processes with multiple states and transitions:

Example: Invoice Approval Process

States:
- SUBMITTED
- PENDING_REVIEW
- APPROVED
- REJECTED
- PAID

Transitions:
SUBMITTED → PENDING_REVIEW (when reviewer assigned)
PENDING_REVIEW → APPROVED (when reviewer approves)
PENDING_REVIEW → REJECTED (when reviewer rejects)
APPROVED → PAID (when payment processed)
REJECTED → SUBMITTED (when resubmitted with corrections)

Code:
class InvoiceStateMachine:
    def __init__(self, invoice):
        self.invoice = invoice
        self.state = invoice.get("state", "SUBMITTED")

    def transition(self, new_state, reason=None):
        """Transition to new state with validation."""
        valid_transitions = {
            "SUBMITTED": ["PENDING_REVIEW"],
            "PENDING_REVIEW": ["APPROVED", "REJECTED"],
            "APPROVED": ["PAID"],
            "REJECTED": ["SUBMITTED"],
            "PAID": []  # Final state
        }

        if new_state not in valid_transitions.get(self.state, []):
            raise ValueError(f"Invalid transition from {self.state} to {new_state}")

        heptora.log.info(f"Transitioning from {self.state} to {new_state}: {reason}")
        self.state = new_state
        self.invoice["state"] = new_state
        self.invoice["last_transition"] = datetime.now()

        return self.state

Advantages:

• Clear business logic
• Prevents invalid transitions
• Audit trail of state changes
• Easy to visualize process flow

Pattern 4: The Saga Pattern

For distributed transactions that need to be coordinated:

Example: Order Fulfillment Saga

Steps:
1. Reserve inventory
2. Process payment
3. Create shipment
4. Send confirmation

If any step fails, compensate previous steps:
- Cancel shipment → Refund payment → Release inventory

Implementation:
def order_fulfillment_saga(order):
    compensations = []

    try:
        # Step 1: Reserve inventory
        heptora.log.info("Reserving inventory")
        reservation = reserve_inventory(order)
        compensations.append(("inventory", lambda: release_inventory(reservation)))

        # Step 2: Process payment
        heptora.log.info("Processing payment")
        payment = process_payment(order)
        compensations.append(("payment", lambda: refund_payment(payment)))

        # Step 3: Create shipment
        heptora.log.info("Creating shipment")
        shipment = create_shipment(order)
        compensations.append(("shipment", lambda: cancel_shipment(shipment)))

        # Step 4: Send confirmation
        heptora.log.info("Sending confirmation")
        send_confirmation(order)

        heptora.log.result("ok", f"Order {order['id']} fulfilled successfully")
        return True

    except Exception as e:
        # Compensate in reverse order
        heptora.log.info(f"Error: {str(e)}. Executing compensations...")

        for step_name, compensate_fn in reversed(compensations):
            try:
                heptora.log.info(f"Compensating {step_name}")
                compensate_fn()
            except Exception as comp_error:
                heptora.log.info(f"Compensation failed for {step_name}: {str(comp_error)}")

        heptora.log.result("error", f"Order {order['id']} failed: {str(e)}")
        return False

Advantages:

• Maintains consistency across systems
• Handles partial failures gracefully
• Clear compensation logic
• Suitable for distributed systems

Pattern 5: The Template Method Pattern

Define the skeleton of an algorithm, letting subclasses override specific steps:

Base template for all data processing:

def process_data_template():
    """Template method for data processing."""

    # Step 1: Extract (different for each source)
    raw_data = extract_data()
    heptora.log.info(f"Extracted {len(raw_data)} records")

    # Step 2: Transform (common logic)
    transformed_data = transform_data(raw_data)
    heptora.log.info(f"Transformed {len(transformed_data)} records")

    # Step 3: Validate (common logic)
    valid_data, invalid_data = validate_data(transformed_data)
    heptora.log.info(f"Valid: {len(valid_data)}, Invalid: {len(invalid_data)}")

    # Step 4: Load (different for each destination)
    result = load_data(valid_data)
    heptora.log.info(f"Loaded {len(result)} records")

    # Step 5: Report (common logic)
    generate_report(valid_data, invalid_data, result)

    return result

# Specific implementations:

def process_excel_to_database():
    """Excel to Database ETL."""
    def extract_data():
        return heptora.excel.load("data.xlsx", "Sheet1", COLUMNS, True)

    def load_data(data):
        return insert_to_database(data)

    return process_data_template()

def process_api_to_excel():
    """API to Excel ETL."""
    def extract_data():
        return fetch_from_api()

    def load_data(data):
        heptora.excel.write(data, "output.xlsx", "Data")
        return data

    return process_data_template()

Advantages:

• Code reuse for common logic
• Consistent process structure
• Easy to create new variants
• Enforces standard patterns

Modularity and Maintainability

The Principle of Modularity

Modular processes are:

• Composable: Build complex workflows from simple pieces
• Maintainable: Update one module without affecting others
• Testable: Test modules independently
• Understandable: Each module has clear purpose
• Reusable: Use same module in multiple processes

Creating Modular Processes in Heptora

Decomposition strategies:

❌ Monolithic Process: “Complete Customer Onboarding”

1. Receive customer application (email or web form)
2. Extract customer data
3. Validate identity documents
4. Check credit score
5. Verify employment
6. Calculate risk score
7. Generate contract
8. Send for e-signature
9. Wait for signature
10. Create customer account in CRM
11. Create customer account in billing system
12. Set up customer portal access
13. Send welcome email with credentials
14. Schedule onboarding call
15. Update sales dashboard

Problems:

• 15 steps in one process
• Mixed concerns (validation, processing, communication)
• Hard to test
• Changes to one part risk breaking others
• Can’t reuse parts

✅ Modular Design: Customer Onboarding Suite

Module 1: "Customer Data Collection"
- Receive application
- Extract and normalize data
→ Output: customer_data

Module 2: "Customer Verification"
- Validate identity documents
- Check credit score
- Verify employment
→ Output: verification_results

Module 3: "Risk Assessment"
- Calculate risk score
- Determine approval status
→ Output: risk_assessment

Module 4: "Contract Generation"
- Generate contract
- Send for e-signature
- Wait for completion
→ Output: signed_contract

Module 5: "Account Setup"
- Create CRM account
- Create billing account
- Set up portal access
→ Output: account_details

Module 6: "Customer Onboarding"
- Send welcome email
- Schedule onboarding call
- Update dashboards
→ Output: onboarding_status

Orchestrator: "Customer Onboarding Pipeline"
- Calls modules in sequence
- Passes data between modules
- Handles errors
- Coordinates the workflow

Benefits:

• Each module is simple and focused
• Modules can be tested independently
• Easy to update one module
• Can reuse modules (e.g., “Customer Verification” in other processes)
• Clear data flow
• Easier to parallelize when possible

Module Design Guidelines

Module size:

• ✅ 5-20 steps per module (sweet spot)
• ⚠️ 20-50 steps (consider breaking down)
• ❌ 50+ steps (definitely break down)

Module cohesion:

• High cohesion: Module performs related operations
• Low cohesion: Module does unrelated things

✅ High cohesion: "Customer Validation Module"
- Validate email format
- Validate phone format
- Validate address
- Check for duplicates
All related to validation

❌ Low cohesion: "Mixed Operations Module"
- Validate email
- Send notification
- Update database
- Generate report
Unrelated operations

Module coupling:

• Loose coupling: Modules minimally dependent on each other
• Tight coupling: Modules highly dependent

✅ Loose coupling:
Module A → Outputs: customer_data (generic dictionary)
Module B → Inputs: customer_data (accepts generic dictionary)
Modules communicate through well-defined interfaces

❌ Tight coupling:
Module A → Directly modifies Module B's internal variables
Module B → Relies on specific implementation details of Module A
Changes to one module break the other

Organizing Process Code

File structure for maintainability:

/processes
  /customer_onboarding
    orchestrator.py          # Main workflow
    /modules
      data_collection.py     # Customer data collection
      verification.py        # Customer verification
      risk_assessment.py     # Risk assessment
      contract_generation.py # Contract generation
      account_setup.py       # Account setup
      onboarding.py          # Onboarding tasks
    /helpers
      validators.py          # Validation functions
      formatters.py          # Data formatting
      api_clients.py         # API integration
    /config
      settings.py            # Configuration
      secrets.py             # Secret references (not values!)
    /tests
      test_verification.py   # Unit tests
      test_risk.py           # Unit tests
    README.md                # Process documentation

Module template:

"""
Module: Customer Verification
Purpose: Validates customer identity and creditworthiness
Author: RPA Team
Last Updated: 2024-01-15
"""

import heptora
from typing import Dict, Tuple

# Module configuration
MODULE_NAME = "Customer Verification"
REQUIRED_INPUTS = ["customer_data"]
OUTPUTS = ["verification_results"]

def verify_customer(customer_data: Dict) -> Tuple[bool, Dict]:
    """
    Verifies customer identity and creditworthiness.

    Args:
        customer_data: Dictionary containing customer information

    Returns:
        Tuple of (success, verification_results)

    Raises:
        ValueError: If required customer data is missing
    """
    heptora.log.info(f"=== Starting {MODULE_NAME} ===")

    try:
        # Validate inputs
        _validate_inputs(customer_data)

        # Perform verification steps
        identity_verified = _verify_identity(customer_data)
        credit_score = _check_credit(customer_data)
        employment_verified = _verify_employment(customer_data)

        # Compile results
        results = {
            "identity_verified": identity_verified,
            "credit_score": credit_score,
            "employment_verified": employment_verified,
            "overall_status": _calculate_overall_status(
                identity_verified, credit_score, employment_verified
            ),
            "timestamp": datetime.now().isoformat()
        }

        heptora.log.info(f"=== Completed {MODULE_NAME} ===")
        return True, results

    except Exception as e:
        heptora.log.result("error", f"{MODULE_NAME} failed: {str(e)}")
        return False, {}

def _validate_inputs(customer_data: Dict) -> None:
    """Validates that required customer data is present."""
    required_fields = ["name", "email", "tax_id", "date_of_birth"]

    for field in required_fields:
        if field not in customer_data:
            raise ValueError(f"Missing required field: {field}")

def _verify_identity(customer_data: Dict) -> bool:
    """Verifies customer identity documents."""
    # Implementation details...
    pass

def _check_credit(customer_data: Dict) -> int:
    """Checks customer credit score."""
    # Implementation details...
    pass

def _verify_employment(customer_data: Dict) -> bool:
    """Verifies customer employment status."""
    # Implementation details...
    pass

def _calculate_overall_status(identity, credit, employment) -> str:
    """Calculates overall verification status."""
    if identity and credit >= 650 and employment:
        return "APPROVED"
    elif not identity or credit < 500:
        return "REJECTED"
    else:
        return "MANUAL_REVIEW"

Version Control Best Practices

Use version control (Git) for all process code:

Branching strategy:

main (production)
├── develop (integration)
├── feature/invoice-validation (new feature)
├── fix/excel-parsing-error (bug fix)
└── hotfix/critical-portal-update (urgent fix)

Commit messages:

✅ Good commit messages:
- "Add retry logic to FACe submission module"
- "Fix: Handle missing customer email gracefully"
- "Refactor: Extract validation into separate module"
- "Docs: Update README with new environment variables"

❌ Poor commit messages:
- "Updates"
- "Fix"
- "Changes"
- "Test"

What to version:

• ✅ All process code (.py files)
• ✅ Configuration files (non-sensitive)
• ✅ Documentation (README, guides)
• ✅ Test files
• ✅ Requirements (dependencies)
• ❌ Secrets or credentials
• ❌ Generated reports
• ❌ Temporary files
• ❌ Large data files

Documentation Best Practices

Good documentation is essential for long-term success.

What to Document

Process-level documentation:

# Invoice Processing Automation

## Purpose
Automates submission of invoices to the FACe electronic invoicing portal.

## Business Value
- Saves 20 hours/week of manual work
- Eliminates data entry errors
- Reduces invoice processing time from days to hours
- Ensures timely invoice submission

## Process Owner
Finance Department - Maria García (maria.garcia@company.com)

## Technical Contact
RPA Team - Carlos López (carlos.lopez@company.com)

## Inputs
- Excel file with invoices (pending_invoices.xlsx)
- FACe portal credentials (stored in Secrets)

## Outputs
- Submission confirmations (PDF)
- Processing report (Excel)
- Email notification to finance team

## Frequency
- Automated: Hourly during business hours (8 AM - 6 PM)
- Manual: On-demand via Heptora dashboard

## Dependencies
- FACe portal availability
- Corporate network access
- Digital certificate validity

## Error Handling
- KO: Invalid invoice data → Quarantine for manual review
- ERROR: Portal unavailable → Retry in 15 minutes, alert if persistent

## Success Criteria
- 95%+ success rate
- All valid invoices submitted within 2 hours
- Zero data entry errors

## Known Issues
- Portal maintenance on first Monday of month (6-7 AM)
- SSL certificate renewal every 2 years

## Change Log
- 2024-01-15: Initial version
- 2024-02-01: Added retry logic for transient failures
- 2024-03-10: Implemented quarantine for bad data

Module-level documentation:

Use docstrings consistently:

def process_invoice(invoice: Dict[str, Any]) -> bool:
    """
    Processes a single invoice submission to FACe portal.

    This function handles the complete lifecycle of an invoice submission:
    1. Validates invoice data
    2. Logs into FACe portal
    3. Fills and submits the invoice form
    4. Waits for confirmation
    5. Downloads confirmation document

    Args:
        invoice: Dictionary containing invoice data with keys:
            - number (str): Invoice number (required)
            - date (str): Invoice date in YYYY-MM-DD format (required)
            - amount (float): Total amount (required)
            - customer_tax_id (str): Customer tax ID (required)
            - items (list): List of invoice line items (optional)

    Returns:
        bool: True if invoice submitted successfully, False otherwise

    Raises:
        ValueError: If required invoice fields are missing or invalid
        ConnectionError: If unable to connect to FACe portal
        TimeoutError: If portal operation times out

    Example:
        >>> invoice = {
        ...     "number": "INV-2024-001",
        ...     "date": "2024-01-15",
        ...     "amount": 1500.00,
        ...     "customer_tax_id": "B12345678"
        ... }
        >>> success = process_invoice(invoice)
        >>> print(success)
        True

    Notes:
        - This function uses browser automation via Playwright
        - Average execution time: 45-60 seconds per invoice
        - Automatically retries once if timeout occurs
        - Results logged via heptora.log.result()
    """
    # Implementation...

Inline comments for complex logic:

# Calculate tax based on customer location and product type
# EU customers: VAT applied based on customer's country
# Non-EU customers: No VAT, but import tax may apply
if customer["country"] in EU_COUNTRIES:
    # B2B transactions in EU: reverse charge
    if customer["has_valid_vat_number"]:
        tax_rate = 0  # Reverse charge
        tax_note = "Reverse charge - Art 196 EU VAT Directive"
    else:
        # B2C transactions: apply destination country VAT
        tax_rate = VAT_RATES[customer["country"]]
        tax_note = f"VAT {tax_rate}% ({customer['country']})"
else:
    # Non-EU: no VAT
    tax_rate = 0
    tax_note = "Export - VAT exempt"

tax_amount = base_amount * tax_rate

Documentation Tools

README.md template:

# [Process Name]

Brief description of what this process does and why it exists.

## Quick Start

```bash
# Prerequisites
- Heptora Robot installed and running
- Access to [system names]
- Required secrets configured

# Run the process
python main_process.py

Table of Contents

• Overview
• Business Value
• Architecture
• Configuration
• Usage
• Troubleshooting
• Development
• FAQ

Overview

Detailed description…

Business Value

• Metric 1: X hours saved per week
• Metric 2: Y% error reduction
• Metric 3: Z% faster processing

Architecture

[ASCII diagram or link to architecture diagram]

Configuration

Environment Variables

Variable	Description	Example	Required
PORTAL_URL	FACe portal URL	https://face.gob.es	Yes
RETRY_ATTEMPTS	Max retry attempts	3	No (default: 3)

Secrets

Configure these in Heptora Secrets Management:

• face_username: FACe portal username
• face_password: FACe portal password

Usage

Running Manually

…

Scheduled Execution

…

Monitoring

…

Troubleshooting

Common Issues

Issue: “Connection timeout”

• Cause: Portal is slow or down
• Solution: Retry later or increase timeout value

Development

Local Development

…

Testing

pytest tests/

Contributing

…

FAQ

Q: How long does the process take? A: Approximately 2-3 minutes per invoice…

## Testing Strategies

Testing is crucial for reliable automation.

### Types of Tests

**1. Unit Tests**

Test individual functions in isolation:

```python
import pytest
from invoice_validation import validate_invoice

def test_validate_invoice_success():
    """Test valid invoice passes validation."""
    invoice = {
        "number": "INV-001",
        "amount": 1000,
        "date": "2024-01-15",
        "tax_id": "B12345678"
    }

    is_valid, error = validate_invoice(invoice)

    assert is_valid == True
    assert error == ""

def test_validate_invoice_missing_number():
    """Test invoice without number fails validation."""
    invoice = {
        "amount": 1000,
        "date": "2024-01-15",
        "tax_id": "B12345678"
    }

    is_valid, error = validate_invoice(invoice)

    assert is_valid == False
    assert "number" in error.lower()

def test_validate_invoice_invalid_amount():
    """Test invoice with zero amount fails validation."""
    invoice = {
        "number": "INV-001",
        "amount": 0,
        "date": "2024-01-15",
        "tax_id": "B12345678"
    }

    is_valid, error = validate_invoice(invoice)

    assert is_valid == False
    assert "amount" in error.lower()

2. Integration Tests

Test modules working together:

def test_end_to_end_invoice_processing():
    """Test complete invoice processing flow."""
    # Setup
    test_invoice = create_test_invoice()

    # Execute process
    result = process_invoice_pipeline(test_invoice)

    # Verify
    assert result["status"] == "success"
    assert result["confirmation_number"] is not None
    assert result["pdf_path"].exists()

def test_error_handling_invalid_data():
    """Test process handles invalid data gracefully."""
    invalid_invoice = {"invalid": "data"}

    result = process_invoice_pipeline(invalid_invoice)

    assert result["status"] == "ko"
    assert "validation" in result["error"].lower()

3. UI Tests

Test browser automation:

def test_face_portal_login():
    """Test successful login to FACe portal."""
    page = heptora.browser.connect()

    # Navigate to portal
    page.goto("https://face.gob.es")

    # Login
    login_success = login_to_face(page, TEST_USER, TEST_PASSWORD)

    # Verify
    assert login_success == True
    assert page.url.endswith("/dashboard")
    assert page.locator(".user-name").text_content() == TEST_USER

4. Data Tests

Test with various data scenarios:

@pytest.mark.parametrize("amount,expected", [
    (100, True),      # Normal amount
    (0, False),       # Zero amount
    (-100, False),    # Negative amount
    (999999, True),   # Large amount
    (0.01, True),     # Small amount
])
def test_amount_validation(amount, expected):
    """Test amount validation with various values."""
    invoice = {"number": "INV-001", "amount": amount}
    is_valid, _ = validate_invoice(invoice)
    assert is_valid == expected

Test Environments

Maintain separate environments:

Development:
- Local machine
- Test data
- Mock external services
- Fast execution

Staging:
- Mirrors production
- Real data (anonymized)
- Real external services (test instances)
- Full integration testing

Production:
- Live environment
- Real data
- Real external services
- Monitored execution

Test Data Management

# test_data.py - Centralized test data

VALID_INVOICE = {
    "number": "TEST-INV-001",
    "date": "2024-01-15",
    "amount": 1000.00,
    "tax_id": "B12345678",
    "customer_name": "Test Customer Ltd"
}

INVALID_INVOICES = {
    "missing_number": {
        "date": "2024-01-15",
        "amount": 1000.00
    },
    "invalid_amount": {
        "number": "TEST-INV-002",
        "amount": -100
    },
    "invalid_date": {
        "number": "TEST-INV-003",
        "date": "invalid-date"
    }
}

EDGE_CASES = {
    "max_amount": {
        "number": "TEST-INV-004",
        "amount": 999999.99
    },
    "special_characters": {
        "number": "TEST-INV-005",
        "customer_name": "Ñoño & Associates <Test>"
    }
}

Version Control and Change Management

Semantic Versioning

Use semantic versioning for processes:

Version format: MAJOR.MINOR.PATCH

Examples:
- 1.0.0: Initial production release
- 1.0.1: Bug fix (portal timeout handling improved)
- 1.1.0: New feature (added PDF attachment support)
- 2.0.0: Breaking change (changed input file format)

Change Management Process

1. PROPOSE
   - Document proposed change
   - Explain business justification
   - Assess impact and risk

2. REVIEW
   - Technical review by team
   - Business review by stakeholders
   - Security review if needed

3. DEVELOP
   - Create feature branch
   - Implement changes
   - Write/update tests
   - Update documentation

4. TEST
   - Unit tests pass
   - Integration tests pass
   - UAT by business users
   - Performance testing

5. DEPLOY
   - Deploy to staging
   - Smoke test
   - Deploy to production
   - Monitor execution

6. VALIDATE
   - Verify working as expected
   - Monitor for issues
   - Gather feedback
   - Document lessons learned

Release Notes Template

# Release Notes - [Process Name] v1.2.0

**Release Date:** 2024-02-15
**Type:** Minor Release

## Summary

This release adds support for bulk invoice processing and improves error handling for network timeouts.

## New Features

- **Bulk Processing**: Now supports processing multiple invoices in a single execution
  - Processes up to 100 invoices per run
  - Generates consolidated report
  - Impact: 5x faster for high-volume scenarios

## Improvements

- Enhanced timeout handling with exponential backoff
- Better logging for debugging
- Updated to use latest Playwright version

## Bug Fixes

- Fixed: Excel parsing error when cells contain formulas
- Fixed: Incorrect date formatting for non-EU locales
- Fixed: Memory leak in long-running processes

## Breaking Changes

None

## Deprecations

- `process_single_invoice()` function deprecated, use `process_invoices([invoice])` instead
- Will be removed in v2.0.0

## Migration Guide

No changes required for existing implementations.

## Known Issues

- Portal maintenance window on first Monday of each month (6-7 AM)
- Large PDF attachments (>10MB) may timeout on slow connections

## Dependencies

- Heptora: 2.5.0 or higher
- Python: 3.8 or higher
- Playwright: 1.40.0

## Contributors

- Carlos López
- María García
- Juan Rodríguez

Performance Optimization

Identifying Performance Bottlenecks

Measure first:

import time

def measure_execution_time(func_name, func, *args, **kwargs):
    """Measures and logs execution time of a function."""
    start_time = time.time()
    result = func(*args, **kwargs)
    elapsed = time.time() - start_time

    heptora.log.info(f"{func_name} took {elapsed:.2f} seconds")
    return result

# Usage
invoices = measure_execution_time(
    "Load invoices",
    heptora.excel.load,
    "invoices.xlsx", "Sheet1", COLUMNS, True
)

results = measure_execution_time(
    "Process invoices",
    process_invoices,
    invoices
)

Optimization Techniques

1. Batch Processing

Process multiple items together:

❌ Slow: Process one at a time
for invoice in invoices:
    connect_to_database()
    insert_invoice(invoice)
    disconnect_from_database()
# 100 invoices × 3 seconds = 300 seconds

✅ Fast: Batch processing
connect_to_database()
for i in range(0, len(invoices), 10):
    batch = invoices[i:i+10]
    insert_batch(batch)
disconnect_from_database()
# 10 batches × 2 seconds = 20 seconds

2. Parallel Processing

For independent operations:

from concurrent.futures import ThreadPoolExecutor

def process_invoices_parallel(invoices, max_workers=5):
    """Process invoices in parallel."""
    results = []

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        # Submit all tasks
        futures = {
            executor.submit(process_invoice, inv): inv
            for inv in invoices
        }

        # Collect results
        for future in concurrent.futures.as_completed(futures):
            invoice = futures[future]
            try:
                result = future.result()
                results.append(result)
            except Exception as e:
                heptora.log.result("error", f"Invoice {invoice['number']} failed: {str(e)}")

    return results

3. Caching

Cache expensive operations:

from functools import lru_cache

@lru_cache(maxsize=128)
def get_customer_details(customer_id):
    """Get customer details with caching."""
    # Expensive API call or database query
    return fetch_customer_from_api(customer_id)

# First call: fetches from API (slow)
customer = get_customer_details("CUST-001")

# Subsequent calls: returns from cache (fast)
customer = get_customer_details("CUST-001")  # Instant

4. Lazy Loading

Load data only when needed:

❌ Load everything upfront
all_customers = load_all_customers()  # 10,000 records
for customer in all_customers:
    if customer["type"] == "VIP":
        process_vip_customer(customer)
# Loaded 10,000 but only used 100

✅ Load incrementally
for customer in iterate_customers_by_type("VIP"):
    process_vip_customer(customer)
# Only loaded 100 VIP customers

5. Optimize Selectors

For web automation, use efficient selectors:

❌ Slow: XPath traversal
page.locator("//div[@class='container']/div[2]/span/a[@class='submit']")

✅ Fast: ID or unique attribute
page.locator("#submit-button")

✅ Good: CSS selector
page.locator(".container .submit-button")

6. Connection Pooling

Reuse database connections:

# Connection pool (managed by Heptora)
with database_connection() as conn:
    for invoice in invoices:
        # Reuses same connection
        insert_invoice(conn, invoice)
# Connection returned to pool

❌ Don't do this:
for invoice in invoices:
    conn = create_new_connection()  # Expensive!
    insert_invoice(conn, invoice)
    close_connection(conn)

Performance Benchmarks

Set and monitor performance targets:

PERFORMANCE_TARGETS = {
    "load_data": 5,          # Max 5 seconds
    "validate_invoice": 0.1,  # Max 100ms per invoice
    "submit_to_portal": 30,   # Max 30 seconds
    "generate_report": 10,    # Max 10 seconds
}

def check_performance(operation, elapsed):
    """Check if operation meets performance target."""
    target = PERFORMANCE_TARGETS.get(operation)
    if target and elapsed > target:
        heptora.log.info(
            f"⚠️ Performance warning: {operation} took {elapsed:.2f}s "
            f"(target: {target}s)"
        )

Security Considerations

Protecting Sensitive Data

1. Secrets Management

Never hardcode credentials:

❌ NEVER do this:
USERNAME = "admin@company.com"
PASSWORD = "MyPassword123"
API_KEY = "sk-1234567890abcdef"

✅ Use Heptora Secrets:
username = heptora.data.get("portal_username")
password = heptora.data.get("portal_password")
api_key = heptora.data.get("api_key")

if not username or not password:
    heptora.log.force_ko("Missing credentials. Configure in Secrets.")

Heptora’s Secrets features:

• Stored locally, never in cloud
• Encrypted at rest
• Not visible in logs
• Per-robot configuration
• Easy rotation

2. Data Sanitization

Remove sensitive data from logs:

def sanitize_for_logging(data):
    """Remove sensitive fields before logging."""
    sanitized = data.copy()

    sensitive_fields = [
        "password", "credit_card", "ssn",
        "tax_id", "api_key", "token"
    ]

    for field in sensitive_fields:
        if field in sanitized:
            sanitized[field] = "***REDACTED***"

    return sanitized

# Usage
heptora.log.info(f"Processing customer: {sanitize_for_logging(customer)}")

3. Secure File Handling

import os
from pathlib import Path

# Use secure temp directories
import tempfile
with tempfile.TemporaryDirectory() as temp_dir:
    sensitive_file = Path(temp_dir) / "data.xlsx"
    # Process file
    # Automatically deleted when context exits

# Set proper file permissions
os.chmod("confidential.pdf", 0o600)  # Read/write for owner only

# Securely delete files
import shutil
def secure_delete(file_path):
    """Securely delete a file."""
    # Overwrite with random data before deleting
    with open(file_path, "wb") as f:
        f.write(os.urandom(os.path.getsize(file_path)))
    os.remove(file_path)

4. Input Validation

Always validate and sanitize inputs:

def validate_and_sanitize_input(user_input):
    """Validate and sanitize user input."""
    # Remove dangerous characters
    sanitized = user_input.strip()
    sanitized = sanitized.replace(";", "")
    sanitized = sanitized.replace("--", "")

    # Validate format
    if not re.match(r'^[A-Za-z0-9\s\-\.]+$', sanitized):
        raise ValueError("Input contains invalid characters")

    # Validate length
    if len(sanitized) > 255:
        raise ValueError("Input too long")

    return sanitized

# Usage
try:
    safe_input = validate_and_sanitize_input(user_provided_value)
except ValueError as e:
    heptora.log.force_ko(f"Invalid input: {str(e)}")

5. SQL Injection Prevention

Use parameterized queries:

❌ NEVER do this (SQL injection risk):
query = f"SELECT * FROM customers WHERE email = '{user_email}'"
results = execute_query(query)

✅ Use parameterized queries:
query = "SELECT * FROM customers WHERE email = ?"
results = execute_query(query, (user_email,))

6. API Security

# Use HTTPS only
❌ api_url = "http://api.company.com"
✅ api_url = "https://api.company.com"

# Validate SSL certificates
import requests
response = requests.get(api_url, verify=True)  # Verify SSL cert

# Set timeouts to prevent hanging
response = requests.get(api_url, timeout=30)

# Rate limiting
from time import sleep
def call_api_with_rate_limit(endpoint, max_calls=10, period=60):
    """Call API with rate limiting."""
    calls = []

    # Remove old calls outside the period
    now = time.time()
    calls = [c for c in calls if now - c < period]

    if len(calls) >= max_calls:
        wait_time = period - (now - calls[0])
        heptora.log.info(f"Rate limit reached. Waiting {wait_time:.0f}s")
        sleep(wait_time)

    response = requests.get(endpoint)
    calls.append(time.time())
    return response

Access Control

Principle of Least Privilege:

Process: "Invoice Viewer"
Permissions:
✅ Read invoices from database
✅ Read customer data
❌ Write to database
❌ Delete records
❌ Access employee data

Process: "Invoice Processor"
Permissions:
✅ Read invoices
✅ Write invoice status
✅ Upload to portal
❌ Delete invoices
❌ Modify customer records

Audit Logging

Track who did what and when:

def audit_log(action, resource, details=None):
    """Create audit log entry."""
    entry = {
        "timestamp": datetime.now().isoformat(),
        "process": "Invoice Processing",
        "action": action,
        "resource": resource,
        "details": details,
        "user": heptora.data.get("current_user"),
        "robot_id": heptora.data.get("robot_id")
    }

    # Log to audit system
    log_to_audit_system(entry)

    # Also log in Heptora
    heptora.log.info(f"AUDIT: {action} on {resource}")

# Usage
audit_log("submit", f"invoice_{invoice_number}", {"amount": amount})
audit_log("delete", f"record_{record_id}", {"reason": "duplicate"})

Team Collaboration

Roles and Responsibilities

Clear role definition:

Process Owner (Business):
- Defines requirements
- Provides test cases
- Validates results
- Approves changes
- Escalates issues

Developer:
- Implements process
- Writes tests
- Documents code
- Fixes bugs
- Optimizes performance

Tester:
- Creates test plans
- Executes tests
- Reports defects
- Validates fixes

Operations:
- Monitors execution
- Manages infrastructure
- Handles incidents
- Maintains environments

Communication Practices

1. Stand-up Meetings

Daily 15-minute sync:

• What did I complete yesterday?
• What will I work on today?
• What blockers do I have?

2. Code Reviews

All changes reviewed before merge:

Code Review Checklist:
□ Code follows style guidelines
□ Logic is clear and well-documented
□ Error handling is comprehensive
□ Tests are included and passing
□ Documentation is updated
□ No hardcoded credentials
□ Performance is acceptable
□ Security considerations addressed

3. Knowledge Sharing

• Weekly tech talks
• Documentation wiki
• Recorded demos
• Pair programming sessions

Collaboration Tools in Heptora

1. Process Sharing

Export and share process configurations:

Export process → Share with team → Import on other robots

2. Centralized Logging

All team members can view execution logs

3. Templates Library

Shared library of reusable components

4. Version History

Track who changed what and when

Practical Checklist: Process Design Review

Before deploying a process to production, review against this checklist:

Design Quality

• Process has clear, single responsibility
• Broken into modules of 5-20 steps each
• Modules are loosely coupled
• Reusable components extracted
• No “God blocks” (overly complex modules)
• Clear data flow between modules
• Follows established design patterns

Code Quality

• Code is readable and well-formatted
• Meaningful variable and function names
• Comprehensive error handling
• No hardcoded values (use configuration)
• No duplicate code
• Helper functions for repeated logic
• Type hints for Python functions

Testing

• Unit tests written and passing
• Integration tests written and passing
• Tested with various data scenarios
• Tested error paths
• Performance tested under load
• Tested in staging environment

Documentation

• README with overview and setup
• All functions have docstrings
• Complex logic has inline comments
• Architecture diagram included
• Configuration documented
• Troubleshooting guide provided

Security

• No hardcoded credentials
• Secrets stored in Heptora Secrets
• Input validation implemented
• SQL injection prevention
• Sensitive data sanitized in logs
• Audit logging in place

Error Handling

• Distinguishes KO from ERROR
• Comprehensive exception handling
• Retry logic for transient failures
• Graceful degradation
• Meaningful error messages
• Rollback capability where needed

Performance

• Execution time is acceptable
• Batch processing used where appropriate
• Database connections pooled
• Caching implemented for expensive operations
• No unnecessary waits or sleeps
• Performance targets defined and met

Operations

• Monitoring and alerting configured
• Log levels appropriate
• Process runs unattended
• Resource usage is reasonable
• Scheduled execution configured correctly
• Notifications set up properly

Maintenance

• Code in version control
• Change log maintained
• Version number assigned
• Dependencies documented
• Process owner identified
• Runbook created

Real-World Example: Complete Process

Let’s see all these practices in action with a complete, production-ready example.

Scenario: Automated Invoice Processing

Business requirement: Process invoices from Excel, validate data, submit to FACe portal, generate reports.

Following best practices:

"""
FACe Invoice Processing Automation
Version: 1.2.0
Author: RPA Team
Last Updated: 2024-02-15

This process automates the submission of invoices to the FACe electronic
invoicing portal, reducing manual processing time from 20 hours/week to 1 hour/week.
"""

import heptora
from typing import List, Dict, Tuple
from datetime import datetime
import json

# =============================================================================
# CONFIGURATION
# =============================================================================

CONFIG = {
    "input_file": "pending_invoices.xlsx",
    "input_sheet": "Invoices",
    "portal_url": "https://face.gob.es",
    "batch_size": 10,
    "max_retries": 3,
    "retry_delay": 5,
}

COLUMNS = ["InvoiceNumber", "Date", "Amount", "CustomerTaxId", "CustomerName"]

# =============================================================================
# MODULE 1: DATA ACQUISITION
# =============================================================================

def load_invoices() -> List[Dict]:
    """
    Load invoices from Excel file.

    Returns:
        List of invoice dictionaries
    """
    heptora.log.info("=== Module 1: Data Acquisition ===")

    try:
        invoices = heptora.excel.load(
            CONFIG["input_file"],
            CONFIG["input_sheet"],
            COLUMNS,
            has_header=True
        )

        heptora.log.info(f"Loaded {len(invoices)} invoices")
        return invoices

    except FileNotFoundError:
        heptora.log.force_ko(f"Input file not found: {CONFIG['input_file']}")
    except Exception as e:
        heptora.log.result("error", f"Failed to load invoices: {str(e)}")
        return []

# =============================================================================
# MODULE 2: DATA VALIDATION
# =============================================================================

def validate_invoice(invoice: Dict) -> Tuple[bool, str]:
    """
    Validate invoice data according to business rules.

    Args:
        invoice: Invoice data dictionary

    Returns:
        Tuple of (is_valid, error_message)
    """
    # Required fields
    if not invoice.get("InvoiceNumber"):
        return False, "Missing invoice number"

    if not invoice.get("CustomerTaxId"):
        return False, "Missing customer tax ID"

    # Amount validation
    try:
        amount = float(invoice.get("Amount", 0))
        if amount <= 0:
            return False, "Amount must be greater than 0"
        if amount > 1000000:
            return False, "Amount exceeds maximum allowed"
    except (ValueError, TypeError):
        return False, "Invalid amount format"

    # Tax ID format (simplified)
    tax_id = str(invoice.get("CustomerTaxId", ""))
    if len(tax_id) != 9:
        return False, f"Invalid tax ID format: {tax_id}"

    # Date validation
    date_str = invoice.get("Date")
    try:
        datetime.strptime(str(date_str), "%Y-%m-%d")
    except ValueError:
        return False, f"Invalid date format: {date_str}"

    return True, ""

def validate_invoices(invoices: List[Dict]) -> Tuple[List[Dict], List[Dict]]:
    """
    Validate all invoices and separate valid from invalid.

    Args:
        invoices: List of invoice dictionaries

    Returns:
        Tuple of (valid_invoices, invalid_invoices)
    """
    heptora.log.info("=== Module 2: Data Validation ===")

    valid = []
    invalid = []

    for invoice in invoices:
        is_valid, error = validate_invoice(invoice)

        if is_valid:
            valid.append(invoice)
        else:
            invoice["_validation_error"] = error
            invalid.append(invoice)
            heptora.log.result(
                "ko",
                f"Invoice {invoice.get('InvoiceNumber', 'UNKNOWN')}: {error}"
            )

    heptora.log.info(f"Valid: {len(valid)}, Invalid: {len(invalid)}")
    return valid, invalid

# =============================================================================
# MODULE 3: PORTAL SUBMISSION
# =============================================================================

def login_to_portal(page) -> bool:
    """
    Log into FACe portal.

    Args:
        page: Playwright page object

    Returns:
        True if login successful, False otherwise
    """
    try:
        heptora.log.info("Logging into FACe portal")

        # Get credentials from secrets
        username = heptora.data.get("face_username")
        password = heptora.data.get("face_password")

        if not username or not password:
            heptora.log.force_ko("Missing FACe credentials. Configure in Secrets.")
            return False

        # Navigate to login page
        page.goto(f"{CONFIG['portal_url']}/login", timeout=30000)
        page.wait_for_selector("#username", timeout=10000)

        # Enter credentials
        page.locator("#username").fill(username)
        page.locator("#password").fill(password)
        page.locator("button[type='submit']").click()

        # Verify login
        page.wait_for_selector(".dashboard", timeout=10000)

        heptora.log.info("✓ Login successful")
        return True

    except Exception as e:
        heptora.log.result("error", f"Login failed: {str(e)}")
        return False

def submit_invoice_to_portal(page, invoice: Dict) -> bool:
    """
    Submit a single invoice to the portal.

    Args:
        page: Playwright page object
        invoice: Invoice data dictionary

    Returns:
        True if successful, False otherwise
    """
    number = invoice["InvoiceNumber"]

    try:
        heptora.log.info(f"Submitting invoice {number}")

        # Navigate to submission form
        page.goto(f"{CONFIG['portal_url']}/new-invoice", timeout=30000)
        page.wait_for_selector("#invoice-form", timeout=10000)

        # Fill form
        page.locator("#invoice-number").fill(str(invoice["InvoiceNumber"]))
        page.locator("#date").fill(str(invoice["Date"]))
        page.locator("#amount").fill(str(invoice["Amount"]))
        page.locator("#customer-tax-id").fill(str(invoice["CustomerTaxId"]))
        page.locator("#customer-name").fill(str(invoice["CustomerName"]))

        # Submit
        page.locator("button#submit").click()

        # Wait for confirmation
        page.wait_for_selector(".success-message", timeout=10000)
        confirmation = page.locator(".confirmation-number").text_content()

        invoice["_confirmation"] = confirmation
        heptora.log.result("ok", f"Invoice {number} submitted: {confirmation}")
        return True

    except TimeoutError:
        heptora.log.result("error", f"Invoice {number}: Timeout waiting for portal")
        return False
    except Exception as e:
        heptora.log.result("error", f"Invoice {number}: {str(e)}")
        return False

def submit_invoices_with_retry(
    page,
    invoices: List[Dict]
) -> Tuple[List[Dict], List[Dict]]:
    """
    Submit invoices with retry logic.

    Args:
        page: Playwright page object
        invoices: List of valid invoices

    Returns:
        Tuple of (successful, failed)
    """
    heptora.log.info("=== Module 3: Portal Submission ===")

    successful = []
    failed = []

    for invoice in invoices:
        number = invoice["InvoiceNumber"]

        # Try with retries
        for attempt in range(1, CONFIG["max_retries"] + 1):
            try:
                if submit_invoice_to_portal(page, invoice):
                    successful.append(invoice)
                    break
                else:
                    if attempt < CONFIG["max_retries"]:
                        heptora.log.info(
                            f"Retry {attempt}/{CONFIG['max_retries']} "
                            f"for invoice {number}"
                        )
                        heptora.time.pause(CONFIG["retry_delay"])
                    else:
                        failed.append(invoice)
            except Exception as e:
                if attempt < CONFIG["max_retries"]:
                    heptora.log.info(f"Retrying after error: {str(e)}")
                    heptora.time.pause(CONFIG["retry_delay"])
                else:
                    failed.append(invoice)
                    heptora.log.result("error", f"Invoice {number} failed after {CONFIG['max_retries']} attempts")

    heptora.log.info(f"Submitted: {len(successful)}, Failed: {len(failed)}")
    return successful, failed

# =============================================================================
# MODULE 4: REPORTING
# =============================================================================

def generate_reports(
    valid: List[Dict],
    invalid: List[Dict],
    successful: List[Dict],
    failed: List[Dict]
) -> None:
    """
    Generate summary and detailed reports.

    Args:
        valid: Valid invoices
        invalid: Invalid invoices
        successful: Successfully submitted invoices
        failed: Failed submissions
    """
    heptora.log.info("=== Module 4: Reporting ===")

    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")

    # Summary report
    summary = [{
        "Timestamp": datetime.now().isoformat(),
        "Total_Invoices": len(valid) + len(invalid),
        "Valid_Invoices": len(valid),
        "Invalid_Invoices": len(invalid),
        "Successfully_Submitted": len(successful),
        "Failed_Submissions": len(failed),
        "Success_Rate": f"{(len(successful) / len(valid) * 100):.1f}%" if valid else "N/A"
    }]

    summary_file = f"summary_{timestamp}.xlsx"
    heptora.excel.write(summary, summary_file, "Summary")
    heptora.report.add_attachment(summary_file)
    heptora.log.info(f"✓ Summary report: {summary_file}")

    # Invalid invoices report
    if invalid:
        invalid_file = f"invalid_invoices_{timestamp}.xlsx"
        heptora.excel.write(invalid, invalid_file, "Invalid")
        heptora.report.add_attachment(invalid_file)
        heptora.log.info(f"✓ Invalid invoices report: {invalid_file}")

    # Failed submissions report
    if failed:
        failed_file = f"failed_submissions_{timestamp}.xlsx"
        heptora.excel.write(failed, failed_file, "Failed")
        heptora.report.add_attachment(failed_file)
        heptora.log.info(f"✓ Failed submissions report: {failed_file}")

    # Successful submissions report
    if successful:
        success_file = f"successful_submissions_{timestamp}.xlsx"
        heptora.excel.write(successful, success_file, "Success")
        heptora.report.add_attachment(success_file)
        heptora.log.info(f"✓ Successful submissions report: {success_file}")

# =============================================================================
# MAIN ORCHESTRATOR
# =============================================================================

def main():
    """
    Main orchestration function.
    Coordinates all modules to complete the invoice processing workflow.
    """
    heptora.log.info("=" * 70)
    heptora.log.info("FACe Invoice Processing Automation v1.2.0")
    heptora.log.info("=" * 70)

    start_time = datetime.now()

    try:
        # Module 1: Load invoices
        invoices = load_invoices()
        if not invoices:
            heptora.log.force_ko("No invoices to process")
            return

        # Module 2: Validate invoices
        valid, invalid = validate_invoices(invoices)
        if not valid:
            heptora.log.force_ko("No valid invoices to submit")
            generate_reports(valid, invalid, [], [])
            return

        # Module 3: Connect to portal and submit
        page = heptora.browser.connect()

        if not login_to_portal(page):
            heptora.log.force_ko("Failed to login to portal")
            return

        successful, failed = submit_invoices_with_retry(page, valid)

        # Module 4: Generate reports
        generate_reports(valid, invalid, successful, failed)

        # Final summary
        elapsed = (datetime.now() - start_time).total_seconds()

        heptora.log.info("=" * 70)
        heptora.log.info("PROCESS COMPLETE")
        heptora.log.info(f"Total time: {elapsed:.1f} seconds")
        heptora.log.info(f"Total invoices: {len(invoices)}")
        heptora.log.info(f"  - Valid: {len(valid)}")
        heptora.log.info(f"  - Invalid (KO): {len(invalid)}")
        heptora.log.info(f"  - Successfully submitted: {len(successful)}")
        heptora.log.info(f"  - Failed (ERROR): {len(failed)}")
        heptora.log.info(f"Success rate: {(len(successful) / len(valid) * 100):.1f}%")
        heptora.log.info("=" * 70)

    except Exception as e:
        heptora.log.result("error", f"Fatal error in main process: {str(e)}")
        raise

# =============================================================================
# ENTRY POINT
# =============================================================================

if __name__ == "__main__":
    try:
        main()
    except Exception as e:
        heptora.log.result("error", f"Process failed: {str(e)}")

What this example demonstrates:

✅ Simplicity: Each module does one thing clearly ✅ Reusability: Functions can be used in other processes ✅ Exception handling: Comprehensive error handling with KO vs ERROR distinction ✅ Modularity: Clear separation of concerns ✅ Documentation: Every function documented ✅ Logging: Informative logs at each step ✅ Retry logic: Handles transient failures ✅ Reporting: Generates actionable reports ✅ Security: Uses secrets for credentials ✅ Performance: Efficient execution ✅ Maintainability: Clean, well-organized code

Related Resources

• Selecting Processes for Automation - How to choose which processes to automate (coming soon)
• Automation Blocks - Using pre-built automation components
• Process Templates - Leveraging production-ready templates
• Secrets Management - Protecting sensitive credentials
• Python in Heptora - Complete Python API reference
• Execution Results - Understanding OK, KO, and ERROR states
• Process Recorder - Creating automations by recording your actions

Need more help?

If this guide didn’t solve your problem or you found an error in the documentation:

• Technical support: help@heptora.com
• Clearly describe the problem you encountered
• Include screenshots if possible
• Indicate which documentation steps you followed

Our support team will help you resolve any issue.