python in memory cache

Question

I am trying to achieve in memory cache for a python application azure functions

/src
  /shared
    do_something.py
  /utils
    do_something_helper.py
  cache.py
  function_app.py

cache.py

class InMemoryCache:
    cache: dict = {}

    def get_cache(key):
        try:
            return InMemoryCache.cache[key]
        except KeyError:
            return None

    def set_cache(key, value):
        InMemoryCache.cache[key] = value

do_something_helper.py

from cache import InMemoryCache

def do_something_helper():
  blob_data = InMemoryCache.get_cache('blob_file_name')
  if blob_data:
    return blob_data

  blob_data = get_blob_data_from_storage()
  InMemoryCache.set_cache(key = 'blob_file_name', value = blob_data)
  return blob_data

do_something.py

from utils import do_something_helper

def do_something():
  # should be able to fetch from cache but always reading from blob
  blob_data = do_something_helper()
  process_blob_data()

function_app.py

import azure.functions as func
from shared.do_something import do_something

app = func.FunctionApp()

@app.function_name(name="hello")
@app.route('getBlobInfo',auth_level=func.AuthLevel.ANONYMOUS)
def get_blob_data(req):
  processed_data = do_something()
  return func.HttpResponse(body = processed_data, status_code = 200)

since azure functions app uses the same instance for multiple calls, The Aim is to reduce the number of calls to blob storage that improves performance.

expected behaviour: should be able to fetch from cache first but always reading from blob. I want this cache accessible globally for the whole project.

Am I doing Something wrong here? reference: https://learn.microsoft.com/en-us/azure/azure-functions/functions-reference-python?tabs=asgi%2Capplication-level&pivots=python-mode-decorators#global-variables

edit : 2 here is the representation of get_blob_data_from_storage() , process_blob_data()

from azure.storage.blob import BlobServiceClient, BlobClient
import json


def get_blob_data_from_storage(file_name):
    container_name = "data"
    BLOB_STORAGE_CONNECTION_STRING = "<connection string>"
    try:
        blob_service_client = BlobServiceClient.from_connection_string(
            BLOB_STORAGE_CONNECTION_STRING
        )
    except Exception as e:
        logging.error(e)
        raise Exception(e)

    if blob_service_client:
        try:
            blob_client: BlobClient = blob_service_client.get_blob_client(
                container=container_name, blob=file_name
            )
        except Exception as e:
            logging.error(e)
            raise Exception(e)
        data = blob_client.download_blob()
        data1 = data.readall()
        return data1


def process_blob_data(blob_data):
    json_data = json.loads(blob_data)
    # match records based on request query
    return json.dumps(processed_json_data)

Answer 1

I have modified your code and it works without error but you need to incorporate a method to save the data in cache before using it.

function_app.py-

import azure.functions as func
from src.shared.do_something import do_something

app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS)

@app.route(route="getBlobInfo")
def get_blob_data(req: func.HttpRequest) -> func.HttpResponse:
    processed_data = do_something()
    return func.HttpResponse(body = processed_data, status_code = 200)

cache-

class InMemoryCache:
    cache: dict = {}

    @staticmethod
    def get_cache(key):
        return InMemoryCache.cache.get(key)

    @staticmethod
    def set_cache(key, value):
        InMemoryCache.cache[key] = value

do_something-

from src.utils.do_something_helper import do_something_helper

def do_something():
  blob_data = do_something_helper()
  return blob_data

do_something_helper-

from cache import InMemoryCache

def do_something_helper():
  blob_data = InMemoryCache.get_cache('blob_file_name')
  if blob_data:
    return blob_data
  InMemoryCache.set_cache(key = 'blob_file_name', value = blob_data)
  return blob_data

Output-

Alternatively, you can also use the below code to invoke an Api to get the Json response and then save it to in memory cache to use further.

import azure.functions as func
import json
import requests

app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS)

@app.route(route="http_trigger")
def http_trigger(req: func.HttpRequest) -> func.HttpResponse:
    # Define an in-memory cache
    in_memory_cache = {}
 
    def fetch_data_from_endpoint():
        response = requests.get('https://reqres.in/api/users?page=2')
        return response.json()
    
    # Check if data is already in cache
    if 'cached_data' in in_memory_cache:
        # Retrieve data from cache
        data = in_memory_cache['cached_data']
    else:
        # Fetch data from endpoint
        data = fetch_data_from_endpoint()
        # Store data in cache
        in_memory_cache['cached_data'] = data
 
    # Convert data to JSON response
    response_data = json.dumps(data)
 
    return func.HttpResponse(
        body=response_data,
        mimetype="application/json",
        status_code=200
    )