Supabase_Storage_migration.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "provenance": [],
      "toc_visible": true,
      "include_colab_link": true
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "view-in-github",
        "colab_type": "text"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/github/chbert/formsnap/blob/main/Supabase_Storage_migration.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "### Step 1 - Install required packages for the migration process"
      ],
      "metadata": {
        "id": "Altqt5LaoXwx"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "!apt-get update && apt-get install -y libmagic1\n",
        "!pip install python-magic supabase"
      ],
      "metadata": {
        "id": "qPeREzcLoipj"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "### Step 2 - Import required packages"
      ],
      "metadata": {
        "id": "8TWI39reWh7Q"
      }
    },
    {
      "cell_type": "code",
      "source": [],
      "metadata": {
        "id": "2Q7yrXzYZNWP"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "import re\n",
        "import os\n",
        "import magic\n",
        "import shutil\n",
        "import zipfile\n",
        "from typing import Optional\n",
        "from pathlib import Path\n",
        "from google.colab import files\n",
        "from supabase import create_client\n",
        "\n",
        "print(\"✓ Imports are set!\")"
      ],
      "metadata": {
        "id": "PuGNb3PrXRat"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "### Step 3 - Setup Supabase database credentials"
      ],
      "metadata": {
        "id": "6Aj4aB8JWnz6"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "NEW_PROJECT_URL = \"https://[PROJECT_REF].supabase.co\" # @param {type:\"string\"}\n",
        "NEW_SERVICE_KEY = \"SERVICE_ROLE_KEY\" # @param {type:\"string\"}\n",
        "\n",
        "print(\"✓ New Project Credentials are set!\")"
      ],
      "metadata": {
        "id": "nbFh011YYMHH"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "### Step 4 - Storage migration script"
      ],
      "metadata": {
        "id": "HX03RDkuZM-r"
      }
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "ZzYfJ1W0oE8p"
      },
      "outputs": [],
      "source": [
        "# Initialize Supabase client\n",
        "try:\n",
        "    new_supabase_client = create_client(NEW_PROJECT_URL, NEW_SERVICE_KEY)\n",
        "except Exception as e:\n",
        "    raise Exception(f\"Failed to initialize Supabase client: {str(e)}\")\n",
        "\n",
        "# Upload and extract files\n",
        "local_root_path = upload_to_colab()\n",
        "\n",
        "# List existing buckets\n",
        "try:\n",
        "    buckets = new_supabase_client.storage.list_buckets()\n",
        "except Exception as e:\n",
        "    raise Exception(f\"Failed to list buckets: {str(e)}\")\n",
        "\n",
        "if not local_root_path.exists():\n",
        "    raise FileNotFoundError(f\"Local root path does not exist: {local_root_path}\")\n",
        "\n",
        "print(\"\\nFound the following buckets:\")\n",
        "for bucket in buckets:\n",
        "    print(f\"- {bucket.name}\")\n",
        "\n",
        "for bucket in buckets:\n",
        "    bucket_name = bucket.name\n",
        "    local_bucket_path = local_root_path /bucket_name\n",
        "    print(f\"\\nLocal path: '{local_bucket_path}'\")\n",
        "    if not local_bucket_path.exists():\n",
        "        print(f\"\\nLocal path for bucket '{bucket_name}' does not exist. Skipping.\")\n",
        "        continue\n",
        "\n",
        "    print(f\"\\nProcessing bucket '{bucket_name}'...\")\n",
        "\n",
        "    # Create bucket if it doesn't exist\n",
        "    try:\n",
        "        new_supabase_client.storage.create_bucket(\n",
        "            bucket_name,\n",
        "            options={\"public\": bucket.public}\n",
        "        )\n",
        "        print(f\"Bucket '{bucket_name}' created successfully.\")\n",
        "    except Exception as e:\n",
        "        if \"already exists\" not in str(e).lower():\n",
        "            print(f\"Error creating bucket '{bucket_name}': {str(e)}\")\n",
        "            continue\n",
        "\n",
        "    # Upload files\n",
        "    upload_local_files_to_bucket(\n",
        "        new_supabase_client,\n",
        "        bucket_name,\n",
        "        str(local_bucket_path),\n",
        "        skip_existing=True\n",
        "    )\n",
        "\n",
        "def upload_to_colab():\n",
        "    \"\"\"Upload and extract ZIP file containing storage buckets\"\"\"\n",
        "    print(\"Please select your storage ZIP file containing all bucket folders...\")\n",
        "    uploaded = files.upload()\n",
        "\n",
        "    if not uploaded:\n",
        "        raise ValueError(\"No file was uploaded\")\n",
        "\n",
        "    # Get the first uploaded file\n",
        "    zip_filename = next(iter(uploaded))\n",
        "\n",
        "    # Create a temporary directory for extraction\n",
        "    temp_dir = Path(\"/content/storage_temp\")\n",
        "    if temp_dir.exists():\n",
        "        shutil.rmtree(temp_dir)\n",
        "    temp_dir.mkdir()\n",
        "\n",
        "    # Extract the ZIP file\n",
        "    with zipfile.ZipFile(zip_filename, 'r') as zip_ref:\n",
        "        zip_ref.extractall(temp_dir)\n",
        "\n",
        "    print(f\"\\nFiles extracted to {temp_dir}\")\n",
        "\n",
        "    root_candidates = [d for d in temp_dir.iterdir() if d.is_dir()]\n",
        "    root_dir = root_candidates[0]\n",
        "\n",
        "    print(f\"\\nNew root dir is {root_dir}\")\n",
        "\n",
        "    return root_dir\n",
        "\n",
        "def upload_local_files_to_bucket(\n",
        "    client,\n",
        "    bucket_name: str,\n",
        "    local_bucket_path: str,\n",
        "    skip_existing: bool = False\n",
        ") -> None:\n",
        "    \"\"\"Upload files from a local directory to a specific Supabase storage bucket.\"\"\"\n",
        "    local_bucket_path = Path(local_bucket_path)\n",
        "\n",
        "    if not local_bucket_path.exists():\n",
        "        raise FileNotFoundError(f\"Local bucket path does not exist: {local_bucket_path}\")\n",
        "\n",
        "    total_files = sum(1 for _ in local_bucket_path.rglob('*') if _.is_file())\n",
        "    processed_files = 0\n",
        "\n",
        "    for file_path in local_bucket_path.rglob('*'):\n",
        "        if not file_path.is_file():\n",
        "            continue\n",
        "\n",
        "        try:\n",
        "            relative_path = str(file_path.relative_to(local_bucket_path))\n",
        "            remote_path = relative_path.replace(os.sep, '/')\n",
        "\n",
        "            processed_files += 1\n",
        "            print(f\"\\rProgress: {processed_files}/{total_files} files\", end=\"\")\n",
        "\n",
        "            if skip_existing:\n",
        "                try:\n",
        "                    client.storage.from_(bucket_name).download(remote_path)\n",
        "                    print(f\"\\nSkipping existing file: {remote_path}\")\n",
        "                    continue\n",
        "                except Exception:\n",
        "                    pass\n",
        "\n",
        "            mime_type = magic.from_file(str(file_path), mime=True)\n",
        "\n",
        "            with open(file_path, 'rb') as file_object:\n",
        "                client.storage.from_(bucket_name).upload(\n",
        "                    remote_path,\n",
        "                    file_object,\n",
        "                    file_options={\"content-type\": mime_type, \"x-upsert\": \"true\"}\n",
        "                )\n",
        "        except Exception as e:\n",
        "            print(f\"\\nError uploading {file_path}: {str(e)}\")\n",
        "\n",
        "    print(f\"\\nCompleted uploading files to bucket '{bucket_name}'\")"
      ]
    }
  ]
}