Basic S3 Transformation

import lume
import boto3

# --- 1. Prepare Your Data ---
# Create a dummy local CSV file for demonstration.
with open("local_sales_data.csv", "w") as f:
    f.write("order_id,product_id,quantity,price\\n")
    f.write("ord_001,prod_a,2,10.50\\n")
    f.write("ord_002,prod_b,1,25.00\\n")
    f.write("ord_003,prod_a,5,9.99\\n")

# In a real workflow, this data would be delivered to S3 by another process.
# Here, we'll upload our sample file to a specific S3 path.
s3 = boto3.client('s3')
source_bucket = "lume-demo-source-bucket"
source_path = "s3://lume-demo-source-bucket/daily_sales/2024-07-30/sales.csv"

# For demonstration, we upload a dummy file to the source_path.
# The S3 Connector in the Flow Version must have read access to this location.
s3.upload_file("local_sales_data.csv", source_bucket, "daily_sales/2024-07-30/sales.csv")
print(f"Uploaded sample data to {source_path}")

# --- 2. Trigger the Run ---
# The `source_path` directly tells Lume which object to process.
try:
    run = lume.run(
        flow_version="sales_transformer:v3",
        source_path=source_path
    )
    print(f"Successfully started run {run.id} for source: {source_path}")

except Exception as e:
    print(f"Failed to start run: {e}")
    # Add your error handling logic here
    exit()

# --- 3. Monitor the Run ---
# `run.wait()` blocks execution until the pipeline finishes.
print("Waiting for run to complete...")
run.wait()
print(f"Run {run.id} finished with status: {run.status}")

<Tip>
For production workflows, consider using [Webhooks](/pages/libraries/python-ent/production#asynchronous-processing-with-webhooks) instead of `run.wait()` for more efficient, event-driven monitoring.
</Tip>

# --- 4. Handle the Outcome ---
if run.status == "SUCCEEDED":
    print("Run completed successfully!")
    # You can now access metadata, including output locations and metrics.
    pipeline_metrics = run.metadata.get('pipeline', {})
    print(f"Pipeline duration: {pipeline_metrics.get('duration_seconds')}s")

elif run.status == "PARTIAL_FAILED":
    print("Run completed with some errors.")
    # You can now use the metadata to find and inspect the rejected rows.
    results = run.metadata.get('results', {})
    print(f"Rejected rows count: {results.get('rejected_rows')}")
    print(f"Rejects file location: {results.get('target_locations', {}).get('rejects')}")

else: # FAILED or CRASHED
    print(f"Run failed with status: {run.status}")
    error_info = run.metadata.get('errors', [{}])[0]
    print(f"Error Stage: {error_info.get('stage')}")
    print(f"Error Message: {error_info.get('message')}")