shared_data/storage.py

from shared_data.supabase_client import supabase, BUCKET


import os
import traceback

def upload_file_to_supabase(local_file_path: str):

    filename = os.path.basename(local_file_path)
    remote_path = f"uploads/{filename}"

    try:

        print("=" * 60)
        print("Uploading document...")
        print("LOCAL :", local_file_path)
        print("REMOTE:", remote_path)
        print("=" * 60)

        with open(local_file_path, "rb") as f:
            file_data = f.read()

        response = supabase.storage.from_(BUCKET).upload(
            path=remote_path,
            file=data
        )

        print("UPLOAD RESPONSE:")
        print(response)
        print("UPLOAD SUCCESS")

    except Exception as e:

        print("=" * 60)
        print("UPLOAD FAILED")
        print(traceback.format_exc())
        print("=" * 60)

        raise

def download_file(remote_path, local_path):

    data = supabase.storage.from_(BUCKET).download(
        remote_path
    )

    with open(local_path, "wb") as f:

        f.write(data)


def list_files(folder):

    return supabase.storage.from_(BUCKET).list(
        folder
    )


def delete_file(remote_path):

    supabase.storage.from_(BUCKET).remove(
        [remote_path]
    )


def delete_all(folder):

    files = list_files(folder)

    if len(files) == 0:
        return

    paths = []

    for file in files:

        paths.append(
            folder + "/" + file["name"]
        )

    supabase.storage.from_(BUCKET).remove(
        paths
    )
def upload_file_to_supabase(local_file_path: str):

    filename = os.path.basename(local_file_path)

    remote_path = f"uploads/{filename}"

    with open(local_file_path, "rb") as f:
        supabase.storage.from_(BUCKET).upload(
            remote_path,
            f,
            {"upsert": True}
        )

    print(f"Uploaded {filename} to Supabase")