Context

Execution context and logging in function handlers

Overview

The III SDK provides execution context that includes a logger and optional OpenTelemetry span. Context is automatically injected into function handlers and can be accessed via get_context().

Context

Execution context containing logger and tracing utilities.

from pydantic import BaseModel

class Context(BaseModel):
    logger: Logger
    trace: Any | None = None  # OpenTelemetry span

logger · Logger

Context-aware logger that emits OpenTelemetry LogRecords

trace · Any | None

The active OpenTelemetry span for adding custom attributes, events, etc.

Accessing Context

get_context

Get the current execution context within a function handler.

from iii import get_context

async def my_function(data):
    ctx = get_context()
    ctx.logger.info("Processing request", data={"user_id": data["user_id"]})
    
    # Your logic here
    
    ctx.logger.info("Request completed")
    return {"status": "success"}

context · Context

The current execution context, or a default context if none is set

Default Context

If no context is active, get_context() returns a default context with a basic logger:

ctx = get_context()  # Always returns a Context, never None
ctx.logger.info("This works even outside function handlers")

Logger

Context-aware logger that emits OpenTelemetry LogRecords when OTel is active.

Log Methods

from iii import get_context

ctx = get_context()

# Info level
ctx.logger.info("User logged in", data={"user_id": "123"})

# Warning level
ctx.logger.warn("High memory usage", data={"usage_percent": 85})

# Error level
ctx.logger.error("Database connection failed", data={"error": "timeout"})

# Debug level
ctx.logger.debug("Request details", data={"headers": headers})

message · str · required

Log message

data · Any

Additional structured data to include in the log record

Logger Behavior

When OpenTelemetry is initialized:

• Logs are emitted as OpenTelemetry LogRecords
• Includes trace context (span ID, trace ID)
• Exported to the III Engine
• Function name is automatically included

When OpenTelemetry is not initialized:

• Falls back to Python’s standard logging module
• Logs to console/file handlers as configured
• Still includes function name and data

Function Name

The logger automatically includes the function name:

iii.register_function("orders.process", process_order)

async def process_order(data):
    ctx = get_context()
    # Log includes function_name="orders.process"
    ctx.logger.info("Order received")

with_context

Execute a function within a specific context (internal use).

from iii import with_context, Context, Logger

logger = Logger(function_name="my.function")
ctx = Context(logger=logger)

async def handler(ctx: Context) -> dict:
    ctx.logger.info("Inside context")
    return {"result": "success"}

result = await with_context(handler, ctx)

This is primarily used internally by the SDK to inject context into function handlers. You typically don’t need to call this directly.

fn · Callable[[Context], Awaitable[T]] · required

Async function to execute with the context

context · Context · required

The context to use

result · T

The function’s return value

Example: Structured Logging

import asyncio
from iii import III, get_context, init_otel

init_otel()  # Enable OpenTelemetry logging

iii = III("ws://localhost:49134")

async def create_user(data):
    ctx = get_context()
    
    ctx.logger.info("Creating user", data={
        "email": data["email"],
        "role": data.get("role", "user")
    })
    
    try:
        # Validate email
        if "@" not in data["email"]:
            ctx.logger.warn("Invalid email format", data={"email": data["email"]})
            raise ValueError("Invalid email")
        
        # Save to database
        user_id = save_user(data)
        
        ctx.logger.info("User created successfully", data={
            "user_id": user_id,
            "email": data["email"]
        })
        
        return {"id": user_id, "email": data["email"]}
    
    except Exception as e:
        ctx.logger.error("Failed to create user", data={
            "error": str(e),
            "email": data.get("email")
        })
        raise

iii.register_function("users.create", create_user)

async def main():
    await iii.connect()
    
    try:
        result = await iii.call("users.create", {
            "email": "[e-mail]",
            "role": "admin"
        })
        print(f"User created: {result}")
    except Exception as e:
        print(f"Failed: {e}")

if __name__ == "__main__":
    asyncio.run(main())

Example: Logging with Trace Context

import asyncio
from iii import III, get_context, init_otel, get_tracer
from opentelemetry import trace

init_otel()
tracer = get_tracer()

iii = III("ws://localhost:49134")

async def process_order(data):
    ctx = get_context()
    
    with tracer.start_as_current_span("validate-order") as span:
        ctx.logger.info("Validating order", data={"order_id": data["order_id"]})
        
        # Validation logic
        span.set_attribute("order.items", len(data.get("items", [])))
        
        ctx.logger.info("Order validated")
    
    with tracer.start_as_current_span("process-payment"):
        ctx.logger.info("Processing payment", data={
            "order_id": data["order_id"],
            "amount": data["amount"]
        })
        
        # Payment logic
        
        ctx.logger.info("Payment processed")
    
    ctx.logger.info("Order completed", data={"order_id": data["order_id"]})
    return {"status": "completed"}

iii.register_function("orders.process", process_order)

Example: Error Tracking

from iii import get_context
import traceback

async def risky_operation(data):
    ctx = get_context()
    
    try:
        ctx.logger.debug("Starting risky operation", data={"input": data})
        
        # Potentially failing operation
        result = divide(data["a"], data["b"])
        
        ctx.logger.debug("Operation succeeded", data={"result": result})
        return {"result": result}
    
    except ZeroDivisionError:
        ctx.logger.error("Division by zero", data={
            "a": data["a"],
            "b": data["b"]
        })
        raise
    
    except Exception as e:
        ctx.logger.error("Unexpected error", data={
            "error": str(e),
            "traceback": traceback.format_exc()
        })
        raise

Example: Request/Response Logging

import time
from iii import get_context

async def api_handler(data):
    ctx = get_context()
    start_time = time.time()
    
    # Log request
    ctx.logger.info("API request received", data={
        "method": data.get("method"),
        "path": data.get("path"),
        "user_id": data.get("user_id")
    })
    
    try:
        # Process request
        result = process_request(data)
        
        # Log response
        duration_ms = (time.time() - start_time) * 1000
        ctx.logger.info("API request completed", data={
            "status": "success",
            "duration_ms": duration_ms
        })
        
        return result
    
    except Exception as e:
        duration_ms = (time.time() - start_time) * 1000
        ctx.logger.error("API request failed", data={
            "status": "error",
            "error": str(e),
            "duration_ms": duration_ms
        })
        raise

Logger Implementation Details

The Logger class automatically:

1. Detects OTel initialization: Uses is_initialized() to check if OTel is active
2. Emits LogRecords: When OTel is active, creates OpenTelemetry LogRecords with:
   • Timestamp (nanoseconds)
   • Severity level and text
   • Message body
   • Function name attribute
   • Custom data attribute
   • Trace context (span ID, trace ID, trace flags)
3. Falls back to Python logging: When OTel is not active, uses Python’s logging module
4. Severity mapping:
   • debug → DEBUG (5)
   • info → INFO (9)
   • warn → WARN (13)
   • error → ERROR (17)

Best Practices

1. Use structured data: Pass dictionaries to the data parameter instead of formatting strings
2. Include context: Add relevant identifiers (user_id, order_id, etc.) to logs
3. Appropriate levels: Use info for business logic, debug for diagnostics, error for failures
4. Avoid sensitive data: Don’t log passwords, tokens, or PII
5. Measure performance: Log durations for slow operations
6. Trace correlation: Logs automatically include trace context when OTel is active