Servicebus.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "name": "Servicebus.ipynb",
      "provenance": [],
      "collapsed_sections": [],
      "authorship_tag": "ABX9TyPcfdsRaUTO+ku9dvEueZkh",
      "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/anuthereaper/PythonLibrary/blob/main/Servicebus.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "n-8eeo1pHUG5"
      },
      "source": [
        "https://github.com/Azure/azure-sdk-for-python/tree/main/sdk/servicebus/azure-servicebus/samples/async_samples\n",
        "https://github.com/Azure/azure-sdk-for-python/tree/main/sdk/servicebus/azure-servicebus/samples/sync_samples\n",
        "https://docs.microsoft.com/en-us/azure/service-bus-messaging/service-bus-python-how-to-use-queues"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "xT1-8KE7v6E2",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "f3cec051-92bf-4e2b-8a2b-06f4329460c8"
      },
      "source": [
        "pip install azure.servicebus"
      ],
      "execution_count": 3,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Collecting azure.servicebus\n",
            "  Downloading azure_servicebus-7.4.0-py2.py3-none-any.whl (202 kB)\n",
            "\u001b[K     |████████████████████████████████| 202 kB 4.2 MB/s \n",
            "\u001b[?25hCollecting azure-core<2.0.0,>=1.14.0\n",
            "  Downloading azure_core-1.21.0-py2.py3-none-any.whl (178 kB)\n",
            "\u001b[K     |████████████████████████████████| 178 kB 42.9 MB/s \n",
            "\u001b[?25hCollecting isodate>=0.6.0\n",
            "  Downloading isodate-0.6.0-py2.py3-none-any.whl (45 kB)\n",
            "\u001b[K     |████████████████████████████████| 45 kB 3.4 MB/s \n",
            "\u001b[?25hCollecting azure-common~=1.1\n",
            "  Downloading azure_common-1.1.27-py2.py3-none-any.whl (12 kB)\n",
            "Collecting msrest<2.0.0,>=0.6.17\n",
            "  Downloading msrest-0.6.21-py2.py3-none-any.whl (85 kB)\n",
            "\u001b[K     |████████████████████████████████| 85 kB 4.2 MB/s \n",
            "\u001b[?25hRequirement already satisfied: six>=1.11.0 in /usr/local/lib/python3.7/dist-packages (from azure.servicebus) (1.15.0)\n",
            "Collecting uamqp<2.0.0,>=1.4.3\n",
            "  Downloading uamqp-1.4.3-cp37-cp37m-manylinux_2_12_x86_64.manylinux2010_x86_64.whl (3.1 MB)\n",
            "\u001b[K     |████████████████████████████████| 3.1 MB 34.4 MB/s \n",
            "\u001b[?25hRequirement already satisfied: requests>=2.18.4 in /usr/local/lib/python3.7/dist-packages (from azure-core<2.0.0,>=1.14.0->azure.servicebus) (2.23.0)\n",
            "Requirement already satisfied: certifi>=2017.4.17 in /usr/local/lib/python3.7/dist-packages (from msrest<2.0.0,>=0.6.17->azure.servicebus) (2021.10.8)\n",
            "Requirement already satisfied: requests-oauthlib>=0.5.0 in /usr/local/lib/python3.7/dist-packages (from msrest<2.0.0,>=0.6.17->azure.servicebus) (1.3.0)\n",
            "Requirement already satisfied: urllib3!=1.25.0,!=1.25.1,<1.26,>=1.21.1 in /usr/local/lib/python3.7/dist-packages (from requests>=2.18.4->azure-core<2.0.0,>=1.14.0->azure.servicebus) (1.24.3)\n",
            "Requirement already satisfied: idna<3,>=2.5 in /usr/local/lib/python3.7/dist-packages (from requests>=2.18.4->azure-core<2.0.0,>=1.14.0->azure.servicebus) (2.10)\n",
            "Requirement already satisfied: chardet<4,>=3.0.2 in /usr/local/lib/python3.7/dist-packages (from requests>=2.18.4->azure-core<2.0.0,>=1.14.0->azure.servicebus) (3.0.4)\n",
            "Requirement already satisfied: oauthlib>=3.0.0 in /usr/local/lib/python3.7/dist-packages (from requests-oauthlib>=0.5.0->msrest<2.0.0,>=0.6.17->azure.servicebus) (3.1.1)\n",
            "Installing collected packages: isodate, uamqp, msrest, azure-core, azure-common, azure.servicebus\n",
            "Successfully installed azure-common-1.1.27 azure-core-1.21.0 azure.servicebus-7.4.0 isodate-0.6.0 msrest-0.6.21 uamqp-1.4.3\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "Qw-hgd4RC1m7"
      },
      "source": [
        "# Receive messages from service bus\n",
        "from azure.servicebus import ServiceBusClient\n",
        "import pandas as pd\n",
        "import json\n",
        "from pandas import json_normalize\n",
        "CONNECTION_STR = 'Endpoint=sb://<servicebusnamespace>.servicebus.windows.net/;SharedAccessKeyName=<accesskeyname>;SharedAccessKey=<Access_Key>;EntityPath=<Service_Bus_Name>'\n",
        "QUEUE_NAME = '<Service_Bus_Name>'\n",
        "\n",
        "servicebus_client = ServiceBusClient.from_connection_string(conn_str=CONNECTION_STR)\n",
        "\n",
        "with servicebus_client:\n",
        "    receiver = servicebus_client.get_queue_receiver(queue_name=QUEUE_NAME)\n",
        "    \n",
        "    # create an Empty DataFrame object\n",
        "    df = pd.DataFrame()\n",
        "    msg_concat = \"\"\n",
        "    dfs = []\n",
        "    with receiver:\n",
        "        received_msgs = receiver.receive_messages(max_message_count=10, max_wait_time=5)  # Receive a batch of 10 messages\n",
        "        for msg in received_msgs:\n",
        "            msg_dict = json.loads(str(msg))\n",
        "            df2 = json_normalize(msg_dict)\n",
        "            df = df.append(df2, ignore_index = True)\n",
        "            receiver.complete_message(msg)\n",
        "print(df)\n",
        "print(\"Receive is done.\")"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "crOz3kYFCzGC"
      },
      "source": [
        "# Send messages to service bus\n",
        "from azure.servicebus import ServiceBusMessage\n",
        "# https://docs.microsoft.com/en-us/azure/service-bus-messaging/service-bus-python-how-to-use-queues\n",
        "from azure.servicebus import ServiceBusClient\n",
        "\n",
        "CONNECTION_STR = 'Endpoint=sb://<servicebusnamespace>.servicebus.windows.net/;SharedAccessKeyName=<accesskeyname>;SharedAccessKey=<Access_Key>;EntityPath=<Service_Bus_Name>'\n",
        "QUEUE_NAME = '<Service_Bus_Name>'\n",
        "\n",
        "servicebus_client = ServiceBusClient.from_connection_string(conn_str=CONNECTION_STR)\n",
        "\n",
        "def send_single_message(sender):\n",
        "    # create a Service Bus message\n",
        "    message = ServiceBusMessage(\"Single Message\")\n",
        "    # send the message to the queue\n",
        "    sender.send_messages(message)\n",
        "    print(\"Sent a single message\")\n",
        "\n",
        "def send_a_list_of_messages(sender):\n",
        "    # create a list of messages\n",
        "    messages = [ServiceBusMessage(\"Message in list\") for _ in range(5)]\n",
        "    # send the list of messages to the queue\n",
        "    sender.send_messages(messages)\n",
        "    print(\"Sent a list of 5 messages\")\n",
        "\n",
        "def send_batch_message(sender):\n",
        "    # create a batch of messages\n",
        "    batch_message = sender.create_message_batch()\n",
        "    for _ in range(10):\n",
        "        try:\n",
        "            # add a message to the batch\n",
        "            batch_message.add_message(ServiceBusMessage(\"{\\\"Name\\\":\\\"\\\",\\\"Seri\\\":21000000,\\\"SiName\\\":\\\"\\\",\\\"As\\\":\\\"\\\"}\"))\n",
        "        except ValueError:\n",
        "            # ServiceBusMessageBatch object reaches max_size.\n",
        "            # New ServiceBusMessageBatch object can be created here to send more data.\n",
        "            break\n",
        "    # send the batch of messages to the queue\n",
        "    sender.send_messages(batch_message)\n",
        "    print(\"Sent a batch of 10 messages\")\n",
        "\n",
        "with servicebus_client:\n",
        "    sender = servicebus_client.get_queue_sender(queue_name=QUEUE_NAME)\n",
        "    with sender:\n",
        "        send_single_message(sender)\n",
        "        send_a_list_of_messages(sender)\n",
        "        send_batch_message(sender)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "SJse8PRWOtQ0"
      },
      "source": [
        "pip install names"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "-b45CPyHN1l1"
      },
      "source": [
        "# SEND BULK LOAD MESSAGES TO SERVICE BUS\n",
        "from random import randrange\n",
        "import random\n",
        "import names\n",
        "import json\n",
        "from datetime import datetime\n",
        "from datetime import timedelta\n",
        "from azure.servicebus import ServiceBusMessage\n",
        "from azure.servicebus import ServiceBusClient\n",
        "import uuid\n",
        "\n",
        "CONNECTION_STR = 'Endpoint=sb://<servicebusnamespace>.servicebus.windows.net/;SharedAccessKeyName=<accesskeyname>;SharedAccessKey=<Access_Key>;EntityPath=<Service_Bus_Name>'\n",
        "QUEUE_NAME = '<Service_Bus_Name>'\n",
        "\n",
        "Number_of_docs = 10\n",
        "d1 = datetime.strptime('1/1/2008 1:30 PM', '%m/%d/%Y %I:%M %p')\n",
        "d2 = datetime.strptime('1/1/2009 4:50 AM', '%m/%d/%Y %I:%M %p')\n",
        "\n",
        "servicebus_client = ServiceBusClient.from_connection_string(conn_str=CONNECTION_STR)\n",
        "\n",
        "def send_single_message(sender,message):\n",
        "    sender.send_messages(message)\n",
        "\n",
        "def random_date(start, end):\n",
        "    \"\"\"\n",
        "    This function will return a random datetime between two datetime \n",
        "    objects.\n",
        "    \"\"\"\n",
        "    delta = end - start\n",
        "    int_delta = (delta.days * 24 * 60 * 60) + delta.seconds\n",
        "    random_second = randrange(int_delta)\n",
        "    return start + timedelta(seconds=random_second)\n",
        "\n",
        "def Create_Record(item_id):\n",
        "    \"\"\"\n",
        "    This function will return a service bus message with random data.\n",
        "    \"\"\"\n",
        "    doc_body = {'id' : str(uuid.uuid4()),\n",
        "            'employee_id' : str(id),\n",
        "            'first_name' : names.get_first_name(),\n",
        "            'last_name' : names.get_last_name(),\n",
        "            'regn_no' : 'TEST00000000' + str(item_id),\n",
        "            'start_date' : str(random_date(d1,d2)),\n",
        "            'salary' : str(random.randint(12000, 2000000)),\n",
        "            'inserted_at' : str(datetime.now())\n",
        "            }\n",
        "    json_object = json.dumps(doc_body, indent = 4) \n",
        "#   print(json_object)\n",
        "    return ServiceBusMessage(bytes(json_object,'utf-8'))\n",
        "\n",
        "sender = servicebus_client.get_queue_sender(queue_name=QUEUE_NAME)\n",
        "starttime = datetime.utcnow()\n",
        "print(\"Starting ingestion: \", starttime.strftime(\"%Y-%m-%d %H:%M:%S.%f\"))\n",
        "with servicebus_client:\n",
        "    sender = servicebus_client.get_queue_sender(queue_name=QUEUE_NAME)\n",
        "    with sender:\n",
        "        for i in range(Number_of_docs):\n",
        "            message = Create_Record(i)\n",
        "            send_single_message(sender,message)\n",
        "            if (i+1)%100 == 0 or (i == Number_of_docs-1):\n",
        "                print(str(i+1) + \" documents completed\")\n",
        "\n",
        "endtime = datetime.utcnow()\n",
        "print(\"\\nrun_sample done :\" + endtime.strftime(\"%Y-%m-%d %H:%M:%S.%f\"))\n",
        "print(\"Time taken :\" + str(endtime-starttime))\n",
        "print(\"Number of messages :\" + str(Number_of_docs))"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "y54FmG3R9SUf",
        "outputId": "8f0a2d23-9f39-4522-f11b-41344ee5cff9"
      },
      "source": [
        "# Send messages with custom and system properties (https://docs.microsoft.com/en-us/python/api/azure-servicebus/azure.servicebus.servicebusmessage?view=azure-python)\n",
        "from azure.servicebus import ServiceBusMessage\n",
        "import datetime\n",
        "# https://docs.microsoft.com/en-us/azure/service-bus-messaging/service-bus-python-how-to-use-queues\n",
        "from azure.servicebus import ServiceBusClient\n",
        "\n",
        "CONNECTION_STR = 'Endpoint=sb://<servicebusnamespace>.servicebus.windows.net/;SharedAccessKeyName=<accesskeyname>;SharedAccessKey=<Access_Key>;EntityPath=<Service_Bus_Name>'\n",
        "QUEUE_NAME = <queue_or_topic_name>\n",
        "\n",
        "servicebus_client = ServiceBusClient.from_connection_string(conn_str=CONNECTION_STR)\n",
        "\n",
        "def send_single_message(sender):\n",
        "    # create a Service Bus message\n",
        "    message = ServiceBusMessage(\"Nivis Message1\",session_id=\"MySessionID\",application_properties={'anucustomkey': 'Nivi'},time_to_live=datetime.timedelta(seconds=30),label='MyLabel')\n",
        "#    message.ApplicationProperties.Add(\"anucustomKey\", \"Nivi\");\n",
        "    # send the message to the queue\n",
        "    sender.send_messages(message)\n",
        "    print(\"Sent a single message\")\n",
        "\n",
        "with servicebus_client:\n",
        "    sender = servicebus_client.get_queue_sender(queue_name=QUEUE_NAME)\n",
        "    with sender:\n",
        "        send_single_message(sender)"
      ],
      "execution_count": 12,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "Sent a single message\n"
          ]
        }
      ]
    }
  ]
}