Flask Quickstart

Learn how to integrate Rabata.io object storage with your Flask applications using Flask-S3.

Introduction

This guide will help you integrate Rabata.io with your Flask application for storing and serving static files and user uploads.

Since Rabata.io is S3-compatible, we can use the boto3 library to interact with it directly from Flask applications.

Prerequisites

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

Installation

First, install the required packages:

$ pip install flask boto3 python-dotenv

For file uploads, you might also want to install:

$ pip install flask-uploads

Or for newer Flask versions:

$ pip install Flask-Reuploaded

Configuration

Create a configuration file for your Flask application:

# config.py
import os
from dotenv import load_dotenv

# Load environment variables from .env file
load_dotenv()

class Config:
    # Flask configuration
    SECRET_KEY = os.environ.get('SECRET_KEY', 'your-secret-key')
    
    # Rabata.io configuration
    RABATA_ACCESS_KEY = os.environ.get('RABATA_ACCESS_KEY')
    RABATA_SECRET_KEY = os.environ.get('RABATA_SECRET_KEY')
    RABATA_BUCKET_NAME = os.environ.get('RABATA_BUCKET_NAME')
    RABATA_ENDPOINT_URL = 'https://s3.eu-west-1.rabata.io'
    RABATA_REGION = 'eu-west-1'
    
    # File upload configuration
    UPLOAD_FOLDER = 'uploads'
    MAX_CONTENT_LENGTH = 16 * 1024 * 1024  # 16MB max upload size
    ALLOWED_EXTENSIONS = {'txt', 'pdf', 'png', 'jpg', 'jpeg', 'gif'}

Create a .env file to store your credentials:

# .env
SECRET_KEY=your-flask-secret-key
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.

Basic Setup

Create a utility module to handle S3 operations:

# s3_utils.py
import boto3
from flask import current_app

def get_s3_client():
    """Create and return an S3 client configured for Rabata.io."""
    return boto3.client(
        's3',
        endpoint_url=current_app.config['RABATA_ENDPOINT_URL'],
        aws_access_key_id=current_app.config['RABATA_ACCESS_KEY'],
        aws_secret_access_key=current_app.config['RABATA_SECRET_KEY'],
        region_name=current_app.config['RABATA_REGION']
    )

def get_s3_resource():
    """Create and return an S3 resource configured for Rabata.io."""
    return boto3.resource(
        's3',
        endpoint_url=current_app.config['RABATA_ENDPOINT_URL'],
        aws_access_key_id=current_app.config['RABATA_ACCESS_KEY'],
        aws_secret_access_key=current_app.config['RABATA_SECRET_KEY'],
        region_name=current_app.config['RABATA_REGION']
    )

Initialize your Flask application with the configuration:

# app.py
from flask import Flask
from config import Config

def create_app(config_class=Config):
    app = Flask(__name__)
    app.config.from_object(config_class)
    
    # Register blueprints, extensions, etc.
    
    return app

app = create_app()

File Uploads

Create routes to handle file uploads to Rabata.io:

# routes.py
import os
import uuid
from flask import Blueprint, request, redirect, url_for, render_template, current_app, flash
from werkzeug.utils import secure_filename
from s3_utils import get_s3_client

bp = Blueprint('uploads', __name__)

def allowed_file(filename):
    return '.' in filename and \
           filename.rsplit('.', 1)[1].lower() in current_app.config['ALLOWED_EXTENSIONS']

@bp.route('/upload', methods=['GET', 'POST'])
def upload_file():
    if request.method == 'POST':
        # Check if the post request has the file part
        if 'file' not in request.files:
            flash('No file part')
            return redirect(request.url)
        
        file = request.files['file']
        
        # If user does not select file, browser also
        # submit an empty part without filename
        if file.filename == '':
            flash('No selected file')
            return redirect(request.url)
        
        if file and allowed_file(file.filename):
            # Secure the filename and generate a unique name
            filename = secure_filename(file.filename)
            unique_filename = f"{uuid.uuid4().hex}_{filename}"
            
            # Upload to Rabata.io
            s3_client = get_s3_client()
            
            try:
                s3_client.upload_fileobj(
                    file,
                    current_app.config['RABATA_BUCKET_NAME'],
                    f"{current_app.config['UPLOAD_FOLDER']}/{unique_filename}"
                )
                flash('File successfully uploaded')
                return redirect(url_for('uploads.list_files'))
            except Exception as e:
                flash(f'Error uploading file: {str(e)}')
                return redirect(request.url)
    
    return render_template('upload.html')

Create a template for the upload form:

<!-- templates/upload.html -->

{% extends 'base.html' %}

{% block content %}
<h1>Upload File</h1>
<form method="post" enctype="multipart/form-data">
    <div>
        <label for="file">Select file</label>
        <input type="file" name="file" id="file">
    </div>
    <button type="submit">Upload</button>
</form>
{% endblock %}

Listing Files

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

# routes.py
@bp.route('/files')
def list_files():
    s3_client = get_s3_client()
    
    try:
        response = s3_client.list_objects_v2(
            Bucket=current_app.config['RABATA_BUCKET_NAME'],
            Prefix=current_app.config['UPLOAD_FOLDER'] + '/'
        )
        
        files = []
        if 'Contents' in response:
            for item in response['Contents']:
                # Extract just the filename from the key
                key = item['Key']
                if key != current_app.config['UPLOAD_FOLDER'] + '/':  # Skip the directory itself
                    filename = key.split('/')[-1]
                    size = item['Size']
                    last_modified = item['LastModified']
                    
                    files.append({
                        'key': key,
                        'filename': filename,
                        'size': size,
                        'last_modified': last_modified
                    })
        
        return render_template('files.html', files=files)
    except Exception as e:
        flash(f'Error listing files: {str(e)}')
        return redirect(url_for('uploads.upload_file'))

Create a template to display the files:

<!-- templates/files.html -->

{% extends 'base.html' %}

{% block content %}
<h1>Files</h1>
<a href="{{ url_for('uploads.upload_file') }}">Upload New File</a>
<table>
    <thead>
        <tr>
            <th>Filename</th>
            <th>Size</th>
            <th>Last Modified</th>
            <th>Actions</th>
        </tr>
    </thead>
    <tbody>
        {% for file in files %}
        <tr>
            <td>{{ file.filename }}</td>
            <td>{{ file.size|filesizeformat }}</td>
            <td>{{ file.last_modified }}</td>
            <td>
                <a href="{{ url_for('uploads.download_file', key=file.key) }}">Download</a>
                <a href="{{ url_for('uploads.delete_file', key=file.key) }}">Delete</a>
            </td>
        </tr>
        {% else %}
        <tr>
            <td colspan="4">No files found.</td>
        </tr>
        {% endfor %}
    </tbody>
</table>
{% endblock %}

Downloading Files

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

# routes.py
from flask import send_file, abort
import io

@bp.route('/files/<path:key>/download')
def download_file(key):
    s3_client = get_s3_client()
    
    try:
        # Get the file from Rabata.io
        response = s3_client.get_object(
            Bucket=current_app.config['RABATA_BUCKET_NAME'],
            Key=key
        )
        
        # Get the file content
        file_content = 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)
        
        # Send the file to the user
        return send_file(
            file_obj,
            download_name=filename,
            as_attachment=True,
            mimetype=response.get('ContentType', 'application/octet-stream')
        )
    except Exception as e:
        abort(404, description=f"Error downloading file: {str(e)}")

Deleting Files

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

# routes.py
@bp.route('/files/<path:key>/delete')
def delete_file(key):
    s3_client = get_s3_client()
    
    try:
        # Delete the file from Rabata.io
        s3_client.delete_object(
            Bucket=current_app.config['RABATA_BUCKET_NAME'],
            Key=key
        )
        
        flash('File successfully deleted')
    except Exception as e:
        flash(f'Error deleting file: {str(e)}')
    
    return redirect(url_for('uploads.list_files'))

Serving Static Files

For serving static files (CSS, JavaScript, images) from Rabata.io:

# static_utils.py
import os
from flask import current_app
from s3_utils import get_s3_client

def upload_static_folder_to_s3(local_path='static', s3_prefix='static'):
    """Upload the static folder to Rabata.io."""
    s3_client = get_s3_client()
    bucket = current_app.config['RABATA_BUCKET_NAME']
    
    for root, dirs, files in os.walk(local_path):
        for file in files:
            local_file_path = os.path.join(root, file)
            
            # Calculate the S3 key
            relative_path = os.path.relpath(local_file_path, local_path)
            s3_key = f"{s3_prefix}/{relative_path}"
            
            # Upload the file
            s3_client.upload_file(
                local_file_path,
                bucket,
                s3_key
            )
            
            print(f"Uploaded {local_file_path} to {s3_key}")

Add a command to your Flask application to upload static files:

# commands.py
import click
from flask.cli import with_appcontext
from static_utils import upload_static_folder_to_s3

@click.command('upload-static')
@with_appcontext
def upload_static_command():
    """Upload static files to Rabata.io."""
    click.echo('Uploading static files...')
    upload_static_folder_to_s3()
    click.echo('Static files uploaded!')

# app.py
from commands import upload_static_command

def create_app(config_class=Config):
    app = Flask(__name__)
    app.config.from_object(config_class)
    
    # Register blueprints, extensions, etc.
    
    # Register commands
    app.cli.add_command(upload_static_command)
    
    return app

Run the command to upload your static files:

$ flask upload-static

Advanced Usage

Generating Presigned URLs

For private files that require temporary access:

# s3_utils.py
def generate_presigned_url(key, expiration=3600):
    """Generate a presigned URL for an object in S3."""
    s3_client = get_s3_client()
    
    url = s3_client.generate_presigned_url(
        'get_object',
        Params={
            'Bucket': current_app.config['RABATA_BUCKET_NAME'],
            'Key': key
        },
        ExpiresIn=expiration
    )
    
    return url

Usage in a route:

# routes.py
@bp.route('/files/<path:key>/share')
def share_file(key):
    try:
        # Generate a presigned URL
        url = generate_presigned_url(key, expiration=3600)  # 1 hour
        
        return render_template('share.html', url=url, filename=key.split('/')[-1])
    except Exception as e:
        flash(f'Error generating sharing link: {str(e)}')
        return redirect(url_for('uploads.list_files'))

Direct Browser Uploads

For large files, you can implement direct browser-to-S3 uploads:

# s3_utils.py
def generate_presigned_post(key, expiration=3600):
    """Generate a presigned POST request for direct browser uploads."""
    s3_client = get_s3_client()
    
    response = s3_client.generate_presigned_post(
        Bucket=current_app.config['RABATA_BUCKET_NAME'],
        Key=key,
        ExpiresIn=expiration
    )
    
    return response

Usage in a route:

# routes.py
import json

@bp.route('/direct-upload', methods=['GET', 'POST'])
def direct_upload():
    if request.method == 'POST':
        filename = request.form.get('filename')
        if not filename:
            flash('No filename provided')
            return redirect(request.url)
        
        if allowed_file(filename):
            # Secure the filename and generate a unique name
            filename = secure_filename(filename)
            unique_filename = f"{uuid.uuid4().hex}_{filename}"
            key = f"{current_app.config['UPLOAD_FOLDER']}/{unique_filename}"
            
            # Generate presigned POST data
            presigned_data = generate_presigned_post(key)
            
            return json.dumps(presigned_data)
        else:
            flash('File type not allowed')
            return redirect(request.url)
    
    return render_template('direct_upload.html')

CORS Configuration

If you’re using direct uploads or accessing files from a different domain, you’ll need to configure CORS on your Rabata.io bucket:

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

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

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

Error Handling

Implement proper error handling for S3 operations:

# s3_utils.py
from botocore.exceptions import ClientError

def safe_s3_operation(operation_func):
    """Decorator for safely executing S3 operations with error handling."""
    def wrapper(*args, **kwargs):
        try:
            return operation_func(*args, **kwargs)
        except ClientError as e:
            error_code = e.response['Error']['Code']
            error_message = e.response['Error']['Message']
            current_app.logger.error(f"S3 error: {error_code} - {error_message}")
            raise
        except Exception as e:
            current_app.logger.error(f"Unexpected error: {str(e)}")
            raise
    return wrapper

@safe_s3_operation
def upload_file(file_obj, key):
    """Safely upload a file to S3."""
    s3_client = get_s3_client()
    s3_client.upload_fileobj(
        file_obj,
        current_app.config['RABATA_BUCKET_NAME'],
        key
    )
    return True

Complete Example

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

# app.py
import os
import uuid
from flask import Flask, request, redirect, url_for, render_template, flash, send_file, abort
from werkzeug.utils import secure_filename
import boto3
import io
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

app = Flask(__name__)

# Configuration
app.config['SECRET_KEY'] = os.environ.get('SECRET_KEY', 'your-secret-key')
app.config['RABATA_ACCESS_KEY'] = os.environ.get('RABATA_ACCESS_KEY')
app.config['RABATA_SECRET_KEY'] = os.environ.get('RABATA_SECRET_KEY')
app.config['RABATA_BUCKET_NAME'] = os.environ.get('RABATA_BUCKET_NAME')
app.config['RABATA_ENDPOINT_URL'] = 'https://s3.eu-west-1.rabata.io'
app.config['RABATA_REGION'] = 'eu-west-1'
app.config['UPLOAD_FOLDER'] = 'uploads'
app.config['MAX_CONTENT_LENGTH'] = 16 * 1024 * 1024  # 16MB max upload size
app.config['ALLOWED_EXTENSIONS'] = {'txt', 'pdf', 'png', 'jpg', 'jpeg', 'gif'}

# S3 client
def get_s3_client():
    return boto3.client(
        's3',
        endpoint_url=app.config['RABATA_ENDPOINT_URL'],
        aws_access_key_id=app.config['RABATA_ACCESS_KEY'],
        aws_secret_access_key=app.config['RABATA_SECRET_KEY'],
        region_name=app.config['RABATA_REGION']
    )

def allowed_file(filename):
    return '.' in filename and \
           filename.rsplit('.', 1)[1].lower() in app.config['ALLOWED_EXTENSIONS']

@app.route('/')
def index():
    return render_template('index.html')

@app.route('/upload', methods=['GET', 'POST'])
def upload_file():
    if request.method == 'POST':
        if 'file' not in request.files:
            flash('No file part')
            return redirect(request.url)
        
        file = request.files['file']
        
        if file.filename == '':
            flash('No selected file')
            return redirect(request.url)
        
        if file and allowed_file(file.filename):
            filename = secure_filename(file.filename)
            unique_filename = f"{uuid.uuid4().hex}_{filename}"
            
            s3_client = get_s3_client()
            
            try:
                s3_client.upload_fileobj(
                    file,
                    app.config['RABATA_BUCKET_NAME'],
                    f"{app.config['UPLOAD_FOLDER']}/{unique_filename}"
                )
                flash('File successfully uploaded')
                return redirect(url_for('list_files'))
            except Exception as e:
                flash(f'Error uploading file: {str(e)}')
                return redirect(request.url)
    
    return render_template('upload.html')

@app.route('/files')
def list_files():
    s3_client = get_s3_client()
    
    try:
        response = s3_client.list_objects_v2(
            Bucket=app.config['RABATA_BUCKET_NAME'],
            Prefix=app.config['UPLOAD_FOLDER'] + '/'
        )
        
        files = []
        if 'Contents' in response:
            for item in response['Contents']:
                key = item['Key']
                if key != app.config['UPLOAD_FOLDER'] + '/':
                    filename = key.split('/')[-1]
                    size = item['Size']
                    last_modified = item['LastModified']
                    
                    files.append({
                        'key': key,
                        'filename': filename,
                        'size': size,
                        'last_modified': last_modified
                    })
        
        return render_template('files.html', files=files)
    except Exception as e:
        flash(f'Error listing files: {str(e)}')
        return redirect(url_for('upload_file'))

@app.route('/files/<path:key>/download')
def download_file(key):
    s3_client = get_s3_client()
    
    try:
        response = s3_client.get_object(
            Bucket=app.config['RABATA_BUCKET_NAME'],
            Key=key
        )
        
        file_content = response['Body'].read()
        filename = key.split('/')[-1]
        file_obj = io.BytesIO(file_content)
        
        return send_file(
            file_obj,
            download_name=filename,
            as_attachment=True,
            mimetype=response.get('ContentType', 'application/octet-stream')
        )
    except Exception as e:
        abort(404, description=f"Error downloading file: {str(e)}")

@app.route('/files/<path:key>/delete')
def delete_file(key):
    s3_client = get_s3_client()
    
    try:
        s3_client.delete_object(
            Bucket=app.config['RABATA_BUCKET_NAME'],
            Key=key
        )
        
        flash('File successfully deleted')
    except Exception as e:
        flash(f'Error deleting file: {str(e)}')
    
    return redirect(url_for('list_files'))

if __name__ == '__main__':
    app.run(debug=True)

Python Quickstart Django Integration FastAPI Integration