Create A File Management Application

In this tutorial, we will guide you through creating a basic HTTP API server for a file management application using Python, the FastAPI HTTP API framework, and the AWS SDK Boto3. This project will enable you to perform common file operations like uploading, downloading, and deleting files in a cloud-based storage system compatible with the S3 protocol.

There's more on GitHub. Look at the complete example and learn how to set up and run in this repo (opens in a new tab).

Prerequisites

Before we begin, ensure you have the following prerequisites:

Python installed on your machine (preferably Python 3.9 or higher). We are using Poetry (opens in a new tab) as the package manager for the app.
Basic knowledge of Python and web development concepts.
An AIOZ Web3 Storage account with access grants.

Setting Up Your Development Environment

Create a project directory and set up your basic project structure.

poetry new file-manager

Installing Python, Poetry. Then install packages.

poetry add fastapi "uvicorn[standard]" boto3 python-multipart

Initialize FastAPI and S3 client

Create main.py in the folderfile_manager. Initialize FastAPI , boto3 client and import necessary libraries.

import os
import re
 
import boto3
import botocore
from fastapi import FastAPI, HTTPException, UploadFile
 
app = FastAPI()
 
# Initialize AWS S3 client
s3_resource = boto3.resource(
    "s3",
    region_name="us-east-1",
    # Get the credentials from the environment variables.
    aws_access_key_id=os.environ["ACCESS_KEY"],
    aws_secret_access_key=os.environ["SECRET_KEY"],
    endpoint_url=os.environ["ENDPOINT_URL"],
)
bucket_name = "file-manager"  # Replace with your bucket name
bucket = s3_resource.Bucket(bucket_name)

You have to set ACCESS_KEY, SECRET_KEY, and ENDPOINT_URL environment variables as your AIOZ W3S access key id, secret access key, and endpoint URL respectively.

Create an upload endpoint

Create an upload endpoint /upload/ to upload a file.

@app.post("/upload/")
async def upload_file(file: UploadFile, file_name: str, return_url: bool = False):
    # Validate the file name to prevent security vulnerabilities
    if not re.match(r"^[a-zA-Z0-9_-]+\.[a-zA-Z0-9]+$", file_name):
        raise HTTPException(status_code=400, detail="Invalid file name")
 
    # Upload the file to the specified bucket
    obj = bucket.Object(file_name)
    obj.put(Body=file.file)
 
    if return_url:
        # Generate a signed URL for downloading the file (if necessary)
        presigned_url = s3_resource.meta.client.generate_presigned_url(
            "get_object",
            Params={"Bucket": bucket_name, "Key": file_name},
            ExpiresIn=EXPIRATION_TIME,
        )
        return {"message": "File uploaded successfully", "file_url": presigned_url}
    return {"message": "File uploaded successfully"}

We set a constant EXPIRATION_TIME for the expiration time of the pre-signed URL. If you set too long, it will not be secured.

EXPIRATION_TIME = 600  # 10 minutes, in seconds, adjusted as needed.

Let's break down the provided code into smaller segments to examine each part individually.

Create an endpoint for uploading files. Ensure that you validate the file name to prevent security vulnerabilities:

@app.post("/upload/")
async def upload_file(file: UploadFile, file_name: str, return_url: bool = False):
    if not re.match(r"^[a-zA-Z0-9_-]+\.[a-zA-Z0-9]+$", file_name):
        raise HTTPException(status_code=400, detail="Invalid file name")

Use Boto3 to upload the file to the specified bucket:

    obj = bucket.Object(file_name)
    obj.put(Body=file.file)

If the return_url parameter is set to True, generate a signed URL for downloading the file:

    if return_url:
        presigned_url = s3_resource.meta.client.generate_presigned_url(
            "get_object",
            Params={"Bucket": bucket_name, "Key": file_name},
            ExpiresIn=EXPIRATION_TIME,
        )
        return {"message": "File uploaded successfully", "file_url": presigned_url}

Finally, respond with a success message:

    return {"message": "File uploaded successfully"}

Create a download endpoint

Create an upload endpoint /download/{filename} to download a file.

@app.get("/download/{file_name}")
async def download_file(file_name: str):
    try:
        # Check if the file exists in the specified bucket
        obj = s3_resource.Object(bucket_name, file_name)
        obj.load()
 
        # Generate a signed URL for downloading the file (if necessary)
        # You may not need this step if your storage provider allows public access.
        presigned_url = s3_resource.meta.client.generate_presigned_url(
            "get_object",
            Params={"Bucket": bucket_name, "Key": file_name},
            ExpiresIn=EXPIRATION_TIME,
        )
 
        return {"download_url": presigned_url}
    except botocore.exceptions.ClientError as e:
        # If the file does not exist, return a 404 error to the client
        if e.response["Error"]["Code"] == "404":
            raise HTTPException(status_code=404, detail="File not found")
        # Otherwise, return a generic error message to the client with a status code of 500
        raise HTTPException(status_code=500, detail="Failed to download file")

Let's break down the provided code into smaller segments to examine each part individually.

Create an endpoint for downloading files. Here, we check if the file exists in the specified bucket:

@app.get("/download/{file_name}")
async def download_file(file_name: str):
    try:
        obj = s3_resource.Object(bucket_name, file_name)
        obj.load()

Generate a signed URL for downloading the file. Note that this step may not be necessary if your storage provider allows public access:

        presigned_url = s3_resource.meta.client.generate_presigned_url(
            "get_object",
            Params={"Bucket": bucket_name, "Key": file_name},
            ExpiresIn=EXPIRATION_TIME,
        )

Respond with the generated download URL:

        return {"download_url": presigned_url}

Handle errors gracefully. If the file does not exist, return a 404 error to the client. For other errors, return a generic error message with a status code of 500:

    except botocore.exceptions.ClientError as e:
        if e.response["Error"]["Code"] == "404":
            raise HTTPException(status_code=404, detail="File not found")
        raise HTTPException(status_code=500, detail="Failed to download file")

Create a file listing endpoint

Create an upload endpoint /list/ to list files in the bucket.

@app.get("/list/")
async def list_files():
    files = [file.key for file in bucket.objects.all()]
    return {"files": files}

Create a file listing endpoint

Create an upload endpoint /delete/{file_name} to delete a file.

@app.delete("/delete/{file_name}")
async def delete_file(file_name: str):
    # Delete the specified object from the bucket
    try:
        obj = bucket.Object(file_name)
        obj.delete()
    except Exception as e:
        return {"message": "Failed to delete file", "error": str(e)}
 
    return {"message": "File deleted successfully"}

PreviousManage Billing NextMigrate From Amazon S3