Issue
I try to deploy a azure function, withs this python code
from this repository : https://github.com/cutout-pro/api-python-examples.git
The goal is use template "BlobTrigger", get the blob from input container and pass to:
file = "input/idphoto.jpg"
and them upload the final image to output container.
This is my init.py
import azure.functions as func
import logging
import requests
import base64
import json
import os
import urllib.request
from azure.storage.blob import BlobServiceClient, BlobClient, ContainerClient
# Load configuration from environment variables
API_BASE_URL = os.environ["https://www.cutout.pro/api/v1/idphoto/printLayout"]
API_KEY = os.environ["dd597b46f810b75ee61"]
BLOB_STORAGE_CONNECTION_STRING = os.environ["DefaultEndpointsProtocol=https;AccountName=testo242;AccountKey=FjY904q11kzgrjfrGTaT5Xbb3CQlnVSBiHfCDrLfuvTPjE0PqsmU3ZzL+AStsUKoJA==;EndpointSuffix=core.windows.net"]
def download_img(img_url, file_name):
try:
request = urllib.request.Request(img_url)
response = urllib.request.urlopen(request)
if response.getcode() == 200:
with open(file_name, "wb") as f:
f.write(response.read())
return file_name
except:
return "failed"
def main(myblob: func.InputStream):
logging.info(f"Python blob trigger function processed blob"
f"Name: {myblob.name}"
f"Blob Size: {myblob.length}")
# Use blob data as input for API call
base64_binary = base64.b64encode(myblob.read())
base64_str = base64_binary.decode(encoding="utf-8")
url = f"{API_BASE_URL}/api/v1/idphoto/printLayout"
headers = {'APIKEY': API_KEY, "Content-type": "application/json"}
data = {
"base64": base64_str,
"bgColor": "FFFFFF",
"dpi": 300,
"mmHeight": 35,
"mmWidth": 25,
}
response = requests.post(url=url, headers=headers, json=data)
content = response.content.decode(encoding="utf-8")
json_result = json.loads(content)
if json_result["code"] == 0:
image_url = json_result["data"]["idPhotoImage"]
file_name = f'output/idphoto.jpg'
downloaded_file = download_img(image_url, file_name)
# Upload the final image to output container
blob_service_client = BlobServiceClient.from_connection_string(BLOB_STORAGE_CONNECTION_STRING)
container_client = blob_service_client.get_container_client("output")
blob_client = container_client.get_blob_client(downloaded_file)
with open(downloaded_file, "rb") as data:
blob_client.upload_blob(data, overwrite=True)
logging.info(f"Image uploaded to output container: {downloaded_file}")
Requeriments.txt
azure-functions requests azure-storage-blob
function.json
{
"scriptFile": "__init__.py",
"bindings": [
{
"name": "myblob",
"type": "blobTrigger",
"direction": "in",
"path": "input/{name}.jpg",
"connection": "testphot_STORAGE"
}
]
}
the config.py:
API_BASE_URL = 'https://www.someurl.pro'
API_KEY = 'Your APIKEY'
the utils.py:
import urllib.request
def download_img(img_url, file_name):
try:
request = urllib.request.Request(img_url)
response = urllib.request.urlopen(request)
if (response.getcode() == 200):
with open(file_name, "wb") as f:
f.write(response.read()) # Write content to picture
return file_name
except:
return "failed"
When I run the function i'm able to get the file from blob. But theres some errors in the log
*[2024-01-21T12:12:49.548Z] Host lock lease acquired by instance ID '000000000000000000000000000BC13F'. [2024-01-21T12:13:17.316Z] Executing 'Functions.BlobTrigger1' (Reason='New blob detected(LogsAndContainerScan): input/CHOA.jpg', Id=589ea295-2411-40e2-821c-690de21adf29) [2024-01-21T12:13:17.321Z] Trigger Details: MessageId: 526c7f78-3f83-43f9-8eea-f53a41ee194c, DequeueCount: 1, InsertedOn: 2024-01-21T12:13:16.000+00:00, BlobCreated: 2024-01-21T12:13:11.000+00:00, BlobLastModified: 2024-01-21T12:13:11.000+00:00 [2024-01-21T12:13:17.421Z] Executed 'Functions.BlobTrigger1' (Failed, Id=589ea295-2411-40e2-821c-690de21adf29, Duration=960ms) [2024-01-21T12:13:17.423Z] System.Private.CoreLib: Exception while executing function: Functions.BlobTrigger1. System.Private.CoreLib: Result: Failure Exception: KeyError: 'https://www.cutout.pro/api/v1/idphoto/printLayout' Stack: File "C:\Program Files\Microsoft\Azure Functions Core Tools\workers\python\3.11/WINDOWS/X64\azure_functions_worker\dispatcher.py", line 387, in _handle__function_load_request func = loader.load_function( ^^^^^^^^^^^^^^^^^^^^^ File "C:\Program Files\Microsoft\Azure Functions Core Tools\workers\python\3.11/WINDOWS/X64\azure_functions_worker\utils\wrappers.py", line 44, in call return func(*args, *kwargs) ^^^^^^^^^^^^^^^^^^^^^ File "C:\Program Files\Microsoft\Azure Functions Core Tools\workers\python\3.11/WINDOWS/X64\azure_functions_worker\loader.py", line 194, in load_function mod = importlib.import_module(fullmodname) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
Any help is wellcome
Solution
To achieve the desire workflow and for you to archive it, you might need to make some modification to your code.
You will need to configure the necessary application settings for your API_BASE_URL and API_KEY. Then, add a connection string for your Blob Storage.
Use the azure.storage.blob library to interact with Blob Storage after you must have import necessary modules and libraries. This will help in updating your Azure Function App.
Check out this sample code and compare with your work:
import azure.functions as func
import logging
import requests
import base64
import json
import os
import urllib.request
from azure.storage.blob import BlobServiceClient, BlobClient, ContainerClient
# Load configuration from environment variables
API_BASE_URL = os.environ["API_BASE_URL"]
API_KEY = os.environ["API_KEY"]
BLOB_STORAGE_CONNECTION_STRING = os.environ["BlobStorageConnectionString"]
def download_img(img_url, file_name):
try:
request = urllib.request.Request(img_url)
response = urllib.request.urlopen(request)
if response.getcode() == 200:
with open(file_name, "wb") as f:
f.write(response.read())
return file_name
except:
return "failed"
def main(myblob: func.InputStream):
logging.info(f"Python blob trigger function processed blob"
f"Name: {myblob.name}"
f"Blob Size: {myblob.length}")
# Use blob data as input for API call
base64_binary = base64.b64encode(myblob.read())
base64_str = base64_binary.decode(encoding="utf-8")
url = f"{API_BASE_URL}/api/v1/idphoto/printLayout"
headers = {'APIKEY': API_KEY, "Content-type": "application/json"}
data = {
"base64": base64_str,
"bgColor": "FFFFFF",
"dpi": 300,
"mmHeight": 35,
"mmWidth": 25,
}
response = requests.post(url=url, headers=headers, json=data)
content = response.content.decode(encoding="utf-8")
json_result = json.loads(content)
if json_result["code"] == 0:
image_url = json_result["data"]["idPhotoImage"]
file_name = f'output/idphoto_{time.strftime("%Y%m%d%H%M%S")}.jpg'
downloaded_file = download_img(image_url, file_name)
# Upload the final image to output container
blob_service_client = BlobServiceClient.from_connection_string(BLOB_STORAGE_CONNECTION_STRING)
container_client = blob_service_client.get_container_client("output-container")
blob_client = container_client.get_blob_client(downloaded_file)
with open(downloaded_file, "rb") as data:
blob_client.upload_blob(data, overwrite=True)
logging.info(f"Image uploaded to output container: {downloaded_file}")
Also, in your function.json file, set the input binding to trigger on blob changes in the input container.
{
"scriptFile": "__init__.py",
"bindings": [
{
"name": "myblob",
"type": "blobTrigger",
"direction": "in",
"path": "input-container/{name}",
"connection": "BlobStorageConnectionString"
}
]
}
Finally, deploy your function app and upload a blob to the input container. Ensure you do test. This should trigger the function, which will download the blob, make the API call, download the resulting image, and finally upload it to the output container. Also, make sure to replace "input-container" and "output-container" with the actual names of your input and output containers.
Answered By - Sina Salam
0 comments:
Post a Comment
Note: Only a member of this blog may post a comment.