البرنامج التعليمي AWS Lambda 2026: وظائف Python وSAM وFastAPI على بدون خادم

# lambda_function.py
import json
import logging

logger = logging.getLogger()
logger.setLevel(logging.INFO)

def lambda_handler(event: dict, context) -> dict:
    logger.info(f"Event: {json.dumps(event)}")

    # context has: function_name, remaining_time_in_millis, memory_limit_in_mb
    logger.info(f"Remaining time: {context.get_remaining_time_in_millis()}ms")

    name = event.get("queryStringParameters", {}).get("name", "World")

    return {
        "statusCode": 200,
        "headers": {
            "Content-Type": "application/json",
            "Access-Control-Allow-Origin": "*",
        },
        "body": json.dumps({
            "message": f"Hello, {name}!",
            "requestId": context.aws_request_id,
        })
    }

# Install AWS SAM CLI
pip install aws-sam-cli

# Initialize project
sam init --runtime python3.12 --app-template hello-world

# Build
sam build

# Local test
sam local invoke HelloWorldFunction --event events/event.json
sam local start-api  # runs locally on port 3000

# Deploy to AWS
sam deploy --guided  # first time (interactive)
sam deploy           # subsequent deploys

# template.yaml (SAM template)
AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31

Globals:
  Function:
    Runtime: python3.12
    Timeout: 30
    MemorySize: 256
    Environment:
      Variables:
        ENVIRONMENT: !Ref Environment
    Layers:
      - !Ref DependenciesLayer

Parameters:
  Environment:
    Type: String
    AllowedValues: [dev, staging, production]

Resources:
  ApiFunction:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: src/
      Handler: api.handler
      Policies:
        - DynamoDBCrudPolicy:
            TableName: !Ref UsersTable
        - S3ReadPolicy:
            BucketName: !Ref AssetsBucket
      Events:
        Api:
          Type: Api
          Properties:
            Path: /{proxy+}
            Method: ANY
        Schedule:
          Type: Schedule
          Properties:
            Schedule: rate(5 minutes)

  DependenciesLayer:
    Type: AWS::Serverless::LayerVersion
    Properties:
      ContentUri: dependencies/
      CompatibleRuntimes: [python3.12]

  UsersTable:
    Type: AWS::DynamoDB::Table
    Properties:
      BillingMode: PAY_PER_REQUEST
      AttributeDefinitions:
        - AttributeName: pk
          AttributeType: S
        - AttributeName: sk
          AttributeType: S
      KeySchema:
        - AttributeName: pk
          KeyType: HASH
        - AttributeName: sk
          KeyType: RANGE

pip install fastapi mangum
# Package: put in layer or zip with requirements.txt

# main.py
from fastapi import FastAPI, Depends, HTTPException
from mangum import Mangum
from pydantic import BaseModel

app = FastAPI(
    title="My API",
    root_path="/production"  # API Gateway stage
)

class User(BaseModel):
    id: int
    name: str
    email: str

@app.get("/users/{user_id}")
async def get_user(user_id: int) -> User:
    user = await db.get_user(user_id)
    if not user:
        raise HTTPException(404, "User not found")
    return user

@app.post("/users")
async def create_user(user: User) -> User:
    return await db.create_user(user)

# Lambda handler
handler = Mangum(app, lifespan="off")

# S3 trigger — process uploaded files
def handle_s3_upload(event: dict, context) -> None:
    for record in event.get('Records', []):
        bucket = record['s3']['bucket']['name']
        key = record['s3']['object']['key']
        size = record['s3']['object']['size']
        print(f"Processing {key} ({size} bytes) from {bucket}")
        process_file(bucket, key)

# SQS trigger — process messages
def handle_sqs_messages(event: dict, context) -> dict:
    batch_failures = []
    for record in event.get('Records', []):
        try:
            body = json.loads(record['body'])
            process_message(body)
        except Exception as e:
            # Return failed message IDs for retry
            batch_failures.append({"itemIdentifier": record['messageId']})
    return {"batchItemFailures": batch_failures}

# DynamoDB Streams trigger
def handle_db_changes(event: dict, context) -> None:
    for record in event.get('Records', []):
        if record['eventName'] == 'INSERT':
            new_item = record['dynamodb']['NewImage']
            on_user_created(new_item)
        elif record['eventName'] == 'MODIFY':
            old = record['dynamodb']['OldImage']
            new = record['dynamodb']['NewImage']
            on_user_updated(old, new)

# EventBridge scheduled task
def run_nightly_report(event: dict, context) -> None:
    print(f"Running nightly report at {event['time']}")
    generate_and_email_report()