Importa archivos de Supabase en Python con eficiencia

Kevin van Zonneveld

Cofundador · Ámsterdam, Países Bajos · Mostrar bio

Para importar archivos de Supabase en Python, descarga sus bytes con el SDK oficial de código abierto y guárdalos localmente. El script siguiente gestiona un solo objeto o todos los objetos visibles dentro de una carpeta, incluidas las páginas posteriores del listado y las carpetas anidadas. Se niega a sobrescribir archivos locales y devuelve un estado de salida de error si una importación se detiene.

Configura Supabase y Python

Necesitas un proyecto de Supabase, un bucket privado y un usuario con email y contraseña que pueda leer sus objetos. Este tutorial usa un shell POSIX y un sistema de archivos local que admite enlaces duros. Se probó en Linux con Python 3.12.13, supabase 2.13.0 y storage3 0.11.3. Usa una versión de parche de Python 3.12 que siga recibiendo mantenimiento; la versión de parche exacta indicada arriba es la que se probó.

1. Instala las bibliotecas de Python necesarias

En un directorio que controles, con python3.12 y su módulo venv instalados, pega lo siguiente:

(
  command -v python3.12 >/dev/null || exit 1
  if [ -e .venv ] || [ -L .venv ]; then
    printf 'Choose a directory without an existing .venv.\n' >&2
    exit 1
  fi
  python3.12 -m venv .venv &&
    .venv/bin/python -m pip install 'supabase==2.13.0'
)

El subshell mantiene intactos tu directorio actual y las opciones del shell. Un fallo de instalación puede dejar un .venv incompleto; consérvalo para diagnosticar el problema e inténtalo de nuevo en un directorio nuevo. Los comandos siguientes usan este entorno virtual explícitamente, sin activarlo.

2. Configura las credenciales de acceso y los permisos en Supabase

Proporciona estas variables de entorno mediante el entorno de tu proceso o tu gestor de secretos:

Esta guía está dirigida a la interfaz de claves JWT heredadas del cliente cuya versión se ha fijado. La versión 2.13.0 valida las claves de proyecto como JWT⁠, por lo que las claves sb_publishable_… requieren un cliente más reciente. Usa aquí la clave anon y autentica al usuario común; no incluyas claves de rol de servicio en el importador. Supabase planea declarar en desuso las claves heredadas a finales de 2026⁠; esta guía se aplica a los proyectos que todavía las ofrecen.

Para un ejemplo pequeño, pide al propietario de tu proyecto que cree un bucket privado my-bucket y suba image.jpg y nested/photo.jpg dentro de <user-uuid>/images. El UUID del usuario es su ID de usuario de Auth, no su email. El propietario del proyecto puede conceder acceso de lectura a la carpeta del UUID de cada usuario con esta política en el editor SQL, reemplazando my-bucket si es necesario:

create policy "Read own UUID folder"
on storage.objects for select to authenticated
using (
  bucket_id = 'my-bucket'
  and (storage.foldername(name))[1] = (select auth.uid())::text
);

Tanto el listado⁠ como la descarga⁠ requieren el permiso SELECT sobre storage.objects. El importador no necesita permisos para crear buckets ni subir archivos. Revisa también las políticas existentes: las políticas permisivas pueden conceder acceso más allá de la restricción de carpeta de este ejemplo. Consulta la guía de control de acceso de Storage⁠.

3. Establece una conexión con tu bucket de Supabase

Guarda este programa completo como import_supabase.py. Inicia sesión antes de crear la referencia al bucket. Mantén bajo tu control los directorios padre del destino y ejecuta las importaciones secuencialmente, sin que otro proceso modifique esos directorios. El script rechaza los directorios padre que sean enlaces simbólicos y los componentes inseguros de las rutas remotas.

import argparse
import os
import sys
import tempfile
from pathlib import Path
from urllib.parse import quote

from supabase import create_client

VALID_NAME_CHARACTERS = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789_-!.*'() &$=@;+,?"


def validate_path(value: str, allow_empty: bool = False) -> None:
    if allow_empty and value == '':
        return
    for name in value.split('/'):
        if (not name or name in {'.', '..'}
                or any(char not in VALID_NAME_CHARACTERS for char in name)):
            raise ValueError('Unsafe remote path component')


def download_file(bucket, key: str, target: Path) -> None:
    validate_path(key)
    print(f'Downloading {key}', flush=True)
    for parent in [target.parent, *target.parent.parents]:
        if parent.is_symlink():
            raise ValueError('Symlink destination parent')
    target.parent.mkdir(parents=True, exist_ok=True)
    data = bucket.download(quote(key, safe='/'))
    with tempfile.TemporaryDirectory(dir=target.parent) as staging:
        temporary = Path(staging) / 'download'
        temporary.write_bytes(data)
        # Linking publishes a complete file and refuses an existing destination.
        os.link(temporary, target)
    print(f'Downloaded {key}')


def import_directory(bucket, prefix: str, root: Path) -> int:
    prefix = prefix.rstrip('/')
    validate_path(prefix, allow_empty=True)
    pending = [(prefix, root)]
    count = 0
    while pending:
        remote, local = pending.pop()
        offset = 0
        while True:
            rows = bucket.list(path=remote, options={
                'limit': 100, 'offset': offset,
                'sortBy': {'column': 'name', 'order': 'asc'},
            })
            for row in rows:
                name = row['name']
                validate_path(name)
                if '/' in name:
                    raise ValueError('Expected an immediate child name')
                key = f'{remote}/{name}' if remote else name
                target = local / name
                if row.get('id') is None:
                    pending.append((key, target))
                    continue
                download_file(bucket, key, target)
                count += 1
            if len(rows) < 100:
                break
            offset += 100
    return count


def main() -> None:
    parser = argparse.ArgumentParser(description='Download Supabase objects without overwriting')
    parser.add_argument('mode', choices=['file', 'directory'])
    parser.add_argument('bucket')
    parser.add_argument('source')
    parser.add_argument('destination')
    args = parser.parse_args()
    client = create_client(os.environ['SUPABASE_URL'], os.environ['SUPABASE_PROJECT_KEY'])
    client.auth.sign_in_with_password({
        'email': os.environ['SUPABASE_USER_EMAIL'],
        'password': os.environ['SUPABASE_USER_PASSWORD'],
    })
    bucket = client.storage.from_(args.bucket)
    destination = Path(args.destination).absolute()
    if args.mode == 'file':
        download_file(bucket, args.source, destination)
        count = 1
    else:
        count = import_directory(bucket, args.source, destination)
    print(f'Imported {count} file(s)')


if __name__ == '__main__':
    try:
        main()
    except Exception as error:
        print(f'Import stopped ({type(error).__name__}). Check credentials, '
              'Storage SELECT policies, and destination paths.', file=sys.stderr)
        sys.exit(1)

download() devuelve el objeto completo como bytes. Primero se escribe el archivo temporal; después, os.link() crea el destino sin reemplazar ningún archivo, directorio o enlace simbólico que ya exista en esa ruta. El archivo temporal se elimina al salir del bloque. Esto requiere que el sistema de archivos de destino admita enlaces duros; no utiliza descargas en streaming.

Cómo importar archivos de Supabase con Python

Importar un solo archivo

Con las variables de entorno anteriores definidas, ejecuta lo siguiente:

.venv/bin/python import_supabase.py file my-bucket "$SUPABASE_SOURCE_PREFIX/image.jpg" local/image.jpg

Los bytes del objeto se guardan en local/image.jpg. Una ejecución correcta informa de la clave descargada y de un recuento de un archivo importado. Un destino existente provoca FileExistsError y el estado de salida 1; elige un destino nuevo en lugar de eliminar un archivo que necesites.

Importar varios archivos de un directorio

Usa el mismo programa para copiar todo el árbol de carpetas visible:

.venv/bin/python import_supabase.py directory my-bucket "$SUPABASE_SOURCE_PREFIX" downloads

Con la configuración de dos objetos anterior, obtienes downloads/image.jpg y downloads/nested/photo.jpg, con un recuento de dos archivos importados. Las filas del listado con un ID de objeto null representan carpetas; el script lista sus elementos secundarios en lugar de descargar esas filas. Cada carpeta se pagina de forma independiente, y el desplazamiento incluye tanto las filas de carpetas como las de archivos.

Este ejemplo acepta nombres ASCII permitidos por Storage⁠, incluidos los espacios y los signos de interrogación. Los codifica para la descarga, pero mantiene sus nombres locales sin cambios. Los archivos vacíos son válidos. Las comprobaciones de rutas también rechazan los componentes vacíos, . y .., las barras invertidas, los dos puntos, los signos de porcentaje, los signos de numeral y los caracteres no ASCII. Los archivos existentes en el destino se conservan; una colisión detiene la importación en lugar de omitir un archivo e informar de un resultado correcto.

Limitaciones y consideraciones del almacenamiento

Cada descarga mantiene un objeto completo en memoria, y cada listado contiene hasta 100 elementos secundarios directos. Elige archivos que quepan en la memoria disponible. Un árbol de carpetas grande también acumula rutas de carpetas pendientes; este ejemplo no establece un límite de memoria ni de tiempo para la importación completa.

Mantén estable el árbol remoto durante la importación. La paginación por desplazamiento no es una instantánea: las subidas, los cambios de nombre o las eliminaciones simultáneas pueden provocar que falten entradas o se repitan. Un fallo detiene la importación con el estado de salida 1 y conserva los archivos ya completados; la importación del directorio no es una transacción. Después de corregir la causa, usa un destino nuevo para que esos archivos completados no bloqueen una nueva ejecución.

Diagnostica una importación detenida o vacía

Un error de autenticación requiere revisar la URL del proyecto, la clave heredada anon y las credenciales del usuario confirmado. Un error de Storage requiere revisar el bucket, la ruta exacta del objeto y las políticas SELECT de ese usuario. Un OSError, incluido FileExistsError, requiere revisar los permisos locales, los destinos existentes, el espacio libre en disco y la compatibilidad con enlaces duros. La última clave que se intentó procesar aparece en la salida de progreso para que puedas identificar un objeto que haya fallado. El script no imprime las respuestas del proveedor ni las credenciales.

Un recuento de cero archivos importados significa que el listado no mostró ningún objeto. No demuestra que el bucket esté vacío: RLS puede ocultar objetos a ese usuario. Confirma el inventario esperado con el propietario del proyecto antes de considerar completa una copia local. Las restricciones MIME del bucket y los límites de tamaño de subida⁠ regulan las subidas; no validan el contenido de los archivos descargados.

Usa los archivos importados

Lee los archivos guardados con las bibliotecas de Python habituales de tu aplicación. Si necesitas un flujo de trabajo independiente de procesamiento multimedia después de la descarga, puedes consultar el SDK de Python de Transloadit como lectura adicional; el descargador anterior no envía archivos para su procesamiento.