Merge pull request #9 from render-examples/rm/sdk-update

update to new sdk

Author: miwialex

README.md

@@ -6,7 +6,7 @@ A curated collection of production-ready workflow examples demonstrating various
 
 These examples demonstrate how to build robust, scalable workflows using Render's Python SDK. All examples follow best practices for production deployments and include comprehensive documentation.
 
-Render Workflows support both Python and TypeScript. This repo contains Python examples using the `render-sdk` package, deployed as Workflow services on Render.
+Render Workflows support both Python and TypeScript. This repo contains Python examples using the `render` package, deployed as Workflow services on Render.
 
 ## Examples
 
@@ -16,7 +16,7 @@ Render Workflows support both Python and TypeScript. This repo contains Python e
 
 | Example | Use Case | Key Patterns | Extra Dependencies |
 |---------|----------|--------------|-------------------|
-| [**Hello World**](./hello-world/) | Learn workflow basics with simple number processing | Task definition, subtask calling with `await`, basic orchestration | None |
+| [**Hello World**](./hello-world/) | Learn workflow basics with simple number processing | Task definition, subtask calling with `ctx.run`, basic orchestration | None |
 | [**ETL Job**](./etl-job/) | Process CSV data with validation and statistics | Subtasks, sequential processing, batch operations, data validation | None |
 | [**OpenAI Agent**](./openai-agent/) | AI customer support agent with tool calling | Tool calling, nested subtasks (3 levels deep), stateful workflows, dynamic orchestration | `openai` |
 | [**File Processing**](./file-processing/) | Batch process multiple file formats in parallel | Parallel execution with `asyncio.gather()`, multi-format handling, aggregation | None |
@@ -28,7 +28,7 @@ Render Workflows support both Python and TypeScript. This repo contains Python e
 The simplest possible workflow — learn the fundamentals through simple number processing.
 
 - Ultra-simple task definitions
-- Clear subtask calling examples
+- Clear `ctx.run` subtask examples
 - Subtasks in loops demonstration
 - Multi-step workflow orchestration
 - Heavily commented code explaining every pattern

data-pipeline/README.md

@@ -179,7 +179,7 @@ Input:
 Once deployed, trigger the pipeline via the Render API or SDK:
 
 ```python
-from render_sdk import Render
+from render import Render
 
 # Uses RENDER_API_KEY environment variable automatically
 render = Render()
@@ -212,18 +212,20 @@ Using `asyncio.gather()` ensures all sources are fetched in parallel for maximum
 
 ### Stage 2: Transform
 
-**`transform_user_data`**: Combines data from all sources and enriches each user by calling subtasks:
+**`transform_user_data`**: Combines data from all sources and enriches each user by running subtasks:
 ```python
 for user in users:
     # SUBTASK CALL: Calculate metrics for this user
-    user_metrics = await calculate_user_metrics(user, transactions, engagement)
+    user_metrics = await ctx.run(
+        calculate_user_metrics, user, transactions, engagement
+    )
 
     # SUBTASK CALL: Enrich with geographic data
-    geo_data = await enrich_with_geo_data(user['email'])
+    geo_data = await ctx.run(enrich_with_geo_data, user['email'])
 
     enriched_users.append({**user_metrics, 'geo': geo_data})
 ```
-This demonstrates **sequential subtask calls per item** in a transformation loop.
+This demonstrates **sequential subtask runs per item** in a transformation loop.
 
 **`calculate_user_metrics`**: Calculates per-user metrics:
 - Total spent and refunded
@@ -283,9 +285,9 @@ This demonstrates **sequential subtask calls per item** in a transformation loop
 
 ```python
 # SUBTASK PATTERN: Launch multiple subtasks in parallel
-user_task = fetch_user_data(user_ids)
-transaction_task = fetch_transaction_data(user_ids)
-engagement_task = fetch_engagement_data(user_ids)
+user_task = ctx.run(fetch_user_data, user_ids)
+transaction_task = ctx.run(fetch_transaction_data, user_ids)
+engagement_task = ctx.run(fetch_engagement_data, user_ids)
 
 # SUBTASK CALLS: Wait for all three subtasks to complete
 user_data, transaction_data, engagement_data = await asyncio.gather(
@@ -304,15 +306,17 @@ Each user is enriched by calling multiple subtasks:
 ```python
 for user in users:
     # SUBTASK CALL: Calculate user-specific metrics
-    metrics = await calculate_user_metrics(user, transactions, engagement)
+    metrics = await ctx.run(
+        calculate_user_metrics, user, transactions, engagement
+    )
 
     # SUBTASK CALL: Enrich with geographic data
-    geo = await enrich_with_geo_data(user['email'])
+    geo = await ctx.run(enrich_with_geo_data, user['email'])
 
     enriched_users.append({**metrics, 'geo': geo})
 ```
 
-This shows **sequential subtask calls** for per-item enrichment.
+This shows **sequential subtask runs** for per-item enrichment.
 
 ### User Segmentation
 
@@ -327,7 +331,7 @@ Business logic classifies users into segments:
 **Add Real APIs**:
 ```python
 @app.task
-async def fetch_user_data_from_api(user_ids: list[str]) -> dict:
+async def fetch_user_data_from_api(ctx: TaskContext, user_ids: list[str]) -> dict:
     client = get_http_client()
     response = await client.post(
         "https://api.yourservice.com/users",
@@ -339,7 +343,7 @@ async def fetch_user_data_from_api(user_ids: list[str]) -> dict:
 **Add Database Integration**:
 ```python
 @app.task
-async def load_to_warehouse(insights: dict) -> dict:
+async def load_to_warehouse(ctx: TaskContext, insights: dict) -> dict:
     # Connect to data warehouse (Snowflake, BigQuery, etc.)
     # Insert aggregated insights
     # Return confirmation
@@ -349,7 +353,7 @@ async def load_to_warehouse(insights: dict) -> dict:
 **Add Caching**:
 ```python
 @app.task
-async def fetch_with_cache(source: str, key: str) -> dict:
+async def fetch_with_cache(ctx: TaskContext, source: str, key: str) -> dict:
     # Check Redis/Memcached
     # If miss, fetch from source and cache
     # Return data
@@ -359,7 +363,7 @@ async def fetch_with_cache(source: str, key: str) -> dict:
 **Add Notifications**:
 ```python
 @app.task
-async def send_pipeline_notification(result: dict) -> dict:
+async def send_pipeline_notification(ctx: TaskContext, result: dict) -> dict:
     # Send to Slack, email, etc.
     # Notify stakeholders of pipeline completion
     pass
@@ -375,7 +379,7 @@ async def send_pipeline_notification(result: dict) -> dict:
 
 ## Important Notes
 
-- **Python-only**: Workflows are only supported in Python via render-sdk
+- **Python-only**: Workflows are only supported in Python via `render`
 - **No Blueprint Support**: Workflows don't support render.yaml blueprint configuration
 - **Mock Data**: Example uses simulated data; replace with real API calls in production
 - **Idempotency**: Design pipeline to be safely re-runnable

data-pipeline/main.py

@@ -17,7 +17,7 @@
 import logging
 from datetime import datetime, timedelta
 
-from render_sdk import Retry, Workflows
+from render import Retry, TaskContext, Workflows
 
 # Configure logging
 logging.basicConfig(
@@ -55,7 +55,7 @@ def get_http_client():
 # ============================================================================
 
 @app.task
-async def fetch_user_data(user_ids: list[str]) -> dict:
+async def fetch_user_data(ctx: TaskContext, user_ids: list[str]) -> dict:
     """
     Fetch user profile data from user service.
 
@@ -90,7 +90,7 @@ async def fetch_user_data(user_ids: list[str]) -> dict:
 
 
 @app.task
-async def fetch_transaction_data(user_ids: list[str], days: int = 30) -> dict:
+async def fetch_transaction_data(ctx: TaskContext, user_ids: list[str], days: int = 30) -> dict:
     """
     Fetch transaction history for users.
 
@@ -130,7 +130,7 @@ async def fetch_transaction_data(user_ids: list[str], days: int = 30) -> dict:
 
 
 @app.task
-async def fetch_engagement_data(user_ids: list[str]) -> dict:
+async def fetch_engagement_data(ctx: TaskContext, user_ids: list[str]) -> dict:
     """
     Fetch user engagement metrics.
 
@@ -174,7 +174,7 @@ async def fetch_engagement_data(user_ids: list[str]) -> dict:
 # ============================================================================
 
 @app.task
-async def enrich_with_geo_data(user_email: str) -> dict:
+async def enrich_with_geo_data(ctx: TaskContext, user_email: str) -> dict:
     """
     Enrich user data with geographic information.
 
@@ -200,6 +200,7 @@ async def enrich_with_geo_data(user_email: str) -> dict:
 
 @app.task
 async def calculate_user_metrics(
+    ctx: TaskContext,
     user: dict,
     transactions: list[dict],
     engagement: dict
@@ -268,6 +269,7 @@ async def calculate_user_metrics(
 
 @app.task
 async def transform_user_data(
+    ctx: TaskContext,
     user_data: dict,
     transaction_data: dict,
     engagement_data: dict
@@ -303,11 +305,13 @@ async def transform_user_data(
         user_engagement = engagement_map.get(user['id'], {})
 
         # Calculate metrics for this user
-        user_metrics = await calculate_user_metrics(user, transactions, user_engagement)
+        user_metrics = await ctx.run(
+            calculate_user_metrics, user, transactions, user_engagement
+        )
 
         # Enrich with geo data
         user_email = user.get('email', f"{user['id']}@example.com")
-        geo_data = await enrich_with_geo_data(user_email)
+        geo_data = await ctx.run(enrich_with_geo_data, user_email)
         user_metrics['geo'] = geo_data
 
         enriched_users.append(user_metrics)
@@ -326,7 +330,7 @@ async def transform_user_data(
 # ============================================================================
 
 @app.task
-def aggregate_insights(enriched_data: dict) -> dict:
+def aggregate_insights(ctx: TaskContext, enriched_data: dict) -> dict:
     """
     Generate aggregate insights from enriched user data.
 
@@ -397,7 +401,7 @@ def aggregate_insights(enriched_data: dict) -> dict:
 # ============================================================================
 
 @app.task
-async def run_data_pipeline(user_ids: list[str]) -> dict:
+async def run_data_pipeline(ctx: TaskContext, user_ids: list[str]) -> dict:
     """
     Execute the complete data pipeline.
 
@@ -426,9 +430,9 @@ async def run_data_pipeline(user_ids: list[str]) -> dict:
     try:
         # Stage 1: EXTRACT - Fetch from all sources in parallel
         logger.info("[PIPELINE] Stage 1/3: EXTRACT (parallel)")
-        user_task = fetch_user_data(user_ids)
-        transaction_task = fetch_transaction_data(user_ids)
-        engagement_task = fetch_engagement_data(user_ids)
+        user_task = ctx.run(fetch_user_data, user_ids)
+        transaction_task = ctx.run(fetch_transaction_data, user_ids)
+        engagement_task = ctx.run(fetch_engagement_data, user_ids)
 
         # Wait for all extractions to complete
         user_data, transaction_data, engagement_data = await asyncio.gather(
@@ -441,7 +445,8 @@ async def run_data_pipeline(user_ids: list[str]) -> dict:
 
         # Stage 2: TRANSFORM - Combine and enrich
         logger.info("[PIPELINE] Stage 2/3: TRANSFORM")
-        enriched_data = await transform_user_data(
+        enriched_data = await ctx.run(
+            transform_user_data,
             user_data,
             transaction_data,
             engagement_data
@@ -451,7 +456,7 @@ async def run_data_pipeline(user_ids: list[str]) -> dict:
 
         # Stage 3: LOAD - Generate insights
         logger.info("[PIPELINE] Stage 3/3: AGGREGATE")
-        insights = await aggregate_insights(enriched_data)
+        insights = await ctx.run(aggregate_insights, enriched_data)
 
         logger.info("[PIPELINE] Insights generated successfully")
 

data-pipeline/requirements.txt

@@ -1,2 +1,2 @@
-render-sdk>=0.5.0
+render>=1.0.1
 httpx>=0.27.0

etl-job/README.md

@@ -13,7 +13,7 @@ Process customer signup data from CSV files with validation, cleaning, and stati
 
 ## Features
 
-- **Subtask Execution**: Demonstrates calling tasks from other tasks using `await`
+- **Subtask Execution**: Demonstrates running tasks from other tasks with `ctx.run`
 - **Extract**: Read data from CSV files (extensible to APIs, databases)
 - **Transform**: Validate records with comprehensive error tracking
 - **Load**: Compute statistics and prepare aggregated insights
@@ -126,7 +126,7 @@ Input:
 Once deployed, trigger the ETL pipeline via the Render API or SDK:
 
 ```python
-from render_sdk import Render
+from render import Render
 
 # Uses RENDER_API_KEY environment variable automatically
 render = Render()
@@ -163,23 +163,23 @@ This demonstrates how the pipeline handles data quality issues.
 - Validates age range (0-120)
 - Returns cleaned data with error tracking
 
-**`transform_batch`**: Processes all records by calling `validate_record` as a subtask for each one:
+**`transform_batch`**: Processes all records by running `validate_record` as a subtask for each one:
 ```python
 for record in records:
-    # Call validate_record as a subtask
-    validated = await validate_record(record)
+    # Run validate_record as a subtask on its own compute
+    validated = await ctx.run(validate_record, record)
 ```
-This demonstrates **calling subtasks in a loop** for batch processing.
+This demonstrates **running subtasks in a loop** for batch processing.
 
 **`compute_statistics`**: Aggregates valid records to produce:
 - Country distribution
 - Age statistics (min, max, average)
 - Data quality metrics
 
-**`run_etl_pipeline`**: Main orchestrator that calls three subtasks sequentially:
-1. `await extract_csv_data(source_file)` - Extract data
-2. `await transform_batch(raw_records)` - Validate records (which calls `validate_record` for each)
-3. `await compute_statistics(valid_records)` - Generate insights
+**`run_etl_pipeline`**: Main orchestrator that runs three subtasks sequentially:
+1. `await ctx.run(extract_csv_data, source_file)` - Extract data
+2. `await ctx.run(transform_batch, raw_records)` - Validate records (which runs `validate_record` for each)
+3. `await ctx.run(compute_statistics, valid_records)` - Generate insights
 
 This demonstrates **sequential subtask orchestration** for multi-stage pipelines.
 
@@ -188,7 +188,7 @@ This demonstrates **sequential subtask orchestration** for multi-stage pipelines
 **Add Database Loading**:
 ```python
 @app.task
-async def load_to_database(records: list[dict]) -> dict:
+async def load_to_database(ctx: TaskContext, records: list[dict]) -> dict:
     # Connect to database
     # Insert records
     # Return confirmation
@@ -198,7 +198,7 @@ async def load_to_database(records: list[dict]) -> dict:
 **Add API Data Source**:
 ```python
 @app.task
-async def extract_from_api(api_url: str) -> list[dict]:
+async def extract_from_api(ctx: TaskContext, api_url: str) -> list[dict]:
     # Fetch from REST API
     # Parse JSON response
     # Return records
@@ -210,16 +210,16 @@ async def extract_from_api(api_url: str) -> list[dict]:
 import asyncio
 
 @app.task
-async def transform_batch_parallel(records: list[dict]) -> dict:
+async def transform_batch_parallel(ctx: TaskContext, records: list[dict]) -> dict:
     # Validate all records in parallel
-    tasks = [validate_record(record) for record in records]
+    tasks = [ctx.run(validate_record, record) for record in records]
     results = await asyncio.gather(*tasks)
     # Aggregate results
     return results
 ```
 
 ## Important Notes
 
-- **Python-only**: Workflows are only supported in Python via render-sdk
+- **Python-only**: Workflows are only supported in Python via `render`
 - **No Blueprint Support**: Workflows don't support render.yaml blueprint configuration
 - **Service Type**: Deploy as a Workflow service on Render (not Background Worker or Web Service)