def deferred_bindings_decode()

in azure_functions_worker/bindings/meta.py [0:0]

• 30 lines of code
• 7 McCabe index (conditional complexity)

def deferred_bindings_decode(binding: typing.Any,
                             pb: protos.ParameterBinding, *,
                             pytype: typing.Optional[type],
                             datum: typing.Any,
                             metadata: typing.Any,
                             function_name: str):
    """
    This cache holds deferred binding types (ie. BlobClient, ContainerClient)
    That have already been created, so that the worker can reuse the
    Previously created type without creating a new one.

    For async types, the function_name is needed as a key to differentiate.
    This prevents a known SDK issue where reusing a client across functions
    can lose the session context and cause an error.

    The cache key is based on: param name, type, resource, function_name

    If cache is empty or key doesn't exist, deferred_binding_type is None
    """
    global deferred_bindings_cache

    # Only applies to Event Hub and Service Bus - cannot cache
    # These types will always produce different content and are not clients
    if (datum.type == "collection_model_binding_data"
            or datum.value.source == "AzureEventHubsEventData"
            or datum.value.source == "AzureServiceBusReceivedMessage"):
        return binding.decode(datum,
                              trigger_metadata=metadata,
                              pytype=pytype)

    if deferred_bindings_cache.get((pb.name,
                                    pytype,
                                    datum.value.content,
                                    function_name), None) is not None:
        return deferred_bindings_cache.get((pb.name,
                                            pytype,
                                            datum.value.content,
                                            function_name))
    else:
        deferred_binding_type = binding.decode(datum,
                                               trigger_metadata=metadata,
                                               pytype=pytype)

        deferred_bindings_cache[(pb.name,
                                 pytype,
                                 datum.value.content,
                                 function_name)] = deferred_binding_type
        return deferred_binding_type