FastAPI Quickstart

Learn how to integrate Rabata.io object storage with your FastAPI applications using aioboto3.

Introduction

This guide will help you integrate Rabata.io with your FastAPI application for storing and serving files. FastAPI is a modern, high-performance web framework for building APIs with Python based on standard Python type hints.

Since Rabata.io is S3-compatible, we can use the aioboto3 library to interact with it asynchronously from FastAPI applications, which is a better fit for FastAPI’s asynchronous nature.

Prerequisites

FastAPI application (0.68.0+)
Rabata.io account with access keys
A bucket created in your Rabata.io account

Installation

First, install the required packages:

$ pip install fastapi uvicorn aioboto3 python-multipart python-dotenv

These packages provide:

fastapi: The FastAPI framework
uvicorn: ASGI server for running FastAPI applications
aioboto3: Async AWS SDK for Python to interact with Rabata.io
python-multipart: For handling form data and file uploads
python-dotenv: For loading environment variables from a .env file

Configuration

Create a configuration file for your FastAPI application:

# config.py
import os
from pydantic import BaseSettings
from dotenv import load_dotenv

# Load environment variables from .env file
load_dotenv()

class Settings(BaseSettings):
    # FastAPI configuration
    app_name: str = "FastAPI Rabata.io Integration"
    
    # Rabata.io configuration
    rabata_access_key: str = os.environ.get("RABATA_ACCESS_KEY", "")
    rabata_secret_key: str = os.environ.get("RABATA_SECRET_KEY", "")
    rabata_bucket_name: str = os.environ.get("RABATA_BUCKET_NAME", "")
    rabata_endpoint_url: str = "https://s3.eu-west-1.rabata.io"
    rabata_region: str = "eu-west-1"
    
    # File upload configuration
    upload_folder: str = "uploads"
    max_content_length: int = 16 * 1024 * 1024  # 16MB max upload size
    allowed_extensions: set = {"txt", "pdf", "png", "jpg", "jpeg", "gif"}
    
    class Config:
        env_file = ".env"

# Create settings instance
settings = Settings()

Create a .env file to store your credentials:

# .env
RABATA_ACCESS_KEY=YOUR_RABATA_ACCESS_KEY
RABATA_SECRET_KEY=YOUR_RABATA_SECRET_KEY
RABATA_BUCKET_NAME=your-bucket-name

Security Note: Never commit your .env file to version control. Add it to your .gitignore file.

S3 Client Setup

Create a utility module to handle S3 operations:

# s3_utils.py
import aioboto3
from config import settings

def get_s3_client():
    """Create and return an S3 client configured for Rabata.io."""
    session = aioboto3.Session(
        aws_access_key_id=settings.rabata_access_key,
        aws_secret_access_key=settings.rabata_secret_key,
        region_name=settings.rabata_region
    )
    return session.client(
        's3',
        endpoint_url=settings.rabata_endpoint_url
    )

def get_s3_resource():
    """Create and return an S3 resource configured for Rabata.io."""
    session = aioboto3.Session(
        aws_access_key_id=settings.rabata_access_key,
        aws_secret_access_key=settings.rabata_secret_key,
        region_name=settings.rabata_region
    )
    return session.resource(
        's3',
        endpoint_url=settings.rabata_endpoint_url
    )

Basic FastAPI Application

Create your FastAPI application:

# main.py
from fastapi import FastAPI
from config import settings

app = FastAPI(title=settings.app_name)

@app.get("/")
async def root():
    return {"message": "Welcome to FastAPI with Rabata.io integration"}

Run your application:

$ uvicorn main:app --reload

Your API will be available at http://localhost:8000 and the automatic API documentation at http://localhost:8000/docs.

File Uploads

Create routes to handle file uploads to Rabata.io:

# main.py
import uuid
from fastapi import FastAPI, File, UploadFile, HTTPException, Depends
from fastapi.responses import JSONResponse
from typing import List
import io
from config import settings
from s3_utils import get_s3_client

app = FastAPI(title=settings.app_name)

def allowed_file(filename: str) -> bool:
    return '.' in filename and \
           filename.rsplit('.', 1)[1].lower() in settings.allowed_extensions

@app.post("/upload/")
async def upload_file(file: UploadFile = File(...)):
    """Upload a file to Rabata.io."""
    if not allowed_file(file.filename):
        raise HTTPException(status_code=400, detail="File type not allowed")
    
    # Secure the filename and generate a unique name
    filename = file.filename
    unique_filename = f"{uuid.uuid4().hex}_{filename}"
    
    # Read file content
    contents = await file.read()
    
    # Upload to Rabata.io
    async with get_s3_client() as s3_client:
        try:
            await s3_client.upload_fileobj(
                io.BytesIO(contents),
                settings.rabata_bucket_name,
                f"{settings.upload_folder}/{unique_filename}"
            )
            
            # Generate a URL for the uploaded file
            file_url = f"{settings.rabata_endpoint_url}/{settings.rabata_bucket_name}/{settings.upload_folder}/{unique_filename}"
            
            return {
                "filename": unique_filename,
                "content_type": file.content_type,
                "file_size": len(contents),
                "file_url": file_url
            }
        except Exception as e:
            raise HTTPException(status_code=500, detail=f"Error uploading file: {str(e)}")

@app.post("/upload-multiple/")
async def upload_multiple_files(files: List[UploadFile] = File(...)):
    """Upload multiple files to Rabata.io."""
    results = []
    errors = []
    
    async with get_s3_client() as s3_client:
        for file in files:
            if not allowed_file(file.filename):
                errors.append(f"File type not allowed: {file.filename}")
                continue
            
            # Secure the filename and generate a unique name
            filename = file.filename
            unique_filename = f"{uuid.uuid4().hex}_{filename}"
            
            # Read file content
            contents = await file.read()
            
            try:
                await s3_client.upload_fileobj(
                    io.BytesIO(contents),
                    settings.rabata_bucket_name,
                    f"{settings.upload_folder}/{unique_filename}"
                )
                
                # Generate a URL for the uploaded file
                file_url = f"{settings.rabata_endpoint_url}/{settings.rabata_bucket_name}/{settings.upload_folder}/{unique_filename}"
                
                results.append({
                    "filename": unique_filename,
                    "content_type": file.content_type,
                    "file_size": len(contents),
                    "file_url": file_url
                })
            except Exception as e:
                errors.append(f"Error uploading {file.filename}: {str(e)}")
    
    return {
        "uploaded_files": results,
        "errors": errors
    }

Listing Files

Create a route to list files from your Rabata.io bucket:

# main.py
@app.get("/files/")
async def list_files():
    """List all files in the upload folder."""
    async with get_s3_client() as s3_client:
        try:
            response = await s3_client.list_objects_v2(
                Bucket=settings.rabata_bucket_name,
                Prefix=settings.upload_folder + '/'
            )
            
            files = []
            if 'Contents' in response:
                for item in response['Contents']:
                    # Extract just the filename from the key
                    key = item['Key']
                    if key != settings.upload_folder + '/':  # Skip the directory itself
                        filename = key.split('/')[-1]
                        size = item['Size']
                        last_modified = item['LastModified'].isoformat()
                        
                        # Generate a URL for the file
                        file_url = f"{settings.rabata_endpoint_url}/{settings.rabata_bucket_name}/{key}"
                        
                        files.append({
                            "key": key,
                            "filename": filename,
                            "size": size,
                            "last_modified": last_modified,
                            "file_url": file_url
                        })
            
            return {"files": files}
        except Exception as e:
            raise HTTPException(status_code=500, detail=f"Error listing files: {str(e)}")

Downloading Files

Create a route to download files from your Rabata.io bucket:

# main.py
from fastapi.responses import StreamingResponse

@app.get("/files/{key:path}")
async def download_file(key: str):
    """Download a file from Rabata.io."""
    async with get_s3_client() as s3_client:
        try:
            # Get the file from Rabata.io
            response = await s3_client.get_object(
                Bucket=settings.rabata_bucket_name,
                Key=key
            )
            
            # Get the file content
            file_content = await response['Body'].read()
            
            # Get the filename from the key
            filename = key.split('/')[-1]
            
            # Create a file-like object from the content
            file_obj = io.BytesIO(file_content)
            
            # Return the file as a streaming response
            return StreamingResponse(
                file_obj,
                media_type=response.get('ContentType', 'application/octet-stream'),
                headers={
                    "Content-Disposition": f"attachment; filename={filename}"
                }
            )
        except Exception as e:
            raise HTTPException(status_code=404, detail=f"Error downloading file: {str(e)}")

Deleting Files

Create a route to delete files from your Rabata.io bucket:

# main.py
@app.delete("/files/{key:path}")
async def delete_file(key: str):
    """Delete a file from Rabata.io."""
    async with get_s3_client() as s3_client:
        try:
            # Delete the file from Rabata.io
            await s3_client.delete_object(
                Bucket=settings.rabata_bucket_name,
                Key=key
            )
            
            return {"message": "File successfully deleted", "key": key}
        except Exception as e:
            raise HTTPException(status_code=500, detail=f"Error deleting file: {str(e)}")

Advanced Usage

Dependency Injection

Use FastAPI’s dependency injection system to provide the S3 client:

# dependencies.py
from fastapi import Depends
from s3_utils import get_s3_client

async def get_s3():
    """Dependency to get the S3 client."""
    async with get_s3_client() as s3:
        yield s3

Use the dependency in your routes:

# main.py
from dependencies import get_s3
import aioboto3

@app.get("/buckets/")
async def list_buckets(s3_client = Depends(get_s3)):
    """List all buckets."""
    try:
        response = await s3_client.list_buckets()
        
        buckets = []
        for bucket in response['Buckets']:
            buckets.append({
                "name": bucket['Name'],
                "creation_date": bucket['CreationDate'].isoformat()
            })
        
        return {"buckets": buckets}
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Error listing buckets: {str(e)}")

Generating Presigned URLs

Create a route to generate presigned URLs for temporary access to private files:

# main.py
@app.get("/files/{key:path}/presigned")
async def generate_presigned_url(key: str, expiration: int = 3600):
    """Generate a presigned URL for temporary access to a file."""
    async with get_s3_client() as s3_client:
        try:
            url = await s3_client.generate_presigned_url(
                'get_object',
                Params={
                    'Bucket': settings.rabata_bucket_name,
                    'Key': key
                },
                ExpiresIn=expiration
            )
            
            return {
                "key": key,
                "presigned_url": url,
                "expires_in": expiration
            }
        except Exception as e:
            raise HTTPException(status_code=500, detail=f"Error generating presigned URL: {str(e)}")

Direct Browser Uploads

Create a route to generate presigned POST requests for direct browser-to-S3 uploads:

# main.py
@app.post("/direct-upload/")
async def direct_upload(filename: str):
    """Generate a presigned POST request for direct browser uploads."""
    if not allowed_file(filename):
        raise HTTPException(status_code=400, detail="File type not allowed")
    
    # Generate a unique filename
    unique_filename = f"{uuid.uuid4().hex}_{filename}"
    key = f"{settings.upload_folder}/{unique_filename}"
    
    async with get_s3_client() as s3_client:
        try:
            # Generate presigned POST data
            presigned_data = await s3_client.generate_presigned_post(
                Bucket=settings.rabata_bucket_name,
                Key=key,
                ExpiresIn=3600
            )
            
            return {
                "key": key,
                "presigned_post": presigned_data
            }
        except Exception as e:
            raise HTTPException(status_code=500, detail=f"Error generating presigned POST: {str(e)}")

CORS Configuration

Configure CORS for your FastAPI application:

# main.py
from fastapi.middleware.cors import CORSMiddleware

# Add CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # Allows all origins
    allow_credentials=True,
    allow_methods=["*"],  # Allows all methods
    allow_headers=["*"],  # Allows all headers
)

# For production, specify the exact origins:
# app.add_middleware(
#     CORSMiddleware,
#     allow_origins=["https://yourdomain.com"],
#     allow_credentials=True,
#     allow_methods=["GET", "POST", "PUT", "DELETE"],
#     allow_headers=["*"],
# )

Also configure CORS on your Rabata.io bucket for direct uploads:

[
  {
    "AllowedHeaders": [
      "*"
    ],
    "AllowedMethods": [
      "GET",
      "PUT",
      "POST",
      "DELETE"
    ],
    "AllowedOrigins": [
      "http://localhost:8000",
      "https://yourdomain.com"
    ],
    "ExposeHeaders": [
      "ETag",
      "Content-Length",
      "Content-Type"
    ],
    "MaxAgeSeconds": 3600
  }
]

Error Handling

Implement proper error handling for S3 operations:

# s3_utils.py
from botocore.exceptions import ClientError
from fastapi import HTTPException

def safe_s3_operation(operation_func):
    """Decorator for safely executing S3 operations with error handling."""
    async def wrapper(*args, **kwargs):
        try:
            return await operation_func(*args, **kwargs)
        except ClientError as e:
            error_code = e.response['Error']['Code']
            error_message = e.response['Error']['Message']
            
            if error_code == 'NoSuchKey':
                raise HTTPException(status_code=404, detail="File not found")
            elif error_code == 'NoSuchBucket':
                raise HTTPException(status_code=404, detail="Bucket not found")
            elif error_code == 'AccessDenied':
                raise HTTPException(status_code=403, detail="Access denied")
            else:
                raise HTTPException(status_code=500, detail=f"S3 error: {error_code} - {error_message}")
        except Exception as e:
            raise HTTPException(status_code=500, detail=f"Unexpected error: {str(e)}")
    return wrapper

Use the decorator in your routes:

# main.py
from s3_utils import safe_s3_operation

@app.get("/files/{key:path}/safe")
@safe_s3_operation
async def download_file_safe(key: str):
    """Download a file from Rabata.io with safe error handling."""
    async with get_s3_client() as s3_client:
        # Get the file from Rabata.io
        response = await s3_client.get_object(
            Bucket=settings.rabata_bucket_name,
            Key=key
        )
        
        # Get the file content
        file_content = await response['Body'].read()
        
        # Get the filename from the key
        filename = key.split('/')[-1]
        
        # Create a file-like object from the content
        file_obj = io.BytesIO(file_content)
        
        # Return the file as a streaming response
        return StreamingResponse(
            file_obj,
            media_type=response.get('ContentType', 'application/octet-stream'),
            headers={
                "Content-Disposition": f"attachment; filename={filename}"
            }
        )

Complete Example

Here’s a minimal but complete FastAPI application that integrates with Rabata.io:

# main.py
import io
import uuid
import aioboto3
from typing import List
from fastapi import FastAPI, File, UploadFile, HTTPException, Depends
from fastapi.responses import StreamingResponse
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseSettings
from dotenv import load_dotenv
import os

# Load environment variables
load_dotenv()

# Settings
class Settings(BaseSettings):
    app_name: str = "FastAPI Rabata.io Integration"
    rabata_access_key: str = os.environ.get("RABATA_ACCESS_KEY", "")
    rabata_secret_key: str = os.environ.get("RABATA_SECRET_KEY", "")
    rabata_bucket_name: str = os.environ.get("RABATA_BUCKET_NAME", "")
    rabata_endpoint_url: str = "https://s3.eu-west-1.rabata.io"
    rabata_region: str = "eu-west-1"
    upload_folder: str = "uploads"
    allowed_extensions: set = {"txt", "pdf", "png", "jpg", "jpeg", "gif"}

settings = Settings()

# Create FastAPI app
app = FastAPI(title=settings.app_name)

# Add CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# S3 client function
def get_s3_client():
    session = aioboto3.Session(
        aws_access_key_id=settings.rabata_access_key,
        aws_secret_access_key=settings.rabata_secret_key,
        region_name=settings.rabata_region
    )
    return session.client(
        's3',
        endpoint_url=settings.rabata_endpoint_url
    )

# Helper functions
def allowed_file(filename: str) -> bool:
    return '.' in filename and \
           filename.rsplit('.', 1)[1].lower() in settings.allowed_extensions

# Routes
@app.get("/")
async def root():
    return {"message": "Welcome to FastAPI with Rabata.io integration"}

@app.post("/upload/")
async def upload_file(file: UploadFile = File(...)):
    if not allowed_file(file.filename):
        raise HTTPException(status_code=400, detail="File type not allowed")
    
    filename = file.filename
    unique_filename = f"{uuid.uuid4().hex}_{filename}"
    contents = await file.read()
    
    async with get_s3_client() as s3_client:
        try:
            await s3_client.upload_fileobj(
                io.BytesIO(contents),
                settings.rabata_bucket_name,
                f"{settings.upload_folder}/{unique_filename}"
            )
            
            file_url = f"{settings.rabata_endpoint_url}/{settings.rabata_bucket_name}/{settings.upload_folder}/{unique_filename}"
            
            return {
                "filename": unique_filename,
                "content_type": file.content_type,
                "file_size": len(contents),
                "file_url": file_url
            }
        except Exception as e:
            raise HTTPException(status_code=500, detail=f"Error uploading file: {str(e)}")

@app.get("/files/")
async def list_files():
    async with get_s3_client() as s3_client:
        try:
            response = await s3_client.list_objects_v2(
                Bucket=settings.rabata_bucket_name,
                Prefix=settings.upload_folder + '/'
            )
            
            files = []
            if 'Contents' in response:
                for item in response['Contents']:
                    key = item['Key']
                    if key != settings.upload_folder + '/':
                        filename = key.split('/')[-1]
                        size = item['Size']
                        last_modified = item['LastModified'].isoformat()
                        
                        file_url = f"{settings.rabata_endpoint_url}/{settings.rabata_bucket_name}/{key}"
                        
                        files.append({
                            "key": key,
                            "filename": filename,
                            "size": size,
                            "last_modified": last_modified,
                            "file_url": file_url
                        })
            
            return {"files": files}
        except Exception as e:
            raise HTTPException(status_code=500, detail=f"Error listing files: {str(e)}")

@app.get("/files/{key:path}")
async def download_file(key: str):
    async with get_s3_client() as s3_client:
        try:
            response = await s3_client.get_object(
                Bucket=settings.rabata_bucket_name,
                Key=key
            )
            
            file_content = await response['Body'].read()
            filename = key.split('/')[-1]
            file_obj = io.BytesIO(file_content)
            
            return StreamingResponse(
                file_obj,
                media_type=response.get('ContentType', 'application/octet-stream'),
                headers={
                    "Content-Disposition": f"attachment; filename={filename}"
                }
            )
        except Exception as e:
            raise HTTPException(status_code=404, detail=f"Error downloading file: {str(e)}")

@app.delete("/files/{key:path}")
async def delete_file(key: str):
    async with get_s3_client() as s3_client:
        try:
            await s3_client.delete_object(
                Bucket=settings.rabata_bucket_name,
                Key=key
            )
            
            return {"message": "File successfully deleted", "key": key}
        except Exception as e:
            raise HTTPException(status_code=500, detail=f"Error deleting file: {str(e)}")

# Run the application
if __name__ == "__main__":
    import uvicorn
    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

Production Considerations

Performance Optimization

Use a CDN in front of Rabata.io for faster content delivery
Implement caching for frequently accessed files
Use background tasks for large file operations

# main.py
from fastapi import BackgroundTasks

@app.post("/upload-background/")
async def upload_file_background(
    background_tasks: BackgroundTasks,
    file: UploadFile = File(...)
):
    """Upload a file to Rabata.io in the background."""
    if not allowed_file(file.filename):
        raise HTTPException(status_code=400, detail="File type not allowed")
    
    filename = file.filename
    unique_filename = f"{uuid.uuid4().hex}_{filename}"
    contents = await file.read()
    
    # Define the background task
    async def upload_to_s3(file_content, key):
        async with get_s3_client() as s3_client:
            try:
                await s3_client.upload_fileobj(
                    io.BytesIO(file_content),
                    settings.rabata_bucket_name,
                    key
                )
                print(f"File uploaded: {key}")
            except Exception as e:
                print(f"Error uploading file: {str(e)}")
    
    # Add the task to the background tasks
    key = f"{settings.upload_folder}/{unique_filename}"
    background_tasks.add_task(upload_to_s3, contents, key)
    
    return {
        "message": "File upload started in background",
        "filename": unique_filename
    }

Security Best Practices

Use environment variables for credentials
Validate file types and sizes before uploading
Implement proper authentication and authorization for file access
Use private ACLs for sensitive files and generate presigned URLs when needed

Deployment

For deploying your FastAPI application with Rabata.io integration:

# Dockerfile
FROM python:3.9

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

# requirements.txt
fastapi==0.95.0
uvicorn==0.21.1
aioboto3==11.2.0
python-multipart==0.0.6
python-dotenv==1.0.0

Python Quickstart Django Integration Flask Integration