I have a azure function that will consume the data from an event hub and redirect the same data to an REST API. I have already written a set of code which is will redirect redirect the Event hub output to the API(testing pending) but now I want to integrate the flask framework with my existing code but I don't know what changes I should made to the following code so that I can use the
======================================================================================================
__init__.py
======================================================================================================
from PWO_EventHubTrigger.postCall import GetPost
from typing import List
import logging
import os
import azure.functions as func
from azure.storage.blob import BlobClient
import datetime
import json
storage_connection_string = os.getenv('eh_rec_storage_connection_string_FromKeyVault')
logging.info('storage_connection_string: %s', storage_connection_string)
container_name = os.getenv('eh_rec_storage_container_name_FromKeyVault')
logging.info('container_name: %s', container_name)
today = datetime.datetime.today()
def main(events: List[func.EventHubEvent]):
logging.info('-----------main function starts-----------')
try:
for event in events:
data = event.get_body().decode('utf-8')
json.loads(data)
logging.info('Python EventHub trigger processed an event: %s', data)
logging.info(f' SequenceNumber = {event.sequence_number}')
logging.info(f' Offset = {event.offset}')
try:
logging.info("Calling GetPost method from main")
res = GetPost(data)
logging.info("GetPost response recived: ", res)
except Exception as Argument:
logging.exception("Error occurred during the GetPost method")
except Exception as Argument:
logging.exception("Error occured in main")
======================================================================================================
postcall.py
======================================================================================================
import requests
import logging
# data={'number': 12524, 'type': 'issue', 'action': 'show'}
# data = {"Source": "localhost", "Timestamp": "2021-05-14 14:30:00","Tags": [{"TagName": "testtag1","TagValue": "10.52","TagQuality": 192}]}
def GetPost(data):
logging.info('-----------GetPost function starts-----------')
try:
headers = {
'user-agent': 'customize header string',
'Content-Type': 'application/json; charset=utf-8'
}
response = requests.post('https://10.199.215.156:443/RestEndpoint', data= data, headers=headers, timeout=3)
# response = requests.post('http://bugs.python.org', data= data, headers=headers, timeout=3)
print("Http response code:", response.status_code)
# print("Http response text:", response.text)
response.raise_for_status() # Raise error in case of failure
except requests.exceptions.HTTPError as httpErr:
logging.info('Http Error:', httpErr)
except requests.exceptions.ConnectionError as connErr:
logging.info('Error Connecting:', connErr)
except requests.exceptions.Timeout as timeOutErr:
logging.info('Timeout Error:', timeOutErr)
except requests.exceptions.RequestException as reqErr:
logging.info('Something Else:', reqErr)
except Exception as err:
logging.info('Something Else:', err)
======================================================================================================
function.json --> Azure configuration file
======================================================================================================
{
"scriptFile": "__init__.py",
"bindings": [
{
"type": "eventHubTrigger",
"name": "events",
"direction": "in",
"eventHubName": "my-events",
"connection": "EventHubReceiverPolicy_FromKeyVault",
"cardinality": "many",
"consumerGroup": "$Default",
"dataType": "binary"
}
]
}
======================================================================================================
local.settings.json --> Azure configuration file
======================================================================================================
{
"IsEncrypted": false,
"Values": {
"AzureWebJobsStorage": "<Endpoint1>",
"FUNCTIONS_WORKER_RUNTIME": "python",
"storage_connection_string_FromKeyVault": "<connectionString",
"storage_container_name_FromKeyVault": "<container_name>",
"EventHubReceiverPolicy_FromKeyVault": "<Endpoint2>"
}
}
I am newbie to python development. Apologies if you think my question is silly. Many thanks !!
No, no one will call the function except the Eventhub.. so even ever a new data arrives to the Event hub then the function will trigger itself and will redirect the data to an REST API. So just want to know that would it be useful to integrate any framework with the azure function.. if yes.. then how ??10.199.215.156your flask app public ip? If yes, may I know what issue that you are facing?No, I just want to know -Can I develop a Azure functions with the help of an Framework.