Subir archivos

UploadFile para archivos individuales

FastAPI proporciona la clase UploadFile para manejar archivos que los usuarios envían a través de formularios web. Esta clase está optimizada para trabajar con archivos de cualquier tamaño y ofrece una interfaz sencilla para acceder tanto al contenido como a los metadatos del archivo.

Configuración básica

Para recibir archivos en FastAPI, necesitas importar UploadFile desde el módulo fastapi y usarla como tipo de parámetro en tus funciones de ruta:

from fastapi import APIRouter, UploadFile, File
from fastapi.responses import JSONResponse

router = APIRouter()

@router.post("/upload")
async def upload_file(file: UploadFile = File(...)):
    return {
        "filename": file.filename,
        "content_type": file.content_type,
        "size": file.size
    }

El decorador File(...) indica que el parámetro debe ser tratado como un archivo. Los tres puntos (...) significan que el archivo es obligatorio.

Propiedades principales de UploadFile

La clase UploadFile te da acceso a información importante sobre el archivo subido:

• filename: El nombre original del archivo tal como lo envió el usuario
• content_type: El tipo MIME del archivo (por ejemplo, image/jpeg o text/plain)
• size: El tamaño del archivo en bytes

@router.post("/file-info")
async def get_file_info(file: UploadFile = File(...)):
    # Obtener información básica del archivo
    file_info = {
        "nombre_original": file.filename,
        "tipo_contenido": file.content_type,
        "tamaño_bytes": file.size
    }
    
    return JSONResponse(content=file_info)

Lectura del contenido del archivo

Para acceder al contenido del archivo, UploadFile ofrece varios métodos asincrónicos:

@router.post("/read-file")
async def read_file_content(file: UploadFile = File(...)):
    # Leer todo el contenido como bytes
    content = await file.read()
    
    # Para archivos de texto, puedes decodificar
    if file.content_type.startswith("text/"):
        text_content = content.decode("utf-8")
        return {"content": text_content}
    
    return {"message": f"Archivo {file.filename} leído correctamente"}

También puedes leer el archivo línea por línea si es un archivo de texto:

@router.post("/read-lines")
async def read_file_lines(file: UploadFile = File(...)):
    lines = []
    
    # Leer línea por línea
    async for line in file:
        lines.append(line.decode("utf-8").strip())
    
    return {"lines": lines, "total_lines": len(lines)}

Validación básica de archivos

Es importante validar los archivos antes de procesarlos. Puedes verificar el tipo de contenido y el tamaño:

@router.post("/upload-image")
async def upload_image(file: UploadFile = File(...)):
    # Validar que sea una imagen
    if not file.content_type.startswith("image/"):
        return JSONResponse(
            status_code=400,
            content={"error": "Solo se permiten archivos de imagen"}
        )
    
    # Validar tamaño (máximo 5MB)
    max_size = 5 * 1024 * 1024  # 5MB en bytes
    if file.size > max_size:
        return JSONResponse(
            status_code=400,
            content={"error": "El archivo es demasiado grande"}
        )
    
    return {
        "message": "Imagen válida",
        "filename": file.filename,
        "size": file.size
    }

Manejo de archivos opcionales

Puedes hacer que la subida de archivos sea opcional eliminando los tres puntos del parámetro File:

@router.post("/optional-upload")
async def optional_file_upload(file: UploadFile = File(None)):
    if file is None:
        return {"message": "No se subió ningún archivo"}
    
    return {
        "message": "Archivo recibido",
        "filename": file.filename
    }

Ejemplo completo con validación

Aquí tienes un ejemplo que combina validación y procesamiento básico:

from fastapi import APIRouter, UploadFile, File, HTTPException
from fastapi.responses import JSONResponse

router = APIRouter()

@router.post("/process-document")
async def process_document(document: UploadFile = File(...)):
    # Validar extensión del archivo
    allowed_extensions = [".txt", ".pdf", ".docx"]
    file_extension = "." + document.filename.split(".")[-1].lower()
    
    if file_extension not in allowed_extensions:
        raise HTTPException(
            status_code=400,
            detail="Tipo de archivo no permitido"
        )
    
    # Validar tamaño
    if document.size > 10 * 1024 * 1024:  # 10MB
        raise HTTPException(
            status_code=400,
            detail="El archivo es demasiado grande"
        )
    
    # Procesar el archivo
    content = await document.read()
    
    return {
        "status": "success",
        "filename": document.filename,
        "size": document.size,
        "processed": True
    }

La clase UploadFile maneja automáticamente la limpieza de memoria y cierra los archivos temporales cuando termina el procesamiento, por lo que no necesitas preocuparte por la gestión manual de recursos.

Guardar archivos en el servidor local

Una vez que recibes un archivo a través de UploadFile, el siguiente paso es almacenarlo de forma permanente en tu servidor. FastAPI te permite guardar archivos en el sistema de archivos local de manera sencilla y eficiente.

Estructura de directorios para archivos

Antes de guardar archivos, es importante crear una estructura organizada en tu servidor. Lo más común es tener un directorio dedicado para los archivos subidos:

import os
from pathlib import Path

# Crear directorio para archivos subidos
UPLOAD_DIR = Path("uploads")
UPLOAD_DIR.mkdir(exist_ok=True)

También puedes organizar los archivos por categorías o fechas:

from datetime import datetime

def create_upload_structure():
    base_dir = Path("uploads")
    today = datetime.now().strftime("%Y-%m-%d")
    
    # Crear directorios por fecha
    daily_dir = base_dir / today
    daily_dir.mkdir(parents=True, exist_ok=True)
    
    return daily_dir

Guardado básico de archivos

Para guardar un archivo en el servidor, necesitas leer su contenido y escribirlo en una ubicación específica:

from fastapi import APIRouter, UploadFile, File
from pathlib import Path
import shutil

router = APIRouter()

@router.post("/save-file")
async def save_file(file: UploadFile = File(...)):
    # Definir la ruta donde guardar el archivo
    upload_dir = Path("uploads")
    upload_dir.mkdir(exist_ok=True)
    
    file_path = upload_dir / file.filename
    
    # Guardar el archivo
    with open(file_path, "wb") as buffer:
        content = await file.read()
        buffer.write(content)
    
    return {
        "message": "Archivo guardado correctamente",
        "filename": file.filename,
        "path": str(file_path)
    }

Uso de shutil para mayor eficiencia

Python ofrece shutil.copyfileobj() que es más eficiente para copiar archivos grandes, ya que no carga todo el contenido en memoria:

@router.post("/save-efficient")
async def save_file_efficient(file: UploadFile = File(...)):
    upload_dir = Path("uploads")
    upload_dir.mkdir(exist_ok=True)
    
    file_path = upload_dir / file.filename
    
    # Método más eficiente para archivos grandes
    with open(file_path, "wb") as buffer:
        shutil.copyfileobj(file.file, buffer)
    
    return {
        "message": "Archivo guardado eficientemente",
        "filename": file.filename,
        "size": file_path.stat().st_size
    }

Generar nombres únicos para evitar conflictos

Para evitar que archivos con el mismo nombre se sobrescriban, puedes generar nombres únicos:

import uuid
from datetime import datetime

@router.post("/save-unique")
async def save_with_unique_name(file: UploadFile = File(...)):
    upload_dir = Path("uploads")
    upload_dir.mkdir(exist_ok=True)
    
    # Generar nombre único
    file_extension = Path(file.filename).suffix
    unique_filename = f"{uuid.uuid4()}{file_extension}"
    file_path = upload_dir / unique_filename
    
    # Guardar archivo
    with open(file_path, "wb") as buffer:
        shutil.copyfileobj(file.file, buffer)
    
    return {
        "message": "Archivo guardado con nombre único",
        "original_name": file.filename,
        "saved_name": unique_filename,
        "path": str(file_path)
    }

Organización por tipo de archivo

Puedes organizar automáticamente los archivos en subdirectorios según su tipo:

@router.post("/save-organized")
async def save_organized_file(file: UploadFile = File(...)):
    base_dir = Path("uploads")
    
    # Determinar subdirectorio según tipo de archivo
    if file.content_type.startswith("image/"):
        subdir = base_dir / "images"
    elif file.content_type.startswith("text/"):
        subdir = base_dir / "documents"
    elif file.content_type == "application/pdf":
        subdir = base_dir / "pdfs"
    else:
        subdir = base_dir / "others"
    
    # Crear directorio si no existe
    subdir.mkdir(parents=True, exist_ok=True)
    
    # Generar nombre único
    file_extension = Path(file.filename).suffix
    unique_filename = f"{uuid.uuid4()}{file_extension}"
    file_path = subdir / unique_filename
    
    # Guardar archivo
    with open(file_path, "wb") as buffer:
        shutil.copyfileobj(file.file, buffer)
    
    return {
        "message": "Archivo organizado y guardado",
        "category": subdir.name,
        "filename": unique_filename,
        "path": str(file_path)
    }

Manejo de errores al guardar

Es importante manejar los errores que pueden ocurrir durante el guardado:

from fastapi import HTTPException
import os

@router.post("/save-safe")
async def save_file_safely(file: UploadFile = File(...)):
    upload_dir = Path("uploads")
    
    try:
        # Verificar que el directorio sea accesible
        upload_dir.mkdir(exist_ok=True)
        
        # Generar ruta del archivo
        file_extension = Path(file.filename).suffix
        unique_filename = f"{uuid.uuid4()}{file_extension}"
        file_path = upload_dir / unique_filename
        
        # Verificar espacio disponible (ejemplo básico)
        if file.size > 50 * 1024 * 1024:  # 50MB
            raise HTTPException(
                status_code=413,
                detail="Archivo demasiado grande"
            )
        
        # Guardar archivo
        with open(file_path, "wb") as buffer:
            shutil.copyfileobj(file.file, buffer)
        
        # Verificar que se guardó correctamente
        if not file_path.exists():
            raise HTTPException(
                status_code=500,
                detail="Error al guardar el archivo"
            )
        
        return {
            "message": "Archivo guardado correctamente",
            "filename": unique_filename,
            "size": file_path.stat().st_size
        }
        
    except OSError as e:
        raise HTTPException(
            status_code=500,
            detail=f"Error del sistema: {str(e)}"
        )
    except Exception as e:
        raise HTTPException(
            status_code=500,
            detail="Error inesperado al guardar el archivo"
        )

Ejemplo completo con metadatos

Aquí tienes un ejemplo que guarda el archivo y también registra metadatos útiles:

import json
from datetime import datetime

@router.post("/save-with-metadata")
async def save_file_with_metadata(file: UploadFile = File(...)):
    upload_dir = Path("uploads")
    upload_dir.mkdir(exist_ok=True)
    
    # Generar nombre único
    file_extension = Path(file.filename).suffix
    unique_filename = f"{uuid.uuid4()}{file_extension}"
    file_path = upload_dir / unique_filename
    
    # Guardar archivo
    with open(file_path, "wb") as buffer:
        shutil.copyfileobj(file.file, buffer)
    
    # Crear archivo de metadatos
    metadata = {
        "original_name": file.filename,
        "saved_name": unique_filename,
        "content_type": file.content_type,
        "size": file.size,
        "upload_date": datetime.now().isoformat(),
        "path": str(file_path)
    }
    
    # Guardar metadatos en archivo JSON
    metadata_path = upload_dir / f"{unique_filename}.json"
    with open(metadata_path, "w") as meta_file:
        json.dump(metadata, meta_file, indent=2)
    
    return {
        "message": "Archivo y metadatos guardados",
        "file_id": str(uuid.uuid4()).split('-')[0],
        "metadata": metadata
    }

Recuerda que al guardar archivos en el servidor local, debes considerar aspectos como permisos de escritura, espacio disponible y seguridad del directorio donde almacenas los archivos.