-
Notifications
You must be signed in to change notification settings - Fork 0
/
timer_blob_logging_functions_sample.py
39 lines (34 loc) · 2.15 KB
/
timer_blob_logging_functions_sample.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
import azure.functions as func
import datetime
import logging
app = func.FunctionApp()
# in order for this function to run as intended, you will need to specify paths pointing to a valid storage container
# I have provided an easy way to create such containers via a powershell function named "Create-StorageContainer":
# your-ps-prompt>Create-StorageContainer "YourFunctionContainingGroup-rg" "yourstorageaccountsa" "test-slot-blobcontainer"
# after creating the storage container that will contain the blob file, you will need to setup (upload) a file (log.txt) that should be used as a target to write to by the sample functions
@app.function_name(name="timer_blob_appending_function")
@app.schedule(schedule="%Schedule%",
arg_name="req",
run_on_startup=False)
@app.blob_input(arg_name="inputblob",
path="%BlobOutpath%/log.txt", # using dynamic path config via appsettings (resolves to different container on default/prod functions and functions running in the test slot)
connection="")
@app.blob_output(arg_name="outputblob",
path="%BlobOutpath%/log.txt", # using dynamic path config via appsettings (resolves to different container on default/prod functions and functions running in the test slot)
connection="")
def appending_function(req: func.TimerRequest, inputblob: str, outputblob: func.Out[str]) -> None:
utc_timestamp = datetime.datetime.utcnow().replace(
tzinfo=datetime.timezone.utc).isoformat()
outputblob.set(inputblob + "\nPython timer-triggered function ran at " + str(utc_timestamp))
@app.function_name(name="timer_blob_wiping_function")
@app.schedule(schedule="0 59 0 * * *",
arg_name="req",
run_on_startup=False)
@app.blob_output(arg_name="outputblob",
path="%BlobOutpath%/log.txt",
connection="")
def wiping_function(req: func.TimerRequest, outputblob: func.Out[str]) -> None:
utc_timestamp = datetime.datetime.utcnow().replace(
tzinfo=datetime.timezone.utc).isoformat()
logging.info("Python timer-triggered function wiped the specified blob file at %s", utc_timestamp)
outputblob.set("")