src/notebooks/local_run.ipynb

{
 "cells": [
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## You can run this locally without being inside the docker containers"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import sys\n",
    "sys.path.append('..')\n",
    "import boto3\n",
    "import json \n",
    "import transform_messages as tm\n",
    "import upload_message as um\n",
    "import database_connection as db\n",
    "\n",
    "def process_messages(queue_url, num_messages, queue_access_key,queue_secret, queue_token, db_host, db_port, db_user, db_password, db_name, table_name):\n",
    "    # Connect to the queue\n",
    "    sqs = boto3.client('sqs', region_name='ap-south-1',endpoint_url='http://localhost:4566',\n",
    "                   aws_access_key_id= queue_access_key,\n",
    "                   aws_secret_access_key= queue_secret,\n",
    "                   aws_session_token=queue_token)\n",
    "    print(queue_url)\n",
    "    # Connect to database\n",
    "    cnx = db.connect_to_db(db_host,\n",
    "        db_port,\n",
    "        db_user,\n",
    "        db_password,\n",
    "        db_name)\n",
    "    \n",
    "    cursor = cnx.cursor()\n",
    "    \n",
    "    # Check for table in database\n",
    "    db.create_tables(cursor,table_name)\n",
    "\n",
    "    messages = []\n",
    "    while True:\n",
    "        # Retrieve messages from the queue\n",
    "        response = get_messages_from_queue(sqs, queue_url)\n",
    "\n",
    "        if 'Messages' not in response:\n",
    "            break\n",
    "\n",
    "        for message in response['Messages']:\n",
    "            print(message)\n",
    "            # Process the message\n",
    "            try:\n",
    "                transformed_message = tm.transform_message(json.loads(message['Body']))\n",
    "            except Exception as e:\n",
    "                print(f\"Error transforming message {message['MessageId']}: {e}\")\n",
    "                continue\n",
    "\n",
    "            messages.append(transformed_message)\n",
    "            print('Transformation Finished')\n",
    "\n",
    "            # Delete the message from the queue\n",
    "            delete_message_from_queue(sqs, queue_url, message)\n",
    "            print('messages have been deleted from queue')\n",
    "\n",
    "            # Upload messages in batches\n",
    "            if len(messages) >= 10:\n",
    "                print('Upload in progress')\n",
    "                upload_messages_in_batches(cnx,cursor, table_name, messages)\n",
    "                messages = []\n",
    "\n",
    "    # Upload any remaining messages\n",
    "    if messages:\n",
    "        print('Remaining messages being uploaded')\n",
    "        upload_messages_in_batches(cnx, cursor, table_name, messages)\n",
    "\n",
    "    close_database_connection(cursor, cnx)\n",
    "\n",
    "\n",
    "def get_messages_from_queue(sqs, queue_url):\n",
    "    return sqs.receive_message(\n",
    "        QueueUrl=queue_url,\n",
    "        MaxNumberOfMessages=10,\n",
    "        VisibilityTimeout=30,\n",
    "        WaitTimeSeconds=20\n",
    "    )\n",
    "\n",
    "\n",
    "def delete_message_from_queue(sqs, queue_url, message):\n",
    "    sqs.delete_message(QueueUrl=queue_url, ReceiptHandle=message['ReceiptHandle'])\n",
    "\n",
    "\n",
    "def upload_messages_in_batches(cnx ,cursor, table_name, messages):\n",
    "    um.upload_messages(cnx, cursor, table_name, messages)\n",
    "\n",
    "\n",
    "def close_database_connection(cursor, cnx):\n",
    "    cursor.close()\n",
    "    cnx.close()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "from dotenv import load_dotenv\n",
    "\n",
    "# Path to the .env file\n",
    "env_path = '././.env'\n",
    "\n",
    "# Load the environment variables from the .env file\n",
    "load_dotenv(dotenv_path=env_path)\n",
    "\n",
    "# Access the environment variables\n",
    "queue_url = 'http://localhost:4566/000000000000/test-queue'\n",
    "queue_access_key = os.getenv('QUEUE_ACCESS_KEY')\n",
    "queue_secret = os.getenv('QUEUE_SECRET')\n",
    "queue_token = os.getenv('QUEUE_TOKEN')\n",
    "db_host = os.getenv('DB_HOST')\n",
    "db_port = os.getenv('DB_PORT')\n",
    "db_user = os.getenv('DB_USER')\n",
    "db_password = os.getenv('DB_PASSWORD')\n",
    "db_name = os.getenv('DB_NAME')\n",
    "table_name = os.getenv('TABLE_NAME')\n",
    "num_messages = 10"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "process_messages(queue_url, num_messages, queue_access_key,queue_secret, queue_token, db_host, db_port, db_user, db_password, db_name, table_name)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.4"
  },
  "orig_nbformat": 4
 },
 "nbformat": 4,
 "nbformat_minor": 2
}