Using FastAPI to Build an S3-Compatible Cloud Storage API

FastAPI is a modern, fast (high-performance), web framework for building APIs with Python 3.7+. It is based on standard Python type hints and the Starlette framework. One of its standout features is the automatic generation of API documentation using Swagger UI, which makes development quicker and more intuitive. When combined with cloud storage systems like Amazon S3, FastAPI can be leveraged to build scalable and robust file management APIs.

In this article, we will demonstrate how to use FastAPI to build an API that interacts with S3-compatible cloud storage services. These services are often used for file storage and management, and many of them (like MinIO or Wasabi) offer similar functionality to Amazon’s S3. By the end of this tutorial, you’ll have a fully functional API that can upload, download, and manage files in S3-compatible cloud storage.

Setting Up the FastAPI Project

To get started, we need to set up a FastAPI project and install necessary dependencies. Create a virtual environment and install the required packages.

# Create a virtual environment
python -m venv venv

# Activate the virtual environment
# On Windows
venv\Scripts\activate
# On macOS/Linux
source venv/bin/activate

# Install FastAPI and Uvicorn for serving the app
pip install fastapi uvicorn

# Install boto3 for interacting with S3-compatible services
pip install boto3
Now that we have the basic environment set up, let’s move on to the code.

Creating the S3-Compatible Client

We will use the boto3 library, which is the Amazon Web Services (AWS) SDK for Python. It provides a simple API to interact with S3-compatible cloud storage systems.
python
import boto3
from botocore.exceptions import NoCredentialsError

# S3 client setup
def get_s3_client():
return boto3.client(
‘s3’,
endpoint_url=”https://s3.your-cloud-provider.com”, # Replace with your provider’s URL
aws_access_key_id=”your_access_key”, # Replace with your access key
aws_secret_access_key=”your_secret_key”, # Replace with your secret key
region_name=”us-east-1″ # Set the region accordingly
)

Uploading Files to S3-Compatible Storage

The first endpoint we’ll create allows users to upload files to our S3-compatible cloud storage. FastAPI’s File and UploadFile classes will be used to handle file uploads.
python
from fastapi import FastAPI, File, UploadFile
from fastapi.responses import JSONResponse

app = FastAPI()

@app.post(“/upload/”)
async def upload_file(file: UploadFile = File(…)):
s3_client = get_s3_client()
try:
# Upload the file to the S3 bucket
s3_client.upload_fileobj(
file.file,
“your-bucket-name”, # Replace with your S3 bucket name
file.filename,
ExtraArgs={“ACL”: “public-read”}
)
return JSONResponse(content={“message”: “File uploaded successfully!”}, status_code=200)
except NoCredentialsError:
return JSONResponse(content={“error”: “Credentials not available”}, status_code=400)
This endpoint will receive the uploaded file and store it in an S3 bucket. The upload_fileobj method is used to upload the file in chunks, making it efficient for large files.

Listing Files in S3-Compatible Storage

Next, let’s implement an endpoint that will list all the files stored in the S3 bucket. This can be useful for managing and retrieving files from the cloud storage.
python
@app.get(“/files/”)
async def list_files():
s3_client = get_s3_client()
try:
# List all objects in the bucket
response = s3_client.list_objects_v2(Bucket=”your-bucket-name”) # Replace with your S3 bucket name
files = [obj[“Key”] for obj in response.get(“Contents”, [])]
return {“files”: files}
except Exception as e:
return JSONResponse(content={“error”: str(e)}, status_code=400)
This code fetches a list of all files stored in the specified bucket and returns the list in the API response.

Downloading Files from S3-Compatible Storage

Another essential functionality is downloading files. FastAPI makes it easy to return files from your endpoints.
python
from fastapi.responses import StreamingResponse

@app.get(“/download/{filename}”)
async def download_file(filename: str):
s3_client = get_s3_client()
try:
# Get the file from the S3 bucket
file_stream = s3_client.get_object(Bucket=”your-bucket-name”, Key=filename)[“Body”]
return StreamingResponse(file_stream, media_type=”application/octet-stream”)
except Exception as e:
return JSONResponse(content={“error”: str(e)}, status_code=400)
This endpoint uses the StreamingResponse to stream the file from the S3-compatible storage to the client. It’s efficient for serving large files without loading them entirely into memory.

Handling Errors and Edge Cases

Error handling is crucial when interacting with cloud services. It’s important to account for potential issues, such as missing credentials, incorrect file types, or exceeding storage limits.
python
@app.exception_handler(Exception)
async def global_exception_handler(request, exc):
return JSONResponse(
status_code=500,
content={“message”: f”An unexpected error occurred: {str(exc)}”},
)
This global error handler will catch any unhandled exceptions and return a generic error message with a 500 status code.

Running the FastAPI Application

To run the application, use uvicorn, which is a fast ASGI server compatible with FastAPI.
uvicorn main:app –reload
This will start the FastAPI application, and it will be accessible at http://127.0.0.1:8000. You can now test the file upload, listing, and download functionalities through the Swagger UI, which is automatically generated by FastAPI at http://127.0.0.1:8000/docs.
html

Endpoints

• Upload File
• List Files
• Download File