From d9dd1281c33f60a9b28c2ad148572ebaac2693b5 Mon Sep 17 00:00:00 2001 From: Rene Luria Date: Tue, 7 Oct 2025 08:12:54 +0200 Subject: [PATCH] refactor: move S3 functions to separate file for clarity --- app/main.py | 137 +++++++++--------------------------------------- app/s3_utils.py | 116 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 141 insertions(+), 112 deletions(-) create mode 100644 app/s3_utils.py diff --git a/app/main.py b/app/main.py index 52715e5..ddf47cb 100644 --- a/app/main.py +++ b/app/main.py @@ -1,12 +1,9 @@ #!/usr/bin/env python3 import json import random -import os import io -import boto3 import zipfile import tempfile -from botocore.exceptions import ClientError from typing import List from fastapi import FastAPI, Request, Form from fastapi.responses import HTMLResponse, RedirectResponse, StreamingResponse @@ -19,6 +16,16 @@ import logging # Import the functions from gen_op.py from app.gen_op import generer_pdf_exercices_en_memoire +# Import S3 functions +from app.s3_utils import ( + S3_BUCKET_NAME, + create_bucket_if_not_exists, + upload_to_s3, + download_from_s3, + list_objects_in_s3, + delete_from_s3, +) + # Custom filter to exclude health check logs class HealthCheckFilter(logging.Filter): @@ -32,53 +39,9 @@ logging.getLogger("uvicorn.access").addFilter(HealthCheckFilter()) app = FastAPI() -# S3 Configuration -S3_BUCKET_NAME = os.environ.get("S3_BUCKET_NAME", "math-exercises") - templates = Jinja2Templates(directory="app/templates") -def get_s3_client(): - """Create and return an S3 client using environment variables""" - s3_access_key = os.environ.get("S3_ACCESS_KEY") - s3_secret_key = os.environ.get("S3_SECRET_KEY") - s3_host_base = os.environ.get("S3_HOST_BASE") - - if not all([s3_access_key, s3_secret_key, s3_host_base]): - raise ValueError("S3 environment variables not properly set") - - s3 = boto3.client( - "s3", - aws_access_key_id=s3_access_key, - aws_secret_access_key=s3_secret_key, - endpoint_url=s3_host_base, - region_name="us-east-1", # Required but unused for Infomaniak - ) - return s3 - - -def create_bucket_if_not_exists(bucket_name): - """Create S3 bucket if it doesn't exist""" - s3_client = get_s3_client() - - try: - s3_client.head_bucket(Bucket=bucket_name) - except ClientError as e: - error_code = int(e.response["Error"]["Code"]) - if error_code == 404: - # Bucket doesn't exist, create it - try: - s3_client.create_bucket(Bucket=bucket_name) - print(f"Bucket {bucket_name} created successfully") - except ClientError as create_error: - print(f"Error creating bucket: {create_error}") - raise - else: - # Some other error - print(f"Error checking bucket: {e}") - raise - - # Create bucket on startup try: create_bucket_if_not_exists(S3_BUCKET_NAME) @@ -86,67 +49,6 @@ except Exception as e: print(f"Warning: Could not create/check S3 bucket: {e}") -def upload_to_s3(file_data, bucket_name, object_name, content_type="application/pdf"): - """Upload file data to S3 bucket""" - s3_client = get_s3_client() - - try: - s3_client.put_object( - Bucket=bucket_name, - Key=object_name, - Body=file_data, - ContentType=content_type, - ) - return True - except ClientError as e: - print(f"Error uploading to S3: {e}") - return False - - -def download_from_s3(bucket_name, object_name): - """Download file data from S3 bucket""" - s3_client = get_s3_client() - - try: - response = s3_client.get_object(Bucket=bucket_name, Key=object_name) - return response["Body"].read() - except ClientError as e: - print(f"Error downloading from S3: {e}") - return None - - -def list_objects_in_s3(bucket_name): - """List all objects in S3 bucket (sorted from newest to oldest)""" - s3_client = get_s3_client() - - try: - response = s3_client.list_objects_v2(Bucket=bucket_name) - if "Contents" in response: - # Filter for PDF files only and sort by last modified (newest first) - pdf_files = [ - obj for obj in response["Contents"] if obj["Key"].endswith(".pdf") - ] - pdf_files.sort(key=lambda x: x["LastModified"], reverse=True) - return pdf_files - else: - return [] - except ClientError as e: - print(f"Error listing objects in S3: {e}") - return [] - - -def delete_from_s3(bucket_name, object_name): - """Delete file from S3 bucket""" - s3_client = get_s3_client() - - try: - s3_client.delete_object(Bucket=bucket_name, Key=object_name) - return True - except ClientError as e: - print(f"Error deleting from S3: {e}") - return False - - class MathExercisesPDF(FPDF): def header(self): self.set_font("Helvetica", "B", 16) @@ -175,7 +77,10 @@ class OperationExerciseRequest(BaseModel): def generate_exercises( - min_table: int, max_table: int, num_exercises: int = 15, multiplication_only: bool = False + min_table: int, + max_table: int, + num_exercises: int = 15, + multiplication_only: bool = False, ) -> List[str]: """Génère des exercices de multiplication et division aléatoires mélangés sans doublons""" exercises: List[str] = [] @@ -248,7 +153,10 @@ def generate_exercises( def create_math_exercises_pdf( - min_table: int, max_table: int, num_exercises: int = 15, multiplication_only: bool = False + min_table: int, + max_table: int, + num_exercises: int = 15, + multiplication_only: bool = False, ) -> str: """Crée un fichier PDF avec des exercices de mathématiques mélangés en 3 colonnes et l'upload sur S3""" import datetime @@ -284,7 +192,9 @@ def create_math_exercises_pdf( pdf.ln(5) # Générer les exercices - exercises = generate_exercises(min_table, max_table, num_exercises, multiplication_only) + exercises = generate_exercises( + min_table, max_table, num_exercises, multiplication_only + ) # Pas d'en-têtes de colonnes @@ -376,7 +286,10 @@ async def generate_exercises_endpoint(request: ExerciseRequest): return {"error": "Le nombre d'exercices doit être supérieur à 0"} pdf_filename = create_math_exercises_pdf( - request.min_table, request.max_table, request.num_exercises, request.multiplication_only + request.min_table, + request.max_table, + request.num_exercises, + request.multiplication_only, ) # Return redirect to automatically download the file diff --git a/app/s3_utils.py b/app/s3_utils.py new file mode 100644 index 0000000..ec9daa5 --- /dev/null +++ b/app/s3_utils.py @@ -0,0 +1,116 @@ +#!/usr/bin/env python3 +import os +import boto3 +from botocore.exceptions import ClientError +from typing import List, Optional + + +# S3 Configuration +S3_BUCKET_NAME = os.environ.get("S3_BUCKET_NAME", "math-exercises") + + +def get_s3_client(): + """Create and return an S3 client using environment variables""" + s3_access_key = os.environ.get("S3_ACCESS_KEY") + s3_secret_key = os.environ.get("S3_SECRET_KEY") + s3_host_base = os.environ.get("S3_HOST_BASE") + + if not all([s3_access_key, s3_secret_key, s3_host_base]): + raise ValueError("S3 environment variables not properly set") + + s3 = boto3.client( + "s3", + aws_access_key_id=s3_access_key, + aws_secret_access_key=s3_secret_key, + endpoint_url=s3_host_base, + region_name="us-east-1", # Required but unused for Infomaniak + ) + return s3 + + +def create_bucket_if_not_exists(bucket_name: str): + """Create S3 bucket if it doesn't exist""" + s3_client = get_s3_client() + + try: + s3_client.head_bucket(Bucket=bucket_name) + except ClientError as e: + error_code = int(e.response["Error"]["Code"]) + if error_code == 404: + # Bucket doesn't exist, create it + try: + s3_client.create_bucket(Bucket=bucket_name) + print(f"Bucket {bucket_name} created successfully") + except ClientError as create_error: + print(f"Error creating bucket: {create_error}") + raise + else: + # Some other error + print(f"Error checking bucket: {e}") + raise + + +def upload_to_s3( + file_data: bytes, + bucket_name: str, + object_name: str, + content_type: str = "application/pdf", +) -> bool: + """Upload file data to S3 bucket""" + s3_client = get_s3_client() + + try: + s3_client.put_object( + Bucket=bucket_name, + Key=object_name, + Body=file_data, + ContentType=content_type, + ) + return True + except ClientError as e: + print(f"Error uploading to S3: {e}") + return False + + +def download_from_s3(bucket_name: str, object_name: str) -> Optional[bytes]: + """Download file data from S3 bucket""" + s3_client = get_s3_client() + + try: + response = s3_client.get_object(Bucket=bucket_name, Key=object_name) + return response["Body"].read() + except ClientError as e: + print(f"Error downloading from S3: {e}") + return None + + +def list_objects_in_s3(bucket_name: str) -> List[dict]: + """List all objects in S3 bucket (sorted from newest to oldest)""" + s3_client = get_s3_client() + + try: + response = s3_client.list_objects_v2(Bucket=bucket_name) + if "Contents" in response: + # Filter for PDF files only and sort by last modified (newest first) + pdf_files = [ + obj for obj in response["Contents"] if obj["Key"].endswith(".pdf") + ] + pdf_files.sort(key=lambda x: x["LastModified"], reverse=True) + return pdf_files + else: + return [] + except ClientError as e: + print(f"Error listing objects in S3: {e}") + return [] + + +def delete_from_s3(bucket_name: str, object_name: str) -> bool: + """Delete file from S3 bucket""" + s3_client = get_s3_client() + + try: + s3_client.delete_object(Bucket=bucket_name, Key=object_name) + return True + except ClientError as e: + print(f"Error deleting from S3: {e}") + return False