mongo2supabase.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "provenance": [],
      "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/mansueli/Supa-Migrate/blob/main/mongo2supabase.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "![Supabase](https://raw.githubusercontent.com/supabase/supabase/master/packages/common/assets/images/supabase-logo-wordmark--light.svg)\n",
        "\n"
      ],
      "metadata": {
        "id": "FKIrxpYbOcPj"
      }
    },
    {
      "cell_type": "markdown",
      "source": [
        "## Installing the requirements\n",
        "\n",
        "\n"
      ],
      "metadata": {
        "id": "q-40SnoVOlgV"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "!pip install mongo\n",
        "!pip install psycopg2"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "zJWnNtaiUO3I",
        "outputId": "df80139a-41ce-46be-a5bb-24636b835535"
      },
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Collecting mongo\n",
            "  Downloading mongo-0.2.0.tar.gz (4.6 kB)\n",
            "  Preparing metadata (setup.py) ... \u001b[?25l\u001b[?25hdone\n",
            "Collecting pymongo (from mongo)\n",
            "  Downloading pymongo-4.5.0-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl (671 kB)\n",
            "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m671.3/671.3 kB\u001b[0m \u001b[31m8.8 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25hCollecting dnspython<3.0.0,>=1.16.0 (from pymongo->mongo)\n",
            "  Downloading dnspython-2.4.2-py3-none-any.whl (300 kB)\n",
            "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m300.4/300.4 kB\u001b[0m \u001b[31m23.1 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
            "\u001b[?25hBuilding wheels for collected packages: mongo\n",
            "  Building wheel for mongo (setup.py) ... \u001b[?25l\u001b[?25hdone\n",
            "  Created wheel for mongo: filename=mongo-0.2.0-py3-none-any.whl size=4936 sha256=1e6aa64ee8ef89bc0418c6aef0b06c8a26f380aeb184b2e3622bf5a633d1d105\n",
            "  Stored in directory: /root/.cache/pip/wheels/3e/74/1e/a910658709522cedacd7097eeb20d691cf8066f94fc33efacf\n",
            "Successfully built mongo\n",
            "Installing collected packages: dnspython, pymongo, mongo\n",
            "Successfully installed dnspython-2.4.2 mongo-0.2.0 pymongo-4.5.0\n",
            "Requirement already satisfied: psycopg2 in /usr/local/lib/python3.10/dist-packages (2.9.7)\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "## Set the connection URIs:"
      ],
      "metadata": {
        "id": "4csasHqIPEpq"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "#Source DB variables:\n",
        "%env supabase_uri=\n",
        "%env mongo_uri=\n",
        "# Leave this field empty to run for all databases\n",
        "%env mongo_db=sample_mflix\n",
        "#Examples:\n",
        "#%env supabase_uri=postgresql://postgres:password@db.ref.supabase.co:5432/postgres\n",
        "#%env mongo_uri=mongodb+srv://nacho:password@cluster001.xxxx.mongodb.net/?retryWrites=true&w=majority\n",
        "#%env mongo_db=sample_analytics"
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "BjQqN86UPCm6",
        "outputId": "8e38fc06-9475-433d-ff87-7132aa0ca379"
      },
      "execution_count": 1,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "env: supabase_uri=\n",
            "env: mongo_uri=\n",
            "env: mongo_db=sample_mflix\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "p5yhiJRGTFeP",
        "outputId": "3e6a4d24-1f75-4dc8-f7f5-3b1fe5e830b8"
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Starting to migrate :Database(MongoClient(host=['ac-a2telix-shard-00-00.jf3yl2c.mongodb.net:27017', 'ac-a2telix-shard-00-01.jf3yl2c.mongodb.net:27017', 'ac-a2telix-shard-00-02.jf3yl2c.mongodb.net:27017'], document_class=dict, tz_aware=False, connect=True, retrywrites=True, w='majority', authsource='admin', replicaset='atlas-40332v-shard-0', tls=True), 'sample_mflix')\n"
          ]
        }
      ],
      "source": [
        "#@title #Running the Migration: { display-mode: \"form\" }\n",
        "from bson.decimal128 import Decimal128\n",
        "import pymongo\n",
        "import psycopg2\n",
        "from psycopg2.extensions import AsIs\n",
        "import json\n",
        "from datetime import datetime\n",
        "from psycopg2 import sql, extensions, connect, Error\n",
        "from bson import ObjectId\n",
        "import os\n",
        "\n",
        "mongo_url = os.environ['mongo_uri']\n",
        "supabase_url = os.environ['supabase_uri']\n",
        "\n",
        "class DateTimeEncoder(json.JSONEncoder):\n",
        "    def default(self, obj):\n",
        "        if isinstance(obj, datetime):\n",
        "            return obj.isoformat()\n",
        "        return super().default(obj)\n",
        "\n",
        "psycopg2.extensions.register_adapter(Decimal128, lambda val: AsIs(str(val.to_decimal())))\n",
        "\n",
        "# Connect to MongoDB\n",
        "mongo_client = pymongo.MongoClient(mongo_url)\n",
        "# Connect to PostgreSQL\n",
        "pg_conn = connect(supabase_url)\n",
        "pg_conn.set_isolation_level(extensions.ISOLATION_LEVEL_AUTOCOMMIT)\n",
        "pg_cur = pg_conn.cursor()\n",
        "\n",
        "# Mapping MongoDB types to PostgreSQL types\n",
        "SQL_DATA_TYPE = {\n",
        "    \"string\": \"TEXT\",\n",
        "    \"ObjectId\": \"TEXT\",\n",
        "    \"datetime\": \"TIMESTAMP WITH TIME ZONE\",\n",
        "    \"int\": \"INT\",\n",
        "    \"list\": \"JSONB\",\n",
        "    \"dict\": \"JSONB\",\n",
        "    \"bool\": \"Boolean\",\n",
        "    \"float\": \"NUMERIC\",\n",
        "    \"default\": \"TEXT\",\n",
        "}\n",
        "\n",
        "# Store the type of each field\n",
        "field_types = {}\n",
        "\n",
        "# Get the list of database names from MongoDB\n",
        "mongo_db_manual = os.environ['mongo_db']\n",
        "mongo_db_names = []\n",
        "if(len(mongo_db_manual)>0):\n",
        "  mongo_db_names.append(mongo_db_manual)\n",
        "else:\n",
        "  mongo_db_names = mongo_client.list_database_names()\n",
        "\n",
        "# Iterate over all MongoDB databases\n",
        "for db_name in mongo_db_names:\n",
        "    print(\"Starting to migrate :\"+ str(db_name))\n",
        "    mongo_db = mongo_client[db_name]\n",
        "\n",
        "    # Iterate over all collections in the current database\n",
        "    for collection_name in mongo_db.list_collection_names():\n",
        "        # Skip system collections\n",
        "        if collection_name.startswith(\"system.\"):\n",
        "            continue\n",
        "\n",
        "        collection = mongo_db[collection_name]\n",
        "        # Create table in PostgreSQL if it doesn't exist\n",
        "        pg_cur.execute(sql.SQL(\"CREATE TABLE IF NOT EXISTS {} ()\").format(\n",
        "            sql.Identifier(collection_name)))\n",
        "\n",
        "        # Iterate over all documents in the collection\n",
        "        cursor = collection.find()\n",
        "        for document in cursor:\n",
        "            # For each document, build a list of fields and a list of values\n",
        "            fields = []\n",
        "            values = []\n",
        "            for field, value in document.items():\n",
        "                # Determine PostgreSQL type based on Python type\n",
        "                if isinstance(value, ObjectId):\n",
        "                    pg_type = SQL_DATA_TYPE[\"ObjectId\"]\n",
        "                    value = str(value)\n",
        "                else:\n",
        "                    pg_type = SQL_DATA_TYPE.get(type(value).__name__, SQL_DATA_TYPE[\"default\"])\n",
        "\n",
        "                # Add type suffix to field name if a new type is encountered\n",
        "                field_with_type = field\n",
        "                if field in field_types:\n",
        "                    if type(value).__name__ not in field_types[field]:\n",
        "                        field_types[field].add(type(value).__name__)\n",
        "                        field_with_type = f\"{field}_{type(value).__name__}\"\n",
        "                else:\n",
        "                    field_types[field] = {type(value).__name__}\n",
        "\n",
        "                # Add column in PostgreSQL if it doesn't exist\n",
        "                try:\n",
        "                    pg_cur.execute(sql.SQL(\"ALTER TABLE {} ADD COLUMN {} {}\").format(\n",
        "                        sql.Identifier(collection_name),\n",
        "                        sql.Identifier(field_with_type),\n",
        "                        sql.SQL(pg_type)))\n",
        "                except Error:\n",
        "                    pass  # Column already exists, no action needed\n",
        "\n",
        "                # Add field and value to the lists\n",
        "                fields.append(sql.Identifier(field_with_type))\n",
        "                if isinstance(value, list) or isinstance(value, dict):\n",
        "                    value = json.dumps(value, cls=DateTimeEncoder)\n",
        "                values.append(value)\n",
        "\n",
        "            # Insert data into PostgreSQL\n",
        "            pg_cur.execute(sql.SQL(\"INSERT INTO {} ({}) VALUES ({})\").format(\n",
        "                sql.Identifier(collection_name),\n",
        "                sql.SQL(', ').join(fields),\n",
        "                sql.SQL(', ').join(sql.Placeholder() * len(values))),\n",
        "                values)\n",
        "\n",
        "pg_cur.close()\n",
        "pg_conn.close()\n"
      ]
    }
  ]
}