function_app.py

import dataclasses
import json
import logging
import os

import azure.functions as func

from api import cases
from api import config
from api import util



app = func.FunctionApp()
conf = config.get_config()
config.configure_logging()



@app.function_name(name="process")
@app.route(route="process")
def process(req: func.HttpRequest) -> func.HttpResponse:
    logging.info("Start processing.")
    cases.process_cases()
    return func.HttpResponse(f"Processed.")


@app.function_name(name="process_image")
@app.queue_trigger(arg_name="msg", queue_name="case-images-deduplication", connection="AzureStorageConnectionString")
def process_image(msg: func.QueueMessage) -> None:
    logging.info('Python EventHub trigger processed an event: %s', msg.get_body().decode('utf-8'))
    #logging.info(f"Start processing image: { msg.get_body() }")
    #case_image = util.dataclass_from_dict(cases.CaseImage, msg.get_json()) 
    #logging.info(f"Start processing an image from a case { case_image.case_id } with url { case_image.image_url }.")

lib-api-cases.py

import dataclasses
import json
import logging
from dataclasses import dataclass

from azure.storage.queue import QueueClient

from api import config


@dataclass
class CaseImage:
    case_id: str
    image_url: str


def process_cases() -> None:
    conf = config.get_config()
    queue_name = "case-images-deduplication"
    storage_account_name = conf["STORAGE_ACCOUNT_NAME"]
    storage_account_key = conf["STORAGE_ACCOUNT_KEY"]
    conn_str = f"DefaultEndpointsProtocol=https;AccountName={storage_account_name};AccountKey={storage_account_key};EndpointSuffix=core.windows.net"
    queue_client = QueueClient.from_connection_string(conn_str, queue_name)

    case_images = get_case_images_batch()
    for case_image in case_images:
        logging.info(f"Adding { case_image.case_id } with image { case_image.image_url } to the queue.")
        queue_item = json.dumps(dataclasses.asdict(case_image))
        queue_client.send_message(queue_item)


def get_case_images_batch() -> list[CaseImage]:
    case_images: list[CaseImage] = []
    import random
    import string
    alphabet = string.ascii_lowercase + string.digits
    for i in range(3):
        case_id = "".join(random.choices(alphabet, k=20))
        filename = "".join(random.choices(alphabet, k=8))
        image_url = f"http://img.us/{ filename }"
        case_images.append(CaseImage(case_id=case_id, image_url=image_url))
    return case_images