内存缓存中的python

Question

我正在尝试在内存缓存中实现Python应用程序的azure函数

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

缓存.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)

由于 Azure Functions 应用程序使用同一实例进行多个调用，因此目的是减少对 Blob 存储的调用次数，从而提高性能。

预期行为：应该能够首先从缓存中获取 但总是从 blob 中读取。 我希望整个项目可以全局访问此缓存。

我在这里做错了什么吗？ 参考： https://learn.microsoft.com/en-us/azure/azure-functions/functions-reference-python?tabs=asgi%2Capplication-level&pivots=python-mode-decorators#global-variables

编辑：2 这是 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

我已经修改了您的代码，它可以正常工作，但您需要在使用之前合并一种方法来将数据保存在缓存中。

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)

缓存-

class InMemoryCache:
    cache: dict = {}

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

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

做某事-

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

输出-

或者，您也可以使用以下代码调用 Api 来获取 Json 响应，然后将其保存到内存缓存中以供进一步使用。

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
    )