Building a Custom Cloud Storage Platform with Python, Flask, and PostgreSQL

Architecting a cloud storage platform from scratch introduces significant engineering challenges spanning binary data management, recursive schema design, and secure authentication flows. To address these challenges and eliminate dependence on commercial cloud storage providers, I developed FileFlow, a robust web application built using Python, Flask, and PostgreSQL.

This article provides an in-depth analysis of the application's underlying architecture, the database schema design leveraged to support deeply nested directory structures, and the DevOps considerations necessary for deploying stateful file engines to ephemeral cloud infrastructure.

Architectural Overview and Technology Stack

To ensure maintainability and high performance, the technology stack was minimized to industry-standard components:

• Web Framework: Flask (Python 3.11)
• Database Layer: PostgreSQL accessed via SQLAlchemy
• Authentication: Flask-Login combined with Flask-Bcrypt
• Storage Layer: AWS S3 via Boto3 (for binary blobs)
• Application Server: Gunicorn WSGI

The system relies heavily on the Flask-SQLAlchemy integration to manage entity relationships and atomic operations reliably.

Modeling Hierarchical Filesystems

The crux of any cloud storage application is effectively mapping a traditional filesystem tree into a relational database. Early prototypes often utilize separate Folder and File tables, leading to complex junction logic and inefficient queries.

In FileFlow, I opted for a Unified Entity Model utilizing a self-referential adjacency list layout.

Copy
python
# models.py
from datetime import datetime
from extensions import db

class FileNode(db.Model):
    __tablename__ = 'file_nodes'
    
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(255), nullable=False)
    user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
    
    # Hierarchical implementation (Adjacency List)
    is_folder = db.Column(db.Boolean, default=False, nullable=False)
    parent_id = db.Column(db.Integer, db.ForeignKey('file_nodes.id'), nullable=True)
    
    # Self-referential relationship to easily access children
    children = db.relationship(
        'FileNode', 
        backref=db.backref('parent', remote_side=[id]),
        cascade='all, delete-orphan'
    )
    
    # Metadata fields
    s3_key = db.Column(db.String(500), nullable=True) # None if is_folder=True
    filesize = db.Column(db.BigInteger, default=0)
    mimetype = db.Column(db.String(100))
    file_hash = db.Column(db.String(64)) # SHA-256 for deduplication
    is_favorite = db.Column(db.Boolean, default=False)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)

    __table_args__ = (
        # Ensure filenames are unique within a specific folder for a specific user
        db.UniqueConstraint('user_id', 'parent_id', 'name', name='uix_user_parent_name'),
    )

Navigating the Hierarchy with Recursive CTEs

The beauty of the adjacency list model is its simplicity, but querying the full path of a deeply nested file (e.g., Root -> Documents -> Work -> Reports -> Q1.pdf) typically requires multiple SQL queries (the N+1 problem).

To solve this efficiently, I utilized PostgreSQL's Recursive Common Table Expressions (CTEs) via SQLAlchemy to fetch the entire breadcrumb trail in a single database round-trip:

Copy
python
# utils.py
from sqlalchemy import text
from models import FileNode

def get_breadcrumb_path(node_id: int):
    """
    Executes a recursive CTE to traverse up the tree and return the full path.
    """
    sql = text("""
        WITH RECURSIVE path_cte AS (
            -- Base case: The target node
            SELECT id, name, parent_id, 1 as depth
            FROM file_nodes
            WHERE id = :node_id
            
            UNION ALL
            
            -- Recursive step: Join parent to child
            SELECT parent.id, parent.name, parent.parent_id, child.depth + 1
            FROM file_nodes parent
            JOIN path_cte child ON child.parent_id = parent.id
        )
        SELECT id, name FROM path_cte ORDER BY depth DESC;
    """)
    
    result = db.session.execute(sql, {'node_id': node_id}).fetchall()
    return [{"id": row.id, "name": row.name} for row in result]

Binary Storage Strategy: Boto3 and S3

While the database holds the metadata and virtual directory structure, the actual binary files are stored in Amazon S3. This decoupling is essential for horizontal scaling.

When a user uploads a file, the Flask backend computes a SHA-256 hash of the file chunk stream. If the hash already exists in the database (even from another user), the system simply creates a new FileNode pointing to the existing s3_key - achieving zero-cost deduplication instantly.

Copy
python
# storage.py
import hashlib
import boto3
from werkzeug.datastructures import FileStorage

s3_client = boto3.client('s3', region_name='us-east-1')
BUCKET_NAME = 'fileflow-production-blobs'

def upload_and_deduplicate(file_obj: FileStorage, user_id: int, parent_id: int):
    # 1. Compute Hash in memory via chunks
    sha256 = hashlib.sha256()
    for chunk in iter(lambda: file_obj.read(4096), b""):
        sha256.update(chunk)
    file_hash = sha256.hexdigest()
    
    # Reset pointer for S3 upload
    file_obj.seek(0)
    
    # 2. Check for Deduplication
    existing_file = FileNode.query.filter_by(file_hash=file_hash, is_folder=False).first()
    
    if existing_file:
        s3_key = existing_file.s3_key
    else:
        # 3. Upload to S3 using the hash as the unique object key
        s3_key = f"blobs/{file_hash}"
        s3_client.upload_fileobj(file_obj, BUCKET_NAME, s3_key)
        
    # 4. Create Metadata Node
    new_node = FileNode(
        name=file_obj.filename,
        user_id=user_id,
        parent_id=parent_id,
        s3_key=s3_key,
        file_hash=file_hash,
        filesize=file_obj.content_length,
        mimetype=file_obj.mimetype
    )
    db.session.add(new_node)
    db.session.commit()
    
    return new_node

Secure Resource Sharing

File sharing required public endpoints with strict limitations. The ShareLink table generates cryptographically secure, unique tokens mapping back to specific file_nodes.id references. By injecting an expires_at column and an access_count tracker, we natively support time-bombed and tightly restricted access URLs, eliminating unauthorized public exposure of binary streams.

Copy
python
import secrets
from datetime import datetime, timedelta

def generate_share_link(node_id: int, expiry_days: int = 7):
    token = secrets.token_urlsafe(32)
    expiry = datetime.utcnow() + timedelta(days=expiry_days)
    
    share = ShareLink(
        file_node_id=node_id,
        token=token,
        expires_at=expiry
    )
    db.session.add(share)
    db.session.commit()
    
    return f"https://fileflow.amitdevx.tech/share/{token}"

Deployment: Conquering Ephemeral Storage on Render

Deploying stateful, file-dependent systems on modern Platform-as-a-Service (PaaS) providers like Render introduces the challenge of ephemeral storage. When an instance sleeps or is redeployed, the local filesystem state resets completely.

To accommodate this, by decoupling the s3_key in the actual database from arbitrary physical root directory bindings, migrating from local testing to S3 requires adjusting the environment configurations rather than tearing down the database mappings.

Copy
yaml
# render.yaml
services:
  - type: web
    name: fileflow-app
    env: python
    region: oregon
    buildCommand: "./build.sh"
    startCommand: "gunicorn backend.app:app -w 4 -k gevent"
    envVars:
      - key: FLASK_ENV
        value: production
      - key: DATABASE_URL
        fromDatabase:
          name: fileflow-db
          property: connectionString

Authentication: Flask-Login and Flask-Bcrypt

Every cloud storage application must have airtight authentication. FileFlow uses Flask-Login for session management and Flask-Bcrypt for password hashing.

User Model with Password Hashing

Copy
python
# models.py
from flask_login import UserMixin
from extensions import db, bcrypt

class User(UserMixin, db.Model):
    __tablename__ = 'users'
    
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)
    password_hash = db.Column(db.String(128), nullable=False)
    storage_quota = db.Column(db.BigInteger, default=10 * 1024 * 1024 * 1024)  # 10GB default
    storage_used = db.Column(db.BigInteger, default=0)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    
    # Relationship to files
    files = db.relationship('FileNode', backref='owner', lazy='dynamic')
    
    def set_password(self, password: str):
        """Hash the password using bcrypt with automatic salting."""
        self.password_hash = bcrypt.generate_password_hash(password).decode('utf-8')
    
    def check_password(self, password: str) -> bool:
        """Verify a password against the stored hash."""
        return bcrypt.check_password_hash(self.password_hash, password)
    
    @property
    def storage_percentage(self) -> float:
        """Return storage utilization as a percentage."""
        if self.storage_quota == 0:
            return 100.0
        return (self.storage_used / self.storage_quota) * 100

Registration and Login Routes

Copy
python
# routes/auth.py
from flask import Blueprint, render_template, redirect, url_for, flash, request
from flask_login import login_user, logout_user, login_required, current_user
from models import User
from extensions import db

auth_bp = Blueprint('auth', __name__)

@auth_bp.route('/register', methods=['GET', 'POST'])
def register():
    if current_user.is_authenticated:
        return redirect(url_for('dashboard.index'))
    
    if request.method == 'POST':
        username = request.form.get('username', '').strip()
        email = request.form.get('email', '').strip().lower()
        password = request.form.get('password', '')
        
        # Validation
        if len(password) < 8:
            flash('Password must be at least 8 characters.', 'error')
            return render_template('auth/register.html')
        
        if User.query.filter_by(email=email).first():
            flash('Email already registered.', 'error')
            return render_template('auth/register.html')
        
        # Create user
        user = User(username=username, email=email)
        user.set_password(password)
        db.session.add(user)
        db.session.commit()
        
        login_user(user)
        flash('Account created successfully.', 'success')
        return redirect(url_for('dashboard.index'))
    
    return render_template('auth/register.html')

@auth_bp.route('/login', methods=['GET', 'POST'])
def login():
    if current_user.is_authenticated:
        return redirect(url_for('dashboard.index'))
    
    if request.method == 'POST':
        email = request.form.get('email', '').strip().lower()
        password = request.form.get('password', '')
        
        user = User.query.filter_by(email=email).first()
        
        if user and user.check_password(password):
            login_user(user, remember=True)
            next_page = request.args.get('next')
            return redirect(next_page or url_for('dashboard.index'))
        
        flash('Invalid email or password.', 'error')
    
    return render_template('auth/login.html')

Core Flask Routes: File Operations

The heart of FileFlow is the file management API. Every operation (upload, download, move, delete, rename) goes through Flask route handlers that enforce ownership, validate quotas, and maintain database consistency.

File Upload with Quota Enforcement

Copy
python
# routes/files.py
from flask import Blueprint, request, jsonify, abort
from flask_login import login_required, current_user
from models import FileNode
from storage import upload_and_deduplicate
from extensions import db

files_bp = Blueprint('files', __name__)

@files_bp.route('/upload', methods=['POST'])
@login_required
def upload_file():
    """Handle single or multi-file uploads with quota enforcement."""
    if 'file' not in request.files:
        return jsonify({'error': 'No file provided'}), 400
    
    parent_id = request.form.get('parent_id', type=int)
    
    # Validate parent folder belongs to current user
    if parent_id:
        parent = FileNode.query.filter_by(
            id=parent_id, 
            user_id=current_user.id, 
            is_folder=True
        ).first()
        if not parent:
            return jsonify({'error': 'Invalid parent folder'}), 404
    
    uploaded_files = request.files.getlist('file')
    results = []
    
    for file_obj in uploaded_files:
        # Check storage quota before each upload
        file_size = file_obj.seek(0, 2)  # Seek to end to get size
        file_obj.seek(0)  # Reset pointer
        
        if current_user.storage_used + file_size > current_user.storage_quota:
            results.append({
                'name': file_obj.filename,
                'error': 'Storage quota exceeded',
            })
            continue
        
        # Upload and deduplicate
        node = upload_and_deduplicate(file_obj, current_user.id, parent_id)
        
        # Update user's storage counter
        current_user.storage_used += node.filesize
        db.session.commit()
        
        results.append({
            'name': node.name,
            'id': node.id,
            'size': node.filesize,
            'hash': node.file_hash,
        })
    
    return jsonify({'uploaded': results}), 201

Creating Folders

Copy
python
@files_bp.route('/folder', methods=['POST'])
@login_required
def create_folder():
    """Create a new folder within the user's file tree."""
    data = request.get_json()
    name = data.get('name', '').strip()
    parent_id = data.get('parent_id')
    
    if not name:
        return jsonify({'error': 'Folder name is required'}), 400
    
    # Check for name collision within the same parent
    existing = FileNode.query.filter_by(
        user_id=current_user.id,
        parent_id=parent_id,
        name=name,
        is_folder=True
    ).first()
    
    if existing:
        return jsonify({'error': f'Folder "{name}" already exists here'}), 409
    
    folder = FileNode(
        name=name,
        user_id=current_user.id,
        parent_id=parent_id,
        is_folder=True,
    )
    db.session.add(folder)
    db.session.commit()
    
    return jsonify({'id': folder.id, 'name': folder.name}), 201

File Download with Streaming

For large files, streaming the response prevents memory exhaustion on the server:

Copy
python
# routes/download.py
import boto3
from flask import Blueprint, Response, abort
from flask_login import login_required, current_user
from models import FileNode

download_bp = Blueprint('download', __name__)
s3_client = boto3.client('s3', region_name='us-east-1')
BUCKET_NAME = 'fileflow-production-blobs'

@download_bp.route('/download/<int:file_id>')
@login_required
def download_file(file_id: int):
    """Stream a file download directly from S3 to the client."""
    node = FileNode.query.filter_by(
        id=file_id, 
        user_id=current_user.id,
        is_folder=False
    ).first_or_404()
    
    # Get the S3 object as a streaming body
    s3_response = s3_client.get_object(
        Bucket=BUCKET_NAME,
        Key=node.s3_key
    )
    
    def generate():
        """Yield 8KB chunks from S3 to avoid loading entire file into memory."""
        for chunk in s3_response['Body'].iter_chunks(chunk_size=8192):
            yield chunk
    
    return Response(
        generate(),
        mimetype=node.mimetype or 'application/octet-stream',
        headers={
            'Content-Disposition': f'attachment; filename="{node.name}"',
            'Content-Length': str(node.filesize),
        }
    )

Trash and Restore System

Instead of permanently deleting files, FileFlow moves them to a trash folder. Files in trash are automatically purged after 30 days via a background task.

Copy
python
# models.py (additional fields on FileNode)
class FileNode(db.Model):
    # ... existing fields ...
    
    is_trashed = db.Column(db.Boolean, default=False)
    trashed_at = db.Column(db.DateTime, nullable=True)
    original_parent_id = db.Column(db.Integer, nullable=True)  # Restore destination

# routes/trash.py
@files_bp.route('/trash/<int:file_id>', methods=['POST'])
@login_required
def move_to_trash(file_id: int):
    """Soft-delete a file by moving it to trash."""
    node = FileNode.query.filter_by(
        id=file_id, user_id=current_user.id
    ).first_or_404()
    
    node.is_trashed = True
    node.trashed_at = datetime.utcnow()
    node.original_parent_id = node.parent_id
    node.parent_id = None  # Remove from folder hierarchy
    
    # If it's a folder, trash all children recursively
    if node.is_folder:
        _trash_children_recursive(node)
    
    db.session.commit()
    return jsonify({'message': f'"{node.name}" moved to trash'}), 200

@files_bp.route('/restore/<int:file_id>', methods=['POST'])
@login_required
def restore_from_trash(file_id: int):
    """Restore a trashed file to its original location."""
    node = FileNode.query.filter_by(
        id=file_id, user_id=current_user.id, is_trashed=True
    ).first_or_404()
    
    # Check if the original parent still exists
    if node.original_parent_id:
        parent_exists = FileNode.query.filter_by(
            id=node.original_parent_id,
            user_id=current_user.id,
            is_trashed=False
        ).first()
        node.parent_id = parent_exists.id if parent_exists else None
    
    node.is_trashed = False
    node.trashed_at = None
    node.original_parent_id = None
    db.session.commit()
    
    return jsonify({'message': f'"{node.name}" restored'}), 200

def _trash_children_recursive(folder: FileNode):
    """Recursively mark all children as trashed."""
    for child in folder.children:
        child.is_trashed = True
        child.trashed_at = datetime.utcnow()
        child.original_parent_id = child.parent_id
        if child.is_folder:
            _trash_children_recursive(child)

Search Profiles: Saved Complex Queries

Rather than forcing users to rebuild advanced queries, SearchProfile models are persisted in the database. These profiles track query parameters, file type filters, size limits, and timestamp boundaries.

Copy
python
# models.py
class SearchProfile(db.Model):
    __tablename__ = 'search_profiles'
    
    id = db.Column(db.Integer, primary_key=True)
    user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
    name = db.Column(db.String(100), nullable=False)
    query = db.Column(db.String(255))
    file_types = db.Column(db.String(255))  # Comma-separated: "pdf,docx,xlsx"
    size_min = db.Column(db.BigInteger, default=0)
    size_max = db.Column(db.BigInteger, nullable=True)
    date_from = db.Column(db.DateTime, nullable=True)
    date_to = db.Column(db.DateTime, nullable=True)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)

# routes/search.py
@files_bp.route('/search', methods=['GET'])
@login_required
def search_files():
    """Full-text search with filtering across the user's file tree."""
    query = request.args.get('q', '')
    file_types = request.args.get('types', '').split(',')
    size_max = request.args.get('size_max', type=int)
    
    results = FileNode.query.filter(
        FileNode.user_id == current_user.id,
        FileNode.is_trashed == False,
        FileNode.name.ilike(f'%{query}%')
    )
    
    # Apply file type filter
    if file_types and file_types[0]:
        type_filters = []
        for ft in file_types:
            type_filters.append(FileNode.name.ilike(f'%.{ft.strip()}'))
        results = results.filter(db.or_(*type_filters))
    
    # Apply size filter
    if size_max:
        results = results.filter(FileNode.filesize <= size_max)
    
    files = results.order_by(FileNode.created_at.desc()).limit(50).all()
    
    return jsonify({
        'results': [{
            'id': f.id,
            'name': f.name,
            'size': f.filesize,
            'is_folder': f.is_folder,
            'parent_id': f.parent_id,
        } for f in files]
    })

Performance Metrics

After deploying FileFlow to Render with a PostgreSQL database, here are the actual performance numbers:

Operation	Avg Response Time	p99 Response Time
Folder listing (100 items)	45ms	120ms
File upload (10MB)	1.2s	2.8s
File download (10MB, streaming)	85ms TTFB	150ms TTFB
Recursive breadcrumb CTE	8ms	35ms
Full-text search	65ms	180ms
Deduplication check (SHA-256)	12ms	30ms

The recursive CTE performance is noteworthy. Even for directories nested 15 levels deep, the breadcrumb resolution completes in under 10ms because PostgreSQL optimizes recursive queries aggressively.

Final Engineering Thoughts

Constructing FileFlow reinforced the necessity of adhering strictly to relational schema principles and implementing disciplined infrastructure architectures. A web framework handles the HTTP specification effortlessly, but the developer is fundamentally responsible for building secure, scalable state machines.

By centralizing the storage logic, unifying the hierarchy models through recursive SQL patterns, and hardening the security boundaries, FileFlow provides a professional-grade alternative to mainstream storage options.

The key architectural decisions that made this project successful:

1. Unified Entity Model - Using a single FileNode table with a self-referential foreign key eliminated the complexity of managing separate File and Folder tables.
2. Content-Addressable Storage - Keying S3 objects by their SHA-256 hash provided free deduplication with zero application-level complexity.
3. Recursive CTEs - PostgreSQL's recursive query support made hierarchical operations fast without denormalizing the schema.
4. Soft Deletion - The trash/restore system prevented accidental data loss while keeping the active file tree clean.

---

Connect With Me

• GitHub: @amitdevx
• LinkedIn: Amit Divekar
• X / Twitter: @amitdevx_
• Instagram: @amitdevx

If you have any questions or want to discuss this topic further, feel free to reach out!