Cours 15 : Cloud Python

1. AWS SDK (boto3)

1.1 S3

import boto3
from pathlib import Path

s3 = boto3.client("s3")
BUCKET = "my-bucket"

def upload_file(local_path: str, s3_key: str) -> bool:
    try:
        s3.upload_file(local_path, BUCKET, s3_key)
        return True
    except Exception as e:
        print(f"Upload error: {e}")
        return False

def download_file(s3_key: str, local_path: str):
    s3.download_file(BUCKET, s3_key, local_path)

def list_files(prefix: str = "") -> list:
    response = s3.list_objects_v2(Bucket=BUCKET, Prefix=prefix)
    return [obj["Key"] for obj in response.get("Contents", [])]

def generate_presigned_url(s3_key: str, expiration=3600):
    return s3.generate_presigned_url(
        "get_object",
        Params={"Bucket": BUCKET, "Key": s3_key},
        ExpiresIn=expiration,
    )

1.2 SQS

sqs = boto3.client("sqs")
queue_url = "https://sqs.eu-west-3.amazonaws.com/123456789/my-queue"

# Send
sqs.send_message(
    QueueUrl=queue_url,
    MessageBody='{"key": "value"}',
    DelaySeconds=0,
)

# Receive
messages = sqs.receive_message(
    QueueUrl=queue_url,
    MaxNumberOfMessages=10,
    WaitTimeSeconds=20,  # Long polling
)

# Delete
if messages.get("Messages"):
    for msg in messages["Messages"]:
        sqs.delete_message(
            QueueUrl=queue_url,
            ReceiptHandle=msg["ReceiptHandle"],
        )

1.3 DynamoDB

dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table("Users")

# CRUD
table.put_item(Item={"id": "1", "name": "Alice", "email": "alice@example.com"})
response = table.get_item(Key={"id": "1"})
item = response.get("Item")
table.update_item(
    Key={"id": "1"},
    UpdateExpression="SET #n = :name",
    ExpressionAttributeNames={"#n": "name"},
    ExpressionAttributeValues={":name": "Alice Updated"},
)
table.delete_item(Key={"id": "1"})

# Query
response = table.query(
    KeyConditionExpression=boto3.dynamodb.conditions.Key("id").eq("1")
)

1.4 Lambda invocation

lambda_client = boto3.client("lambda")
response = lambda_client.invoke(
    FunctionName="my-function",
    InvocationType="RequestResponse",  # ou "Event" pour async
    Payload=json.dumps({"key": "value"}),
)
result = json.loads(response["Payload"].read())

2. AWS Lambda et Powertools

2.1 Handler basique

import json

def handler(event, context):
    return {
        "statusCode": 200,
        "body": json.dumps({"message": "Hello from Lambda!"}),
        "headers": {"Content-Type": "application/json"},
    }

2.2 Lambda Powertools

from aws_lambda_powertools import Logger, Tracer, Metrics
from aws_lambda_powertools.event_handler import APIGatewayRestResolver
from aws_lambda_powertools.utilities.typing import LambdaContext

logger = Logger()
tracer = Tracer()
metrics = Metrics()
app = APIGatewayRestResolver()

@app.get("/users/{id}")
@tracer.capture_method
def get_user(id: str):
    logger.info(f"Getting user {id}")
    metrics.add_metric(name="UserRetrieved", unit="Count", value=1)
    return {"id": id, "name": "Alice"}

@logger.inject_lambda_context
@tracer.capture_lambda_handler
@metrics.log_metrics(capture_cold_start=True)
def handler(event: dict, context: LambdaContext):
    return app.resolve(event, context)

3. AWS SAM

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

Globals:
  Function:
    Runtime: python3.12
    Timeout: 30
    Tracing: Active

Resources:
  MyApi:
    Type: AWS::Serverless::Api
    Properties:
      StageName: prod

  MyFunction:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: src/
      Handler: app.handler
      Events:
        ApiEvent:
          Type: Api
          Properties:
            RestApiId: !Ref MyApi
            Path: /{proxy+}
            Method: ANY

sam build
sam deploy --guided
sam logs -n MyFunction --tail

4. AWS CDK

from aws_cdk import (
    App, Stack, Duration,
    aws_lambda as lambda_,
    aws_s3 as s3,
    aws_dynamodb as dynamodb,
    aws_sqs as sqs,
)

class DataPipelineStack(Stack):
    def __init__(self, app: App, id: str):
        super().__init__(app, id)

        bucket = s3.Bucket(self, "DataBucket",
            versioned=True,
            removal_policy=RemovalPolicy.DESTROY,
            lifecycle_rules=[s3.LifecycleRule(expiration=Duration.days(90))],
        )

        table = dynamodb.Table(self, "MetadataTable",
            partition_key=dynamodb.Attribute(name="id", type=dynamodb.AttributeType.STRING),
            billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST,
        )

        queue = sqs.Queue(self, "TaskQueue",
            visibility_timeout=Duration.seconds(300),
        )

        fn = lambda_.Function(self, "Processor",
            runtime=lambda_.Runtime.PYTHON_3_12,
            handler="processor.handler",
            code=lambda_.Code.from_asset("src"),
            timeout=Duration.minutes(5),
            environment={
                "BUCKET_NAME": bucket.bucket_name,
                "TABLE_NAME": table.table_name,
                "QUEUE_URL": queue.queue_url,
            },
        )

        bucket.grant_read_write(fn)
        table.grant_read_write_data(fn)
        queue.grant_send_messages(fn)

app = App()
DataPipelineStack(app, "DataPipelineStack")
app.synth()

cdk deploy
cdk destroy
cdk diff

5. Google Cloud

5.1 GCS

from google.cloud import storage

client = storage.Client()

# Upload
bucket = client.bucket("my-bucket")
blob = bucket.blob("data/file.csv")
blob.upload_from_filename("local_file.csv")

# Download
blob.download_to_filename("downloaded.csv")

# List
blobs = client.list_blobs("my-bucket", prefix="data/")
for blob in blobs:
    print(blob.name)

# Generate signed URL
url = blob.generate_signed_url(
    version="v4",
    expiration=3600,
    method="GET",
)

5.2 Pub/Sub

from google.cloud import pubsub_v1

project_id = "my-project"

# Publish
publisher = pubsub_v1.PublisherClient()
topic = publisher.topic_path(project_id, "my-topic")
future = publisher.publish(topic, b"Message data", source="python")
future.result()

# Subscribe
subscriber = pubsub_v1.SubscriberClient()
subscription = subscriber.subscription_path(project_id, "my-subscription")

def callback(message):
    print(f"Received: {message.data}")
    message.ack()

streaming_pull = subscriber.subscribe(subscription, callback=callback)
streaming_pull.result()

5.3 Cloud Run

# main.py
from fastapi import FastAPI
import os

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Hello from Cloud Run!", "revision": os.environ.get("K_REVISION", "local")}

@app.get("/health")
async def health():
    return {"status": "healthy"}

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY src/ src/
CMD ["uvicorn", "src.main:app", "--host", "0.0.0.0", "--port", "8080"]

gcloud builds submit --tag gcr.io/PROJECT/my-app
gcloud run deploy my-app --image gcr.io/PROJECT/my-app --platform managed

6. Kubernetes

6.1 Python client

from kubernetes import client, config, watch

# Load config
config.load_kube_config()

# Core API
v1 = client.CoreV1Api()
pods = v1.list_pod_for_all_namespaces()
for pod in pods.items:
    print(f"{pod.metadata.name} - {pod.status.phase}")

# Deploy pod
pod = client.V1Pod(
    metadata=client.V1ObjectMeta(name="my-pod"),
    spec=client.V1PodSpec(
        containers=[client.V1Container(
            name="app",
            image="python:3.12-slim",
            command=["python", "-c", "print('hello')"],
        )],
        restart_policy="Never",
    ),
)
v1.create_namespaced_pod(namespace="default", body=pod)

# Watch events
w = watch.Watch()
for event in w.stream(v1.list_namespaced_pod, namespace="default"):
    print(f"{event['type']}: {event['object'].metadata.name}")
    if event['object'].status.phase == 'Running':
        w.stop()

7. Serverless

7.1 Chalice

from chalice import Chalice

app = Chalice(app_name="my-api")
app.debug = True

@app.route("/users/{id}")
def get_user(id):
    return {"id": id, "name": "Alice"}

@app.route("/users", methods=["POST"])
def create_user():
    body = app.current_request.json_body
    return {"created": body["name"]}

@app.schedule("rate(1 hour)")
def periodic_task(event):
    print("Running scheduled task...")

8. Terraform CDK

import cdktf
from constructs import Construct
from imports.aws import AwsProvider, S3Bucket, DynamodbTable, LambdaFunction

class MyStack(cdktf.TerraformStack):
    def __init__(self, scope: Construct, id: str):
        super().__init__(scope, id)
        AwsProvider(self, "AWS", region="eu-west-3")
        S3Bucket(self, "Data", bucket="my-unique-bucket-123")
        DynamodbTable(self, "Users",
            name="Users",
            hash_key="id",
            attribute=[{"name": "id", "type": "S"}],
            billing_mode="PAY_PER_REQUEST",
        )

app = cdktf.App()
MyStack(app, "my-stack")
app.synth()

cdktf init --template python
cdktf get
cdktf deploy

Diagramme en cours de génération...