In this post we are going to cover security and configuration in Azure Functions: 7 real Azure Functions interview questions for data engineers, asked the way they come up on a real project. The last one is a scenario: a function must read from a locked-down storage account and write to a private SQL database, secured end to end.
Security in Functions comes down to four things: where the secrets live, how the app proves who it is, who is allowed to call it, and which networks it can reach. Configuration is how the same code runs in dev, test and prod with different settings. Interviewers ask about it because a security review is part of every real project, and one leaked key is enough to fail it.
Part 8 of the Real Time Azure Functions Interview Questions and Answers series on TechBrothersIT.
- The short answer
- Let's set up the project
- Question 1: A security review asks where function secrets are stored. How do app settings and Key Vault references work?
- Question 2: Someone asks how a function connects to storage without connection strings. How do identity-based connections work?
- Question 3: A teammate asks what function keys and authorization levels are. Are they enough to protect an API?
- Question 4: Security wants the function app off the public internet. What networking options do you use?
- Question 5: Someone asks how to keep settings different for dev, test and prod. What do you set up?
- Question 6: A teammate committed local.settings.json with a real key. What do you do?
- Scenario 7: A function must read from a locked-down storage account and write to a private SQL database. How do you secure it end to end?
- The full script
The short answer
Secrets live in Key Vault references, read as app settings; storage, queues and Event Hubs use identity-based connections with least-privilege roles. Function keys aren't authentication: put Microsoft Entra ID or API Management in front. Lock the network on Flex Consumption or Premium, keep one app and one vault per environment, and rotate anything that leaks.
Let's set up the project
We run the real Azure Functions host locally with Core Tools, with Azurite, Microsoft's open-source storage emulator, for storage. The main app is a small Python function app, orders_app, with its settings in local.settings.json. What only Azure can show, such as managed identity, Key Vault and networking, is shown as code.
{
"version": "2.0",
"extensionBundle": {
"id": "Microsoft.Azure.Functions.ExtensionBundle",
"version": "[4.*, 5.0.0)"
},
"extensions": {
"queues": {"batchSize": 1, "newBatchThreshold": 0,
"maxPollingInterval": "00:00:01"}
},
"logging": {
"logLevel": {"default": "Warning", "Function": "Information",
"Azure.Core": "Error",
"Host.Function.Console": "Information"}
}
}azure-functions{
"IsEncrypted": false,
"Values": {
"FUNCTIONS_WORKER_RUNTIME": "python",
"AzureWebJobsStorage": "UseDevelopmentStorage=true",
"SOURCE_SYSTEM": "oracle-fusion",
"DB_PWD": "@Microsoft.KeyVault(VaultName=kv-dev;SecretName=dbpw)",
"PARTNER_URL": "https://partner.example.com/api"
}
}Question 1: A security review asks where function secrets are stored. How do app settings and Key Vault references work?
App settings are encrypted and injected as environment variables. A real secret stays in Key Vault: the app setting holds a Key Vault reference, and the app's managed identity resolves it.
Let's go through it in order:
- App settings: encrypted at rest, decrypted into the process at start
- Python:
os.getenv("NAME")is always a string, or None - Function keys: blob storage of
AzureWebJobsStorageby default, or Key Vault
A Key Vault reference names the vault and the secret. The managed identity needs only the Key Vault Secrets User role:
# 1. the app gets a system-assigned managed identity
az functionapp identity assign -g rg-data -n fn-orders
# 2. it may read secrets, nothing more
az role assignment create --assignee <principal-id> \
--role "Key Vault Secrets User" --scope <vault-id>
# 3. the app setting holds a reference, not the secret
REF="@Microsoft.KeyVault(VaultName=kv-prod;SecretName=dbpw)"
az functionapp config appsettings set -g rg-data -n fn-orders \
--settings "DB_PWD=$REF"Locally, the host does not resolve references. Our timer config_check reports the state of four settings, never the values. We start the host and run it on demand through the local admin API with fire.py:
import logging, os
import azure.functions as func
app = func.FunctionApp()
KV_REF = "@Microsoft.KeyVault("
NAMES = ["SOURCE_SYSTEM", "DB_PWD", "PARTNER_URL", "SFTP_PASSWORD"]
@app.timer_trigger(schedule="0 0 6 * * *", arg_name="t",
run_on_startup=False)
def config_check(t: func.TimerRequest) -> None:
for name in NAMES:
value = os.getenv(name) # app settings = env vars
if value is None:
state = "missing"
elif value.startswith(KV_REF):
state = "Key Vault reference, not resolved"
else:
state = "set (value never logged)"
logging.info("%s: %s", name, state){
"IsEncrypted": false,
"Values": {
"FUNCTIONS_WORKER_RUNTIME": "python",
"AzureWebJobsStorage": "UseDevelopmentStorage=true",
"SOURCE_SYSTEM": "oracle-fusion",
"DB_PWD": "@Microsoft.KeyVault(VaultName=kv-dev;SecretName=dbpw)",
"PARTNER_URL": "https://partner.example.com/api"
}
}import sys, requests
# run a timer now through the admin API (Azure: needs the master key)
URL = "http://localhost:7071/admin/functions/"
r = requests.post(URL + sys.argv[1], json={"input": ""})
print("run", sys.argv[1], "->", r.status_code)$ cd orders_app && fhost start
host started, functions:
config_check: timerTrigger
$ python fire.py config_check
run config_check -> 202
$ cd orders_app && fhost wait "Executed 'Functions.config_check'" 30
$ cd orders_app && fhost log && fhost stop
Executing 'Functions.config_check' (Reason='This function was programmatically called via the host
APIs.')
SOURCE_SYSTEM: set (value never logged)
DB_PWD: Key Vault reference, not resolved
PARTNER_URL: set (value never logged)
SFTP_PASSWORD: missing
Executed 'Functions.config_check' (Succeeded)
host stoppedDB_PWD shows Key Vault reference, not resolved, SOURCE_SYSTEM and PARTNER_URL are set (value never logged), and SFTP_PASSWORD is missing.
In Azure, an unresolved reference reaches your code as that same literal string, and the portal shows the status of each reference. Versionless references refresh within 24 hours, or on restart.
Give each environment its own Key Vault, and never put a secret in code or in a log line.
Question 2: Someone asks how a function connects to storage without connection strings. How do identity-based connections work?
The app uses its managed identity instead of a key. In code nothing changes, because a binding's connection property is only a prefix for app settings.
Here the binding says connection="OrdersQueue", and there is no secret anywhere in the code:
@app.queue_trigger(arg_name="msg", queue_name="orders",
connection="OrdersQueue") # no secret in code
def load_order(msg: func.QueueMessage) -> None:
...Then the app settings carry the identity details. For host storage, AzureWebJobsStorage__accountName is enough. The queue binding reads OrdersQueue__queueServiceUri. Event Hubs and Service Bus use fullyQualifiedNamespace. For a user-assigned identity, also set credential and clientId:
# host storage: no key, the app's managed identity
AzureWebJobsStorage__accountName = stordersprod
# a binding: connection="OrdersQueue" is only a prefix
OrdersQueue__queueServiceUri = https://stprod.queue.core.windows.net
# user-assigned identity: say which one
OrdersQueue__credential = managedidentity
OrdersQueue__clientId = <client-id>
# Event Hubs and Service Bus
EventsHub__fullyQualifiedNamespace = ehorders.servicebus.windows.netThen grant data roles, least privilege, never Owner:
| Connection | Roles for the managed identity |
|---|---|
| Host storage (AzureWebJobsStorage) | Storage Blob Data Owner (+ Storage Table Data Contributor) |
| Queue trigger | Storage Queue Data Reader + Storage Queue Data Message Processor |
| Blob trigger | Storage Blob Data Owner + Storage Queue Data Contributor |
| Event Hubs trigger | Azure Event Hubs Data Receiver |
This works on runtime 4.x only. Role changes can take up to 10 minutes, so a 403 right after an assignment can be temporary. Locally, your own developer identity is used instead, and against Azurite identity connections do not work at all, so this part is code.
The Azure Files content share on Consumption and Premium cannot use identity, so keep that one in Key Vault. Flex Consumption has no content share.
Question 3: A teammate asks what function keys and authorization levels are. Are they enough to protect an API?
The authorization level says which key a caller must send, and no, keys alone are not enough. A key is a shared secret, not an identity: anyone holding it gets in.
Let's look at the levels. Anonymous needs no key. Function needs a function key or a host key. Admin needs the master key. Callers send the key as the code query parameter or the x-functions-key header. In Python I always set the level explicitly.
Our app has three routes: orders at function level, purge at admin level, and health as anonymous. We call each one without a key using call.py, then list the key names:
import azure.functions as func
app = func.FunctionApp(http_auth_level=func.AuthLevel.FUNCTION)
@app.route(route="orders", methods=["GET"]) # function or host key
def orders(req: func.HttpRequest) -> func.HttpResponse:
return func.HttpResponse("3 open orders")
@app.route(route="purge", auth_level=func.AuthLevel.ADMIN)
def purge(req: func.HttpRequest) -> func.HttpResponse: # master key
return func.HttpResponse("purged")
@app.route(route="health", auth_level=func.AuthLevel.ANONYMOUS)
def health(req: func.HttpRequest) -> func.HttpResponse:
return func.HttpResponse("ok")import requests
BASE = "http://localhost:7071"
for route in ["orders", "purge", "health"]:
r = requests.get(f"{BASE}/api/{route}") # no key at all
print(f"GET /api/{route} without a key -> {r.status_code}")
fn = requests.get(f"{BASE}/admin/functions/orders/keys").json()
print("orders function keys:", [k["name"] for k in fn["keys"]])
host = requests.get(f"{BASE}/admin/host/keys").json()
print("host keys:", [k["name"] for k in host["keys"]]) # names only$ cd orders_app && fhost start
host started, functions:
health: http://localhost:7071/api/health
orders: [GET] http://localhost:7071/api/orders
purge: http://localhost:7071/api/purge
$ python call.py
GET /api/orders without a key -> 200
GET /api/purge without a key -> 200
GET /api/health without a key -> 200
orders function keys: ['default']
host keys: ['default']
$ cd orders_app && fhost stop
host stoppedAll three return 200, because Core Tools does not enforce keys locally. The keys still exist: orders function keys: ['default'] and host keys: ['default']. In Azure the same call without a key returns 401:
$ curl -s -o /dev/null -w "%{http_code}\n" \
https://fn-orders.azurewebsites.net/api/orders
401
$ curl -s -H "x-functions-key: $ORDERS_KEY" \
https://fn-orders.azurewebsites.net/api/orders
3 open ordersFor a real API, I put something with real identity in front:
| Option | Who is the caller? | Use it for |
|---|---|---|
| Function keys | anyone holding the shared secret | an extra layer; rotate on a schedule |
| App Service built-in auth (Easy Auth) | a Microsoft Entra ID identity; 401 if not signed in | APIs for people and services |
| API Management in front | API Management; the app accepts only it | public APIs, many consumers |
Keys stay useful as an extra layer. Rotating one key revokes just that key, so rotate on a schedule.
The master key is an admin credential. Never hand it out.
Question 4: Security wants the function app off the public internet. What networking options do you use?
I split it by direction. Inbound is who can call the app: access restrictions and private endpoints. Outbound is how the app reaches storage, SQL and Key Vault: virtual network integration.
Let's compare the plans:
| Plan | Inbound: access restrictions | Inbound: private endpoint | Outbound: VNet integration |
|---|---|---|---|
| Flex Consumption | yes | yes | yes (all outbound traffic) |
| Premium | yes | yes | yes |
| Dedicated (App Service) | yes | yes | yes |
| Consumption (legacy) | yes | no | no |
The Consumption plan has neither VNet integration nor inbound private endpoints, only access restrictions. That is one more reason new apps go on Flex Consumption.
Then plan the subnet:
- Flex: subnet /27 minimum, delegated to
Microsoft.App/environments - Premium: subnet /28 minimum, delegated to
Microsoft.Web/serverFarms - Plan the size: a subnet cannot be resized after it is assigned
Flex routes all outbound traffic through the VNet. As code, outbound integration plus an inbound rule that only allows the API Management subnet:
# outbound: reach storage, SQL and Key Vault through the VNet
az functionapp vnet-integration add -g rg-data -n fn-orders \
--vnet vnet-data --subnet snet-functions
# inbound: only the API Management subnet may call the app
az functionapp config access-restriction add -g rg-data -n fn-orders \
--rule-name apim-only --action Allow --priority 100 \
--vnet-name vnet-data --subnet snet-apimPrivate endpoints only work when DNS resolves privately, so add private DNS zones or forwarding. And with no access restriction rules, the app accepts traffic from any address.
An Event Grid based blob trigger cannot push to a private endpoint, so a fully private app needs another pattern, like a queue.
Question 5: Someone asks how to keep settings different for dev, test and prod. What do you set up?
Same code, different settings. Every name that changes per environment, like queues, databases and URLs, comes from an app setting.
In a binding, wrap the setting name in percent signs. Here load_order listens on %ORDERS_QUEUE% and reads APP_ENV and SQL_DATABASE from the environment. local.settings.json says dev. We start the host and send order o-1 to orders-dev:
import logging, os
import azure.functions as func
app = func.FunctionApp()
@app.queue_trigger(arg_name="msg", queue_name="%ORDERS_QUEUE%",
connection="AzureWebJobsStorage")
def load_order(msg: func.QueueMessage) -> None:
body = msg.get_body().decode()
env, db = os.environ["APP_ENV"], os.environ["SQL_DATABASE"]
logging.info("%s: load %s into %s", env, body, db){
"IsEncrypted": false,
"Values": {
"FUNCTIONS_WORKER_RUNTIME": "python",
"AzureWebJobsStorage": "UseDevelopmentStorage=true",
"APP_ENV": "dev",
"ORDERS_QUEUE": "orders-dev",
"SQL_DATABASE": "dw_dev"
}
}{
"version": "2.0",
"extensionBundle": {
"id": "Microsoft.Azure.Functions.ExtensionBundle",
"version": "[4.*, 5.0.0)"
},
"extensions": {
"queues": {"batchSize": 1, "newBatchThreshold": 0,
"maxPollingInterval": "00:00:01"}
},
"logging": {
"logLevel": {"default": "Warning", "Function": "Information",
"Azure.Core": "Error",
"Host.Function.Console": "Information"}
}
}azure-functionsimport os, sys
from azure.storage.queue import QueueClient, TextBase64EncodePolicy
q = QueueClient.from_connection_string(
os.environ["AZURE_STORAGE_CONNECTION_STRING"], sys.argv[1],
message_encode_policy=TextBase64EncodePolicy())
try:
q.create_queue()
except Exception:
pass # already there
for body in sys.argv[2:]:
q.send_message(body)
print("sent to", sys.argv[1] + ":", ", ".join(sys.argv[2:]))$ cd env_app && fhost start
host started, functions:
load_order: queueTrigger
$ python send.py orders-dev o-1
sent to orders-dev: o-1
$ cd env_app && fhost wait 'Executed' 30 && fhost log && fhost stop
Executing 'Functions.load_order' (Reason='New queue message detected on 'orders-dev'.')
dev: load o-1 into dw_dev
Executed 'Functions.load_order' (Succeeded)
host stoppedThe log says dev: load o-1 into dw_dev. A shell variable wins over the file, so now we start the same code with test values and send o-2 to orders-test:
$ cd env_app && APP_ENV=test ORDERS_QUEUE=orders-test SQL_DATABASE=dw_test fhost start
host started, functions:
load_order: queueTrigger
$ python send.py orders-test o-2
sent to orders-test: o-2
$ cd env_app && fhost wait 'Executed' 30 && fhost log && fhost stop
Executing 'Functions.load_order' (Reason='New queue message detected on 'orders-test'.')
test: load o-2 into dw_test
Executed 'Functions.load_order' (Succeeded)
host stoppedThis time it says test: load o-2 into dw_test. Not one line of code changed.
In Azure, I set it up like this:
- One function app per environment
- One Key Vault per environment
- Slot settings, which stay with the slot in a swap
- Infrastructure as code: the same template, other values
host.json is shared by every environment, but any value can be overridden with an app setting that starts with AzureFunctionsJobHost__ and the path:
# prod: its own app, its own settings, its own vault
az functionapp config appsettings set -g rg-prod -n fn-orders-prod \
--settings APP_ENV=prod ORDERS_QUEUE=orders SQL_DATABASE=dw
# a slot: values that stay with the slot during a swap
az functionapp config appsettings set -g rg-prod -n fn-orders-prod \
--slot staging --slot-settings APP_ENV=staging
# override a host.json value for one environment
az functionapp config appsettings set -g rg-prod -n fn-orders-prod \
--settings AzureFunctionsJobHost__functionTimeout=00:10:00Dev and prod sharing one storage account or one vault. Keep them apart, so a dev mistake cannot touch production.
Question 6: A teammate committed local.settings.json with a real key. What do you do?
Treat it as leaked the moment it is pushed, even if the file is deleted a minute later, and rotate the key at the source first.
Let's go in order:
- Rotate at the source: storage account key, function key; the old value stops working
- Purge the history: deleting the file leaves it in old commits
- Review access: who could read the repository, and the storage logs for the old key
# 1. rotate at the source: the old values stop working
az storage account keys renew -g rg-data -n stordersdev --key primary
az functionapp function keys set -g rg-data -n fn-orders-dev \
--function-name orders --key-name default # new random valuePrevention is built in. Let's create a project with func init, commit everything, and see what git does with the settings file:
$ func init orders_api --worker-runtime python --model V2
The new Python programming model is generally available. Learn more at
https://aka.ms/pythonprogrammingmodel
Writing requirements.txt
Writing function_app.py
Writing .gitignore
Writing host.json
Writing local.settings.json
Writing orders_api/.vscode/extensions.json
$ cd orders_api && git init -q && git add -A && git ls-files
.gitignore
.vscode/extensions.json
function_app.py
host.json
requirements.txt
$ cd orders_api && git check-ignore -v local.settings.json
.gitignore:25:local.settings.json local.settings.json
$ cd orders_api && git add -f local.settings.json && git status -s local.settings.json
A local.settings.json
$ cd orders_api && git rm -q --cached local.settings.json && git ls-files | wc -l
5func init writes a .gitignore, and git ls-files lists five files without local.settings.json. git check-ignore -v names the rule: .gitignore:25:local.settings.json. The usual leak is git add -f, which bypasses the ignore file, and the status shows A local.settings.json. Undo that with git rm --cached, and we are back to 5 files.
Better still, keep nothing secret in the file: Azurite and your own sign-in locally, identity-based connections and Key Vault in Azure.
I also add a secret scanner to the pipeline, so the next one is caught before the push.
Scenario 7: A function must read from a locked-down storage account and write to a private SQL database. How do you secure it end to end?
I secure it in layers, and the plan comes first: Flex Consumption or Premium, because the Consumption plan cannot join a virtual network.
Let's walk the picture: API Management is the only caller, the function app has VNet integration and a managed identity, and the storage account, the SQL database and Key Vault each sit behind a private endpoint.
First the network: VNet integration for outbound traffic, and private endpoints for storage, SQL and Key Vault, each with a private DNS zone. Then disable public network access on storage and SQL.
Then identity: a managed identity with identity-based connections and the data roles we saw, plus Microsoft Entra authentication to SQL, so there is no password anywhere. Whatever secret is left lives in Key Vault, read over the VNet.
Then inbound: only API Management may call the app, through an access restriction or a private endpoint. If you filter outbound traffic, allow the extension bundle download from the Functions CDN.
Before go-live, I close the public doors with code like this:
# a private endpoint for the blob service (repeat: queue, SQL, vault)
az network private-endpoint create -g rg-data -n pe-st-blob \
--vnet-name vnet-data --subnet snet-pe --group-id blob \
--private-connection-resource-id $ST_ID --connection-name st-blob
# then close the public doors
az storage account update -g rg-data -n stordersprod \
--public-network-access Disabled
az sql server update -g rg-data -n sql-orders \
--enable-public-network falseThen I test from outside the network:
- Storage refuses a call from outside
- SQL refuses a public login
- The function app is reachable only through API Management
Every layer covers for another. Private endpoints with public access still on, or identity with a password still in a setting, is only half done.
Key points to remember
- Secrets in Key Vault references, read as app settings; never in code or logs
- Identity-based connections: managed identity + least-privilege data roles, no connection strings
- Function keys aren't authentication: Microsoft Entra ID (Easy Auth) or API Management in front
- Off the internet: private endpoints in, VNet integration out, on Flex Consumption or Premium
- One app and one vault per environment; a leaked key is rotated at the source first
The full script
This Python script writes every file and runs every command in order with Core Tools and Azurite; run pip install azure-functions azure-functions-durable azure-storage-blob azure-storage-queue azure-data-tables first.
Show the full script that writes the files and runs every step
# ========================================================================
# Azure Functions Interview Q&A #8 - Security and configuration
# TechBrothersIT. Names and sample data are invented for teaching.
# Every step below was run with Azure Functions Core Tools 4.16.0 (Functions host 4.x, Python 3.13 worker,
# programming model v2, extension bundle 4.x) and Azurite 3.37.0; the outputs in the video and the blog post are the real outputs.
# To run it yourself (Linux or macOS shell, or WSL on Windows):
# 1. Install Azure Functions Core Tools 4 and Azurite (npm install -g azurite), then start Azurite:
# azurite --skipApiVersionCheck --location ./azurite-data (blob 10000, queue 10001, table 10002)
# 2. pip install azure-functions azure-functions-durable azure-storage-blob azure-storage-queue azure-data-tables
# 3. Set AZURE_STORAGE_CONNECTION_STRING to "UseDevelopmentStorage=true", then run python Functions_Interview_QA_08_Code.py.
# "fhost" in the commands is a small helper written by Step 0 (fhost.py): it runs `func start` in the background,
# waits until the functions are loaded, and prints the host log lines without timestamps and invocation ids.
# Run it in an empty folder and on a freshly started emulator: it creates its files (and any local data
# files) there, and a second run on old data gives different output.
# Steps that restart the emulator say so in a comment: stop and start it by hand at that point.
# ========================================================================
import os, subprocess, time
def write(path, text):
if text is None:
if os.path.exists(path): os.remove(path)
return
os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
with open(path, "w", encoding="utf-8") as f:
f.write(text.strip("\n") + "\n")
def run(cmd, expect_error=False, allow_error=False, env=None):
print("\n$ " + cmd, flush=True)
rc = subprocess.run(cmd, shell=True, env=dict(os.environ, **(env or {}))).returncode
if rc != 0 and not (expect_error or allow_error):
raise SystemExit("stopped: this command was expected to succeed")
# Step 0: the fhost helper and the starting files
write("fhost.py", '"""fhost: run the Azure Functions host (func start) in the background of the current folder, for scripted demos.\n fhost start [port] [-- extra func args] start `func start`, wait until the functions are indexed, print them\n (and any indexing errors)\n fhost wait PATTERN [secs] [count] wait until `count` (default 1) NEW host log lines (not yet shown by\n `fhost log`) match the regex PATTERN (default timeout 60 s)\n fhost log [--all] [--details] [--since-start]\n print the host log lines written since the last `fhost log` (function logs,\n Executing / Executed lines, exceptions); --details also keeps the\n "Trigger Details" lines, --all keeps every line, --since-start starts at the\n first line after the function list. Timestamps, invocation ids, durations and\n other per-run values are removed.\n fhost stop stop the host\nThe log is kept in .fhost.log in the current folder (one host per folder)."""\nimport os, re, signal, subprocess, sys, time\nLOG, PID, OFF, START = \'.fhost.log\', \'.fhost.pid\', \'.fhost.off\', \'.fhost.start\'\nTS = re.compile(r\'\\[\\d{4}-\\d\\d-\\d\\dT[\\d:.]+Z\\] \')\n# host messages to hide (anchored at the start of the cleaned line, so user log lines that merely contain these words stay)\nNOISE = re.compile(r\'^(Worker process started|Host lock lease|Found Python version|Azure Functions Core Tools|Core Tools Version|\'\n r\'Function Runtime Version|For detailed output|Using cached version|When you have network connectivity|\'\n r\'Warning: Unable to download extension|Host started|Job host started|Host initialized|Loading functions metadata|\'\n r\'Reading functions metadata|Starting Host|Initializing Warmup|Generating \\d+ job function|Functions:$|\'\n r\'Debugger listening|Host Status|Starting worker process|Worker failed to load|Retrieving function|\'\n r\'\\d+ functions? (found|loaded)|Successfully (found|loaded|registered)|Application Insights|WARNING: )\')\nDETAILS = re.compile(r\'^Trigger Details:\')\n\ndef clean(l):\n l = TS.sub(\'\', l.rstrip(\'\\n\'))\n l = re.sub(r\', Id=[0-9a-f-]{36}\', \'\', l)\n l = re.sub(r\'Id=[0-9a-f-]{36}\', \'\', l)\n l = re.sub(r\', Duration=\\d+ms\', \'\', l)\n l = re.sub(r\'\\b[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}\\b\', \'<id>\', l)\n l = re.sub(r\'\\b[0-9a-f]{32}\\b\', \'<instance-id>\', l)\n l = re.sub(r\'\\d{4}-\\d\\d-\\d\\dT\\d\\d:\\d\\d:\\d\\d(\\.\\d+)?(Z|[+-]\\d\\d:\\d\\d)?\', \'<time>\', l)\n l = re.sub(r\'"?0x[0-9A-F]{15,16}"?\', \'<etag>\', l)\n l = re.sub(r\'([?&]code=)[^&\\s"]+\', r\'\\1<key>\', l)\n l = re.sub(r\'Threadpool id: \\d+\', \'Threadpool id: <n>\', l)\n l = re.sub(r\'(processUptime\\W+)\\d+\', r\'\\1<n>\', l)\n return l.replace(\' ()\', \'\').rstrip()\n\ndef read_from(off):\n """complete lines from byte offset `off`; returns (lines, new_offset). A glued line ([ts] a [ts] b) is split."""\n if not os.path.exists(LOG): return [], off\n with open(LOG, \'rb\') as f:\n f.seek(off); data = f.read()\n end = data.rfind(b\'\\n\')\n if end < 0: return [], off\n chunk = data[:end + 1].decode(\'utf-8\', \'replace\')\n out = []\n for l in chunk.split(\'\\n\')[:-1]:\n parts = [p for p in re.split(r\'(?=\\[\\d{4}-\\d\\d-\\d\\dT[\\d:.]+Z\\] )\', l) if p.strip()] or [\'\']\n out += parts\n return out, off + end + 1\n\ndef get(f, d=0):\n return int(open(f).read()) if os.path.exists(f) else d\n\ndef filtered(lines, all_=False, details=False):\n out, in_stack, last_exc = [], False, None\n for l in lines:\n raw = TS.sub(\'\', l)\n if in_stack:\n if raw.startswith((\' \', \'\\t\')): continue\n in_stack = False\n c = clean(l)\n if not c: continue\n if c.startswith(\'Stack:\') and not all_: in_stack = True; continue\n if not all_ and (c in (\'Type:\', \'.\') or c == last_exc): continue\n if c.startswith(\'Exception: \'): last_exc = c[len(\'Exception: \'):]\n if all_ or (details and DETAILS.search(c)): out.append(c); continue\n if NOISE.search(c) or DETAILS.search(c): continue\n out.append(c)\n return out\n\ndef start(*args):\n port, extra = \'7071\', []\n if args and args[0] != \'--\': port = args[0]; args = args[1:]\n if args and args[0] == \'--\': extra = list(args[1:])\n if os.path.exists(PID): stop(quiet=True)\n for f in (LOG, OFF, START):\n if os.path.exists(f): os.remove(f)\n out = open(LOG, \'w\')\n p = subprocess.Popen([\'func\', \'start\', \'--port\', port] + extra, stdout=out, stderr=subprocess.STDOUT,\n stdin=subprocess.DEVNULL, start_new_session=True)\n open(PID, \'w\').write(str(p.pid))\n t0 = time.time()\n while time.time() - t0 < 120:\n txt = open(LOG, encoding=\'utf-8\', errors=\'replace\').read()\n if p.poll() is not None:\n print(\'\\n\'.join(filtered(txt.split(\'\\n\'), all_=True)[-30:])); sys.exit(\'host exited\')\n m = re.search(r\'\\nFunctions:\\n(.*?)\\n(For detailed output[^\\n]*\\n)\', txt, re.S)\n if m and \'Worker process started and initialized\' in txt:\n errs = [clean(l) for l in txt[:m.start()].split(\'\\n\')\n if re.search(r\'in error|failed indexing|Error indexing|will be disabled\', l)]\n names = [clean(l).strip() for l in m.group(1).split(\'\\n\') if l.strip()]\n boundary = len(txt[:m.end()].encode(\'utf-8\'))\n open(START, \'w\').write(str(boundary)); open(OFF, \'w\').write(str(boundary))\n time.sleep(1.5) # let triggers start listening\n for e in errs: print(e)\n print(\'host started, functions:\')\n for n in names: print(\' \' + re.sub(r\'http://localhost:\\d+\', \'http://localhost:\' + port, n))\n return\n if not m and re.search(r\'\\n\\[[^\\]]*\\] (Job host started|Host started)\', txt) and \'Worker process started and initialized\' in txt:\n time.sleep(1.5)\n boundary = len(txt.encode(\'utf-8\')); open(START, \'w\').write(str(boundary)); open(OFF, \'w\').write(str(boundary))\n for l in txt.split(\'\\n\'):\n if re.search(r\'in error|failed indexing|is disabled\', l): print(clean(l))\n print(\'host started, no enabled functions listed\'); return\n if re.search(r\'No job functions found|A host error has occurred\', txt):\n time.sleep(1); print(\'\\n\'.join(filtered(txt.split(\'\\n\'), all_=True)[-30:])); stop(quiet=True); sys.exit(\'host error\')\n time.sleep(0.3)\n print(\'\\n\'.join(filtered(open(LOG).read().split(\'\\n\'), all_=True)[-30:])); stop(quiet=True)\n sys.exit(\'host did not start in 120 s\')\n\ndef wait(pat, secs=\'60\', count=\'1\'):\n rx, t0, need = re.compile(pat), time.time(), int(count)\n while time.time() - t0 < float(secs):\n lines, _ = read_from(get(OFF))\n if sum(1 for l in lines if rx.search(clean(l))) >= need: return\n time.sleep(0.2)\n lines, _ = read_from(get(OFF))\n print(\'\\n\'.join(filtered(lines)[-20:])); sys.exit(\'timed out waiting for %r\' % pat)\n\ndef log(all_=False, details=False, since_start=False):\n off = get(START) if since_start else get(OFF)\n lines, new = read_from(off)\n if not since_start or new > get(OFF): open(OFF, \'w\').write(str(new))\n for c in filtered(lines, all_, details): print(c)\n\ndef stop(quiet=False):\n if not os.path.exists(PID): return\n pid = int(open(PID).read())\n for sig in (signal.SIGTERM, signal.SIGKILL):\n try: os.killpg(pid, sig)\n except Exception: break\n for _ in range(50):\n try: os.killpg(pid, 0); time.sleep(0.1)\n except Exception: break\n os.remove(PID)\n if not quiet: print(\'host stopped\')\n\nif __name__ == \'__main__\':\n a = sys.argv[1:] or [\'help\']\n if a[0] == \'start\': start(*a[1:])\n elif a[0] == \'wait\': wait(*a[1:])\n elif a[0] == \'log\': log(\'--all\' in a, \'--details\' in a, \'--since-start\' in a)\n elif a[0] == \'stop\': stop()\n else: print(__doc__)\n')
os.environ["FHOST"] = os.path.abspath("fhost.py") # commands below run it as: python "$FHOST" ...
write('orders_app/host.json', """\
{
"version": "2.0",
"extensionBundle": {
"id": "Microsoft.Azure.Functions.ExtensionBundle",
"version": "[4.*, 5.0.0)"
},
"extensions": {
"queues": {"batchSize": 1, "newBatchThreshold": 0,
"maxPollingInterval": "00:00:01"}
},
"logging": {
"logLevel": {"default": "Warning", "Function": "Information",
"Azure.Core": "Error",
"Host.Function.Console": "Information"}
}
}
""")
write('orders_app/requirements.txt', """\
azure-functions
""")
write('orders_app/local.settings.json', """\
{
"IsEncrypted": false,
"Values": {
"FUNCTIONS_WORKER_RUNTIME": "python",
"AzureWebJobsStorage": "UseDevelopmentStorage=true",
"SOURCE_SYSTEM": "oracle-fusion",
"DB_PWD": "@Microsoft.KeyVault(VaultName=kv-dev;SecretName=dbpw)",
"PARTNER_URL": "https://partner.example.com/api"
}
}
""")
# Step 1: App settings are environment variables; a Key Vault reference stays literal locally
write('orders_app/function_app.py', """\
import logging, os
import azure.functions as func
app = func.FunctionApp()
KV_REF = "@Microsoft.KeyVault("
NAMES = ["SOURCE_SYSTEM", "DB_PWD", "PARTNER_URL", "SFTP_PASSWORD"]
@app.timer_trigger(schedule="0 0 6 * * *", arg_name="t",
run_on_startup=False)
def config_check(t: func.TimerRequest) -> None:
for name in NAMES:
value = os.getenv(name) # app settings = env vars
if value is None:
state = "missing"
elif value.startswith(KV_REF):
state = "Key Vault reference, not resolved"
else:
state = "set (value never logged)"
logging.info("%s: %s", name, state)
""")
write('orders_app/local.settings.json', """\
{
"IsEncrypted": false,
"Values": {
"FUNCTIONS_WORKER_RUNTIME": "python",
"AzureWebJobsStorage": "UseDevelopmentStorage=true",
"SOURCE_SYSTEM": "oracle-fusion",
"DB_PWD": "@Microsoft.KeyVault(VaultName=kv-dev;SecretName=dbpw)",
"PARTNER_URL": "https://partner.example.com/api"
}
}
""")
write('fire.py', """\
import sys, requests
# run a timer now through the admin API (Azure: needs the master key)
URL = "http://localhost:7071/admin/functions/"
r = requests.post(URL + sys.argv[1], json={"input": ""})
print("run", sys.argv[1], "->", r.status_code)
""")
run('cd orders_app && python "$FHOST" start')
run('python fire.py config_check')
run('cd orders_app && python "$FHOST" wait "Executed \'Functions.config_check\'" 30')
run('cd orders_app && python "$FHOST" log && python "$FHOST" stop')
# Step 2: Auth levels in code: the local host does not enforce keys, but the keys exist
write('orders_app/function_app.py', """\
import azure.functions as func
app = func.FunctionApp(http_auth_level=func.AuthLevel.FUNCTION)
@app.route(route="orders", methods=["GET"]) # function or host key
def orders(req: func.HttpRequest) -> func.HttpResponse:
return func.HttpResponse("3 open orders")
@app.route(route="purge", auth_level=func.AuthLevel.ADMIN)
def purge(req: func.HttpRequest) -> func.HttpResponse: # master key
return func.HttpResponse("purged")
@app.route(route="health", auth_level=func.AuthLevel.ANONYMOUS)
def health(req: func.HttpRequest) -> func.HttpResponse:
return func.HttpResponse("ok")
""")
write('call.py', """\
import requests
BASE = "http://localhost:7071"
for route in ["orders", "purge", "health"]:
r = requests.get(f"{BASE}/api/{route}") # no key at all
print(f"GET /api/{route} without a key -> {r.status_code}")
fn = requests.get(f"{BASE}/admin/functions/orders/keys").json()
print("orders function keys:", [k["name"] for k in fn["keys"]])
host = requests.get(f"{BASE}/admin/host/keys").json()
print("host keys:", [k["name"] for k in host["keys"]]) # names only
""")
run('cd orders_app && python "$FHOST" start')
run('python call.py')
run('cd orders_app && python "$FHOST" stop')
# Step 3: Dev settings: the queue name and database come from local.settings.json
write('env_app/function_app.py', """\
import logging, os
import azure.functions as func
app = func.FunctionApp()
@app.queue_trigger(arg_name="msg", queue_name="%ORDERS_QUEUE%",
connection="AzureWebJobsStorage")
def load_order(msg: func.QueueMessage) -> None:
body = msg.get_body().decode()
env, db = os.environ["APP_ENV"], os.environ["SQL_DATABASE"]
logging.info("%s: load %s into %s", env, body, db)
""")
write('env_app/local.settings.json', """\
{
"IsEncrypted": false,
"Values": {
"FUNCTIONS_WORKER_RUNTIME": "python",
"AzureWebJobsStorage": "UseDevelopmentStorage=true",
"APP_ENV": "dev",
"ORDERS_QUEUE": "orders-dev",
"SQL_DATABASE": "dw_dev"
}
}
""")
write('env_app/host.json', """\
{
"version": "2.0",
"extensionBundle": {
"id": "Microsoft.Azure.Functions.ExtensionBundle",
"version": "[4.*, 5.0.0)"
},
"extensions": {
"queues": {"batchSize": 1, "newBatchThreshold": 0,
"maxPollingInterval": "00:00:01"}
},
"logging": {
"logLevel": {"default": "Warning", "Function": "Information",
"Azure.Core": "Error",
"Host.Function.Console": "Information"}
}
}
""")
write('env_app/requirements.txt', """\
azure-functions
""")
write('send.py', """\
import os, sys
from azure.storage.queue import QueueClient, TextBase64EncodePolicy
q = QueueClient.from_connection_string(
os.environ["AZURE_STORAGE_CONNECTION_STRING"], sys.argv[1],
message_encode_policy=TextBase64EncodePolicy())
try:
q.create_queue()
except Exception:
pass # already there
for body in sys.argv[2:]:
q.send_message(body)
print("sent to", sys.argv[1] + ":", ", ".join(sys.argv[2:]))
""")
run('cd env_app && python "$FHOST" start')
run('python send.py orders-dev o-1')
run('cd env_app && python "$FHOST" wait \'Executed\' 30 && python "$FHOST" log && python "$FHOST" stop')
# Step 4: Same code, test values: shell variables override local.settings.json
run('cd env_app && APP_ENV=test ORDERS_QUEUE=orders-test SQL_DATABASE=dw_test fhost start')
run('python send.py orders-test o-2')
run('cd env_app && python "$FHOST" wait \'Executed\' 30 && python "$FHOST" log && python "$FHOST" stop')
# Step 5: The .gitignore from func init keeps local.settings.json out; git add -f does not
run('func init orders_api --worker-runtime python --model V2')
run('cd orders_api && git init -q && git add -A && git ls-files')
run('cd orders_api && git check-ignore -v local.settings.json')
run('cd orders_api && git add -f local.settings.json && git status -s local.settings.json')
run('cd orders_api && git rm -q --cached local.settings.json && git ls-files | wc -l')
Part 9: Monitoring, Testing and Deployment
The sample data on this page is invented for teaching. Every terminal output is the real output of the command above it, run with Azure Functions Core Tools 4.16.0 (the real Functions host, Python 3.13 worker) and Azurite 3.37.0 for storage. Features the local host cannot show truthfully are shown as code. Part of the Real Time Azure Functions Interview Questions and Answers series on TechBrothersIT. Features, limits and their status (GA or preview) are current as of November 2026; check the official Azure Functions documentation (learn.microsoft.com/azure/azure-functions) for anything you quote in an interview.
No comments:
Post a Comment
Note: Only a member of this blog may post a comment.