pylambdatasks 0.1.10

A library to simplify creating AWS Lambda images and managing multiple tasks within one image.

PyLambdaTasks

A Pythonic, Celery-like framework for building, testing, and deploying task queues on AWS Lambda.

PyLambdaTasks simplifies AWS Lambda development by letting you package multiple, independent tasks into a single container image. It combines the developer ergonomics of frameworks like Celery and FastAPI with a high-fidelity local emulator, enabling a seamless development-to-production workflow.

---

Core Pillars

• Celery-like Simplicity: Define tasks with a simple @app.task decorator and invoke them remotely with .invoke() or .delay().
• Unified Deployment: Package an entire suite of tasks into one Lambda function, reducing deployment complexity and cold starts.
• High-Fidelity Local Emulator: A built-in, Boto3-compatible server lets you run and test your entire Lambda function locally.
• FastAPI-Style Dependency Injection: Manage resources like database connections or user context with a powerful, clean DI system (Depends).
• Powerful CLI: A dedicated command-line interface to run the local emulator with live-reloading.

Installation

Install the library and its CLI dependencies from PyPI.

pip install "pylambdatasks[cli]"

Quick Start: Local-First Development

This guide demonstrates how to build and run a complete task system on your local machine.

1. Project Structure

Create a new project with the following structure:

my_lambda_project/
├── handler.py
├── tasks.py
└── client.py

2. Define Your Tasks (tasks.py)

Create your task functions in tasks.py. The @app.task decorator registers them. You can use Depends to inject dependencies like configuration or database connections.

# tasks.py
from typing import Annotated, Dict, Any
from handler import app  # Import the app instance from your handler file
from pylambdatasks import Depends

# 1. Define a dependency (e.g., a function to get user context)
def get_user_context() -> Dict[str, str]:
    """A simple dependency providing a mock user context."""
    return {"user_id": "user-123", "role": "admin"}

# 2. Create a type alias for the dependency
UserContext = Annotated[Dict[str, str], Depends(get_user_context)]

@app.task(name="ADD_NUMBERS")
async def add_numbers(a: int, b: int) -> Dict:
    """A simple task that adds two numbers."""
    print(f"Executing ADD_NUMBERS: {a} + {b}")
    return {"result": a + b}

@app.task(name="PROCESS_DATA")
async def process_data(data: Dict[str, Any], context: UserContext) -> Dict:
    """
    A task that processes data and uses an injected dependency.
    The 'context' argument will be automatically resolved and provided.
    """
    print(f"Executing PROCESS_DATA for user: {context['user_id']}")
    processed_data = {key.upper(): value for key, value in data.items()}
    return {"processed_data": processed_data}

3. Configure the Application (handler.py)

This file is the heart of your Lambda function. You instantiate LambdaTasks and expose the handler.

# handler.py
from pylambdatasks import LambdaTasks

app = LambdaTasks(
    # A list of modules where your @app.task-decorated functions live.
    task_modules=['tasks'],

    # The default AWS Lambda function name tasks will be invoked against.
    default_lambda_function_name="PyLambdaTasks-Local",

    # Boto3 client configuration.
    region_name="us-east-1",
    aws_access_key_id="test",
    aws_secret_access_key="test",

    # For local development, this URL points to our emulator.
    # Boto3 will automatically connect to it instead of AWS.
    endpoint_url="http://lambda:8080"
)

# This is the handler entrypoint that the emulator (and AWS Lambda) will use.
handler = app.handler

4. Run the Local Emulator

With your tasks and handler defined, start the local emulator from your terminal using the pylambdatasks CLI.

# This command tells the CLI to:
# 1. Look for the `app` object inside the `handler.py` module.
# 2. Start the emulator server.
# 3. Watch for file changes and reload automatically.
pylambdatasks run handler:app --reload

You will see output confirming the server is running and ready to accept Boto3 requests.

5. Invoke Tasks (client.py)

From a separate process (like a web API or another script), you can invoke your tasks. The task objects (add_numbers, process_data) now have .invoke() and .delay() methods that trigger a Lambda invocation behind the scenes.

# client.py
import asyncio
from tasks import add_numbers, process_data

async def main():
    print("Invoking 'ADD_NUMBERS' task synchronously...")
    # .invoke() calls the Lambda, waits for the result, and returns it.
    sync_result = await add_numbers.invoke(a=10, b=5)
    print(f"  -> Result: {sync_result}")

    print("\nInvoking 'PROCESS_DATA' task synchronously...")
    # .delay() is for "fire-and-forget" asynchronous invocation.
    await process_data.delay(data={"message": "hello world"})
    print("  -> 'PROCESS_DATA' task dispatched.")

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

Run the client in a new terminal:

python client.py

You'll see the results printed in your client terminal and the execution logs (Executing ADD_NUMBERS...) in your emulator terminal.

---

Docker & AWS Lambda

The strategy is to build a single, production-ready Docker image that can be switched into "development mode" using a docker-compose.yml override.

The Production Dockerfile

A production image for AWS Lambda requires an entrypoint that uses the official AWS Runtime Interface Client (awslambdaric).

# Dockerfile
FROM python:3.11-slim-bookworm

WORKDIR /var/task
ENV PYTHONDONTWRITEBYTECODE 1
ENV PYTHONUNBUFFERED 1

# Install project dependencies
COPY requirements.txt .
# Ensure pylambdatasks is listed in your requirements.txt
RUN pip install --no-cache-dir "pylambdatasks[cli]" -r requirements.txt

# Copy application code
COPY . .

# This is the official AWS entrypoint for custom Python runtimes.
# THIS IS REQUIRED FOR THE IMAGE TO RUN IN AWS LAMBDA.
ENTRYPOINT [ "/usr/local/bin/python", "-m", "awslambdaric" ]

# The CMD specifies the handler that awslambdaric should run.
CMD [ "handler.handler" ]

Development with Docker Compose

When running locally, we override the production ENTRYPOINT and CMD to run our local emulator instead. This is the key to a seamless workflow: one image, two modes.

# compose.yml
services:
  # This service runs our Lambda code in emulator mode
  lambda:
    build: . # Build the production Dockerfile
    container_name: pylambdatasks-lambda
    volumes:
      - .:/var/task # Mount code for live-reloading
    ports:
      - "8080:8080" # Expose the emulator port

    # --- DEVELOPMENT OVERRIDE ---
    # We discard the production ENTRYPOINT and CMD from the Dockerfile
    # and replace them with our local emulator command.
    entrypoint: "" # Clear the production entrypoint
    command:
      - pylambdatasks
      - run
      - handler:app   # The app instance to run
      - --reload      # Enable live-reloading
      - --host
      - 0.0.0.0
      - --port
      - "8080"

With this setup, docker compose up will use your production image for local development. When you're ready to deploy, push the exact same image to AWS ECR, and it will work correctly in Lambda.

Lifecycle Hooks

You can register functions to run at different stages of the Lambda lifecycle.

• @app.on_startup() — Runs only once on cold-start. Ideal for setting up database connection pools.
• @app.on_shutdown() — Runs when the Lambda container is about to be shut down.
• @app.before_request() — Runs before every single invocation.
• @app.after_request() — Runs after every single invocation, even if it failed.

Example (handler.py):

# ... app = LambdaTasks(...)

@app.on_startup()
async def setup_connections():
    print("Cold Start: Creating DB pool...")

@app.on_shutdown()
async def close_connections():
    print("Shutdown: Closing DB pool...")

@app.before_request()
async def log_invocation_start():
    print("-> Invocation started.")

handler = app.handler

CLI Reference

pylambdatasks run [OPTIONS] APP_PATH

Starts the local Lambda emulator.

• APP_PATH: Path to your app instance in MODULE:VARIABLE format (e.g., handler:app).
• --host TEXT: Host to bind to. Defaults to 0.0.0.0.
• --port INTEGER: Port to bind to. Defaults to 8080.
• --reload: Enable auto-reloading on code changes.

License

This project is licensed under the GNU General Public License v3.0.

---

Roadmap

• Custom Logging: Provide hooks or a configurable system to integrate with custom logging solutions and structured log formats.
• AIOBoto3 Support: Offer native integration with aioboto3 for a fully asynchronous AWS client experience.
• UI for Task Monitoring: Develop a simple web-based user interface to visualize task invocations and monitor the system locally.
• Batching and SQS Integration: Add native support for SQS triggers and batch processing of tasks.