Importar arquivos do Supabase com eficiência em Python

Kevin van Zonneveld

Cofundador · Amsterdã, Países Baixos · Mostrar biografia

Para importar arquivos do Supabase em Python, baixe seus bytes com o SDK oficial de código aberto e salve-os localmente. O script abaixo processa um único objeto ou todos os objetos visíveis dentro de uma pasta, incluindo as páginas seguintes da listagem e as pastas aninhadas. Ele se recusa a sobrescrever arquivos locais e retorna um status de saída de falha se uma importação for interrompida.

Configurar o Supabase e o Python

Você precisa de um projeto no Supabase, um bucket privado e um usuário com e-mail e senha que possa ler os objetos desse bucket. Este tutorial usa um shell POSIX e um sistema de arquivos local com suporte a links físicos. Ele foi testado no Linux com Python 3.12.13, supabase 2.13.0 e storage3 0.11.3. Use uma versão de correção do Python 3.12 que ainda receba manutenção; a versão exata acima é a que foi testada.

1. Instalar as bibliotecas Python necessárias

Em um diretório sob seu controle, com python3.12 e seu módulo venv instalados, cole:

(
  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'
)

O subshell mantém seu diretório atual e as opções do shell intactos. Uma falha na instalação pode deixar um .venv incompleto; mantenha-o para diagnóstico e tente novamente em um diretório novo. Os comandos abaixo usam esse ambiente virtual explicitamente, sem ativá-lo.

2. Configurar credenciais de acesso e permissões no Supabase

Forneça estas variáveis de ambiente pelo ambiente do processo ou pelo gerenciador de segredos:

Este exemplo usa a interface de chaves JWT legadas da versão fixada do cliente. A versão 2.13.0 valida as chaves do projeto como JWTs⁠, por isso, as chaves sb_publishable_… exigem um cliente mais recente. Use a chave anon aqui e autentique o usuário comum; mantenha as chaves de função de serviço fora do importador. O Supabase planeja deprecar as chaves legadas no fim de 2026⁠; este exemplo se aplica a projetos que ainda as disponibilizam.

Para um pequeno exemplo, peça ao proprietário do projeto que crie um bucket privado my-bucket e faça upload de image.jpg e nested/photo.jpg dentro de <user-uuid>/images. O UUID do usuário é seu ID de usuário no Auth, não seu e-mail. O proprietário do projeto pode conceder acesso de leitura à pasta correspondente ao UUID de cada usuário com esta política no editor SQL, substituindo my-bucket se necessário:

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 a listagem⁠ quanto o download⁠ exigem a permissão SELECT em storage.objects. O importador não precisa de permissão para criar buckets nem para fazer upload. Revise também as políticas existentes: políticas permissivas podem conceder acesso além da restrição de pasta deste exemplo. Consulte o guia de controle de acesso do Storage⁠.

3. Estabelecer uma conexão com seu bucket do Supabase

Salve este programa completo como import_supabase.py. Ele faz login antes de criar a referência ao bucket. Mantenha os diretórios-pai do destino sob seu controle e execute as importações sequencialmente, sem que outro processo altere esses diretórios. O script rejeita diretórios-pai que sejam links simbólicos e componentes inseguros de caminhos remotos.

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() retorna o objeto inteiro como bytes. O arquivo temporário é gravado primeiro; em seguida, os.link() cria o destino sem substituir um arquivo, diretório ou link simbólico já existente nesse caminho. O diretório temporário e seu conteúdo são removidos ao sair do bloco. Isso exige suporte a links físicos no sistema de arquivos de destino; o programa não faz downloads em streaming.

Como importar arquivos do Supabase usando Python

Importar um único arquivo

Com as variáveis de ambiente acima definidas, execute:

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

Os bytes do objeto são salvos em local/image.jpg. Uma execução bem-sucedida informa a chave baixada e o total de um arquivo importado. Se o destino já existir, ocorrerá FileExistsError e o programa terminará com status de saída 1; escolha um novo destino em vez de remover um arquivo de que você precisa.

Importar vários arquivos de um diretório

Use o mesmo programa para copiar toda a árvore de pastas visível:

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

Com a configuração de dois objetos acima, você obtém downloads/image.jpg e downloads/nested/photo.jpg, com um total de dois arquivos importados. As entradas da listagem com ID de objeto null representam pastas; o script lista seus filhos em vez de baixar essas entradas. Cada pasta é paginada de forma independente, e o deslocamento inclui entradas de pastas e de arquivos.

Este exemplo aceita nomes ASCII permitidos pelo Storage⁠, incluindo espaços e pontos de interrogação. Ele aplica a codificação de URL a esses nomes para o download, sem alterar os nomes locais. Arquivos vazios são válidos. As verificações de caminho também rejeitam componentes vazios, . e .., barras invertidas, dois-pontos, sinais de porcentagem, cerquilhas e caracteres não ASCII. Os arquivos de destino existentes são preservados; uma colisão interrompe a importação em vez de ignorar um arquivo e informar sucesso.

Limitações e considerações de armazenamento

Cada download mantém um objeto completo na memória, e cada listagem mantém até 100 filhos diretos. Escolha arquivos que caibam na memória disponível. Uma árvore de pastas grande também acumula caminhos de pastas pendentes; este exemplo não define um limite de memória ou de tempo para a importação como um todo.

Mantenha a árvore remota estável durante a importação. A paginação por deslocamento não é um retrato fixo dos dados: uploads, renomeações ou exclusões simultâneos podem causar entradas ausentes ou repetidas. Uma falha interrompe a importação com status de saída 1 e mantém os arquivos já concluídos; a importação do diretório não é uma transação. Após corrigir a causa, use um destino novo para que esses arquivos concluídos não bloqueiem uma nova execução.

Diagnosticar uma importação interrompida ou vazia

Um erro de autenticação exige verificar a URL do projeto, a chave legada anon e as credenciais do usuário confirmado. Um erro do Storage exige verificar o bucket, o caminho exato do objeto e as políticas SELECT desse usuário. Um OSError, incluindo FileExistsError, exige verificar as permissões locais, os destinos existentes, o espaço livre em disco e o suporte a links físicos. A última chave tentada aparece na saída de progresso para que você possa identificar um objeto cuja importação falhou. O script não exibe as respostas do provedor nem as credenciais.

Um total de zero arquivos importados significa que a listagem não disponibilizou nenhum objeto. Isso não prova que o bucket esteja vazio: a RLS pode ocultar objetos desse usuário. Confirme com o proprietário do projeto o inventário esperado antes de considerar uma cópia local completa. As restrições de MIME do bucket e os limites de tamanho de upload⁠ se aplicam aos uploads; eles não validam o conteúdo dos arquivos baixados.

Usar os arquivos importados

Leia os arquivos salvos com as bibliotecas Python usuais da sua aplicação. Se precisar de um fluxo de processamento de mídia separado após o download, você pode consultar também o SDK Python da Transloadit; o programa de download acima não envia arquivos para processamento.