[Python] Handle cosmos DB connection in Function apps

I have a singleton class to instance the connection to az cosmos DB

functions/cosmos_client.py from azure.cosmos import CosmosClient, exceptions import threading import json import os

class CosmosDbClient: _cosmos_client = None _lock= threading.Lock()

def __new__(cls):
if not cls._cosmos_client: #validacion para unica instancia, bloqueando el hilo
with cls._lock:#lock thread, ensure all calls reference to unique instance
if not cls._cosmos_client:
cls._cosmos_client = super(CosmosClient,cls).__new__(cls)
#instance class attribute, self to access class methods & properties o
return cls._cosmos_client

def __init__(self):
CosmosUrl=None
CosmosDbKey = None
try:
CosmosUrl = os.getEnv("CosmosDbUri")
CosmosDbKey = os.getEnv("CosmosDbKey")
self._cosmos_client = CosmosClient(CosmosUrl,credentials = CosmosDbKey)
except json.JSONDecodeError:
raise json.JSONDecodeError("incorrect format")

def cosmos_client(self):
return self._cosmos_client


I'm trying to create a instance from an AZ function,

fn_role_description/function_app.py

import azure.functions as func from functions.cosmos_client import CosmosDbClient, exceptions import logging import json

class FnRoleDescription:

app = func.FunctionApp()
#db = FnAppHandler.cosmos_service.get_database_client('webresumedb')
@app.function_name(name="fn_role_descriptions")
@app.route(route="descriptions", auth_level=func.AuthLevel.FUNCTION)

def init(req: func.HttpRequest) -> func.HttpResponse:
logging.info('Python HTTP trigger function processed a request.')

Id = req.params.get('id')
item =None
if not Id:
try:
req_body = req.get_json()
cosmos_service = CosmosDbClient()
db = cosmos_service.get_database_client('webresumedb')
container = db.get_container_client('descriptions')

item = db.query_items(
query='SELECT * from descriptions WHERE Id=@model',
parameters =[dict(name = '@model', value = Id)]
)
except exceptions.CosmosResourceExistError:
raise
except exceptions.CosmosHttpResponseError:
raise


if not item:
return func.HttpResponse(
json.dumps(item),
mimetype="application/json",
status_code=200
)
#func.HttpResponse(HttpStatusCode.Ok)
else:
return func.HttpResponse("This HTTP triggered function executed successfully. Pass a name in the query string or in the request body for a personalized response.", status_code=200
)


Im currently stuck importing the singleton class

moduleNotFound


I would like to get your recommendations to get rid of this bug and also about methods, some recommendations?

my folder structure :

/webresume/


├── functions/ │ ├── init.py # (Optional) Marks the directory as a Python package │ ├── cosmos_client.py # Contains the CosmosDbClient class │ ├── fn_role_description/ │ │ ├── init.py # (Optional) Marks the directory as a Python package │ │ └── function_app.py Function logic

Thank you in advance

error ModuleNotfound

Continue reading...