Question 298637 in Azure Question posted in
The official documentation to get you started can be found here.

Azure function – Logging to Azure blob with python

I have an Azure function where I tried multiple ways to write/log, my python script logs to an Azure blob.
However, I can’t seem to make it work, even the documentation leads to a dead end.

This question has been asked before, but it’s a bit outdated.

The azure-storage-logging package also seems deserted.

Here is some example code I use for testing different methods:

import logging
import azure.functions as func

app = func.FunctionApp()

@app.schedule(schedule="*/30 * * * *", arg_name="myTimer", run_on_startup=True,
              use_monitor=False) 
def func_demo_timer_trigger(myTimer: func.TimerRequest) -> None:
    if myTimer.past_due:
        logging.info('The timer is past due!')

    logging.info('Python timer trigger function executed.')

How would I get this log message in to a blob?
Also, every run there is a log message it should be added to that same file.

(Application insights or related topics are not relevant).

Any tips and or small examples are welcome.

Tags:

2

Answers


  1. 0

    To log data to an Azure Blob Storage container from an Azure Function written in Python, one can use the azure-storage-blob library.

    import azure.functions as func
from azure.storage.blob import BlobServiceClient, BlobClient, ContainerClient##

    Ensure that you have the azure-functions and azure-storage-blob libraries installed.

    Login or Signup to reply.
  2. 0

    You can use below code to get required results:

    import logging
import azure.functions as func
from azure.storage.blob import BlobServiceClient
import tempfile

app = func.FunctionApp()
@app.schedule(schedule="*/30 * * * * *", arg_name="myTimer", run_on_startup=True,
              use_monitor=False) 
def timer_trigger(myTimer: func.TimerRequest) -> None:  
    logging.basicConfig(level=logging.INFO)
    Rithlogger = logging.getLogger(__name__)
    with tempfile.NamedTemporaryFile(mode='a', delete=False) as temp_file:
        temp_file_name = temp_file.name
    file_handler = logging.FileHandler(temp_file_name)
    file_handler.setLevel(logging.INFO)
    Rithformatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    file_handler.setFormatter(Rithformatter)
    Rithlogger.addHandler(file_handler)
    Rithlogger.info("Hello Rithwik, This is an info message")
    Rithlogger.warning("Hello Rithwik, This is a warning message")
    Rithlogger.error("Hello Rithwik, This is an error message")
    Rith_con_string = "DefaultEndpointsProtocol=https;AccountName=rithwik;AccountKey=pPBW9jWu77Sqpf+ASt5NLWkA==;EndpointSuffix=core.windows.net"
    Rith_con_name = "rithwik"
    blob_name = "RithwikTimerTriggerLogs"
    rithbsc = BlobServiceClient.from_connection_string(Rith_con_string)
    rithcc = rithbsc.get_container_client(Rith_con_name)
    rithb = rithcc.get_blob_client(blob_name)
    blob_exists = rithb.exists()
    if not blob_exists:
        with open(temp_file_name, "rb") as temp_file:
            rithcc.upload_blob(name=blob_name, data=temp_file, content_settings=ContentSettings(content_type='text/plain'))
        print(f"Blob '{blob_name}' created with content from the temporary file.")
    else:
        existing_content = rithb.download_blob().readall().decode('utf-8')
        with open(temp_file_name, "r") as temp_file:
            new_content = existing_content + temp_file.read()
        rithb.upload_blob(data=new_content, overwrite=True)

    requirements.txt:

    azure-functions
azure-storage-blob

    Output:

    enter image description here

    In blob:

    enter image description here

    Login or Signup to reply.
Please signup or login to give your own answer.

Back To Top
Search