Lambda Forge

A Modern, Celery-like Framework for AWS Lambda.

Lambda Forge brings the developer experience of modern frameworks like FastAPI and Celery to the world of serverless, allowing you to build powerful, stateful, and maintainable AWS Lambda functions with ease.

Tired of boilerplate code and complex setups? Lambda Forge provides a clean, class-based approach to defining, invoking, and managing asynchronous tasks, complete with dependency injection and state tracking.

Core Features

• Celery-like Task Declaration: Define your Lambda tasks with a simple @lambda_task decorator.
• Powerful Dependency Injection: Manage dependencies like database connections or user context elegantly using LambdaDepends.
• Asynchronous Support: Built from the ground up with asyncio for high-performance, non-blocking tasks.
• Stateful Task Tracking: Automatically track the status (PROGRESS, SUCCESS, FAILED) and results of your tasks using a Valkey (or Redis) backend.
• Local Development Environment: A streamlined Docker Compose setup allows you to develop and test your functions locally, simulating the AWS Lambda environment.
• Clean, Explicit Configuration: No more magic environment variables deep in your code. Configure everything through clear, typed configuration objects.

Installation

Since Lambda Forge is not yet on PyPI, you can install it directly from the GitHub repository using pip:

pip install git+https://github.com/oktylab/lambda-forge.git

For a more stable build, you can pin it to a specific commit hash:

pip install git+https://github.com/oktylab/lambda-forge.git@<commit_hash>

Quick Start: Building Your First Forged Lambda

Let's build a simple function that adds two numbers, includes a dependency, and tracks its state.

1. Project Structure

Organize your project as follows:

your_project/
├── app/
│   ├── __init__.py
│   └── tasks.py        # Where your tasks are defined
├── Dockerfile
├── requirements.txt
├── compose.yml
└── handler.py          # Your Lambda entrypoint

2. Define Your Task (app/tasks.py)

Create a task using the @lambda_task decorator. We'll also define a simple dependency to inject user context.

# app/tasks.py
from typing import Annotated
from lambda_forge import lambda_task, LambdaTask, LambdaDepends

def get_user_context():
    """A simple dependency that provides a user context dictionary."""
    print("Dependency 'get_user_context' was resolved.")
    return {"user_id": "user-123", "role": "admin"}

# Use Annotated and LambdaDepends for dependency injection
UserContextDep = Annotated[dict, LambdaDepends(get_user_context)]

@lambda_task(task_name="ADD_NUMBERS", lf_name="my-example-lambda-function")
async def add_numbers_task(
    self: LambdaTask, # The task instance for state updates
    a: int,
    b: int,
    context: UserContextDep, # The injected dependency
):
    """
    Adds two numbers, updates the task state, and returns the result.
    """
    print(f"Task {self.id} started. Received numbers: {a}, {b}.")
    print(f"Injected context: {context}")
    
    result = a + b
    
    # Update the task's state in Valkey
    await self.update_state(status="SUCCESS", result={"sum": result})
    
    return {"sum": result, "processed_by": context.get("user_id")}

3. Configure the Application (handler.py)

This is your main entrypoint. Here you configure AWS, Valkey, and instantiate the LambdaForge app.

# handler.py
import os
from lambda_forge import LambdaForge, AWSConfig, ValkeyConfig

# 1. Define where your tasks are located
LAMBDA_TASK_MODULES: list[str] = ['app.tasks']

# 2. Create an explicit AWS configuration object
aws_settings = AWSConfig(
    aws_access_key_id=os.environ["AWS_ACCESS_KEY_ID"],
    aws_secret_access_key=os.environ["AWS_SECRET_ACCESS_KEY"],
    region_name=os.environ["AWS_REGION"],
)

# 3. Create an explicit Valkey configuration object
valkey_settings = ValkeyConfig(
    host=os.environ["VALKEY_HOST"],
    port=int(os.environ["VALKEY_PORT"]),
)

# 4. Instantiate the LambdaForge app with the configurations
app = LambdaForge(
    task_modules=LAMBDA_TASK_MODULES,
    aws_config=aws_settings,
    valkey_config=valkey_settings,
    local_invocation_mode=os.environ.get("LOCAL_INVOCATION_MODE", "HTTP"),
)

# 5. Expose the handler for the AWS Lambda runtime
handler = app.handler

4. Set Up Your Local Environment

To run and test everything locally, create the following files.

requirements.txt

boto3>=1.34.0
valkey>=0.2.0
httpx>=0.27.0
# Add your other dependencies here

Dockerfile

# STAGE 1: BASE
FROM python:3.11-slim-bookworm AS base
ENV PYTHONDONTWRITEBYTECODE 1
ENV PYTHONUNBUFFERED 1
RUN apt-get update && apt-get install -y curl && apt-get clean && rm -rf /var/lib/apt/lists/*

# STAGE 2: BUILDER
FROM base AS builder
WORKDIR /install
RUN pip install --no-cache-dir --upgrade pip
COPY ./requirements.txt .
# Install lambda-forge from GitHub
RUN pip install --no-cache-dir git+https://github.com/oktylab/lambda-forge.git
RUN pip install --no-cache-dir -r requirements.txt

# STAGE 3: LAMBDA (Final Stage)
FROM builder AS lambda
WORKDIR /var/task
COPY --from=builder /install /var/task
ENV PYTHONPATH=/var/task

RUN pip install --no-cache-dir awslambdaric
RUN curl -Lo /usr/local/bin/aws-lambda-rie https://github.com/aws/aws-lambda-runtime-interface-emulator/releases/latest/download/aws-lambda-rie
RUN chmod +x /usr/local/bin/aws-lambda-rie

# Copy your application code
COPY ./app /var/task/app
COPY ./handler.py /var/task/

# Lambda Runtime Interface Emulator entrypoint script
COPY ./lambda_entry.sh /lambda_entry.sh
RUN chmod +x /lambda_entry.sh

ENTRYPOINT [ "/lambda_entry.sh" ]
CMD [ "handler.handler" ]

lambda_entry.sh

#!/bin/sh
if [ -z "${AWS_LAMBDA_RUNTIME_API}" ]; then
  exec /usr/local/bin/aws-lambda-rie /usr/local/bin/python -m awslambdaric "$@"
else
  exec /usr/local/bin/python -m awslambdaric "$@"
fi

docker-compose.yml

services:
  valkey:
    image: valkey/valkey:7.2-alpine
    container_name: valkey
    ports:
      - "6379:6379"
    networks:
      - lambda-net

  lambda_service:
    container_name: lambda_service
    build:
      context: .
      dockerfile: ./Dockerfile
    ports:
      - "9000:8080"
    networks:
      - lambda-net
    depends_on:
      - valkey
    environment:
      - AWS_ACCESS_KEY_ID=test
      - AWS_SECRET_ACCESS_KEY=test
      - AWS_REGION=us-east-1
      - VALKEY_HOST=valkey
      - VALKEY_PORT=6379
      - LOCAL_INVOCATION_MODE=HTTP

networks:
  lambda-net:
    driver: bridge

5. Run and Invoke!

With all the files in place, start your local environment:

docker-compose up --build

Once the services are running, invoke your function with a simple cURL request:

curl -X POST "http://localhost:9000/2015-03-31/functions/function/invocations" -d '{
  "task_name": "ADD_NUMBERS",
  "a": 50,
  "b": 50
}'

You should receive a successful JSON response:

{
  "sum": 100,
  "processed_by": "user-123"
}

Congratulations! You have successfully built, deployed, and invoked a stateful, dependency-injected Lambda function using Lambda Forge.

Next Steps

Explore the core components to understand the full power of Lambda Forge:

• LambdaForge(...): The central application object. Pass your AWSConfig, ValkeyConfig, and task_modules here.
• @lambda_task(...): The decorator for your task functions.
  • task_name: The unique identifier for this task, used in the invocation event.
  • lf_name: The deployed AWS Lambda function name, used for remote invocations.
• LambdaDepends(...): Used with typing.Annotated to declare dependencies for your task functions.
• AsyncResult: The object returned when you invoke a task asynchronously (e.g., my_task.delay()). Use result.get() to retrieve the state and final result from Valkey.

Contributing

Contributions are welcome! Please feel free to open an issue or submit a pull request.

License

This project is licensed under the MIT License. See the LICENSE file for details.