Azure Functions Interview Questions and Answers Part 10 - Real-World Manager Scenarios

Azure Functions Interview Q&A · Part 10 of 10

In this post we are going to cover real-world manager scenarios for 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 the final-round question: the hiring manager asks you to design an event-driven ingestion layer with Functions.

These are the questions that do not fit one feature. A queue keeps growing, a timer loads data twice, the bill doubles, or a function works on your laptop and fails in Azure. Interviewers ask them because they show how you investigate under pressure: what you check first, what you check next, and what you change so it does not happen again.

VIDEO COMING SOON

Part 10 of the Real Time Azure Functions Interview Questions and Answers series on TechBrothersIT.

On this page

1. The short answer
2. Let's set up the project
3. Question 1: A queue keeps growing and the function seems stuck. How do you investigate?
4. Question 2: A timer function ran twice at the same time and loaded data twice. Why, and how do you prevent it?
5. Question 3: Someone asks whether to use Functions, Data Factory or Databricks for a new ingestion job. How do you decide?
6. Question 4: The function bill doubled this month. Where do you look?
7. Question 5: A function works locally but fails in Azure with missing settings and import errors. How do you troubleshoot it?
8. Question 6: A team wants to migrate from the Consumption plan to Flex Consumption. How do you plan it?
9. Scenario 7: In the final round, the hiring manager asks you to design an event-driven ingestion layer with Functions. What is your design?
10. The full script

The short answer

Say this first

Investigate from enabled to failing to scale. Stop double timer runs with disabled staging timers and an idempotent claim. Keep Functions for small event-driven work, find bill growth in executions and baselines, compare settings and packages when Azure fails, migrate to Flex side by side, and design around buffers.

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 queues, blobs and tables. Each scenario that runs code has its own small Python function app, plus client scripts that send messages, call the app and check queues. What only Azure can show, such as billing, migration and settings in the portal, is shown as code.

Question 1: A queue keeps growing and the function seems stuck. How do you investigate?

I go from cheap checks to deep ones: is it enabled, is it failing, or is it healthy but slow or capped? The queue and poison queue lengths answer most of that.

• Enabled? Disabled flag, function in error, host startup errors
• Failing? Exceptions, rising dequeue count, poison queue length
• Slow or capped? Maximum instance count, downstream throttling, batch size

Let's reproduce it. We start orders_app, queue five orders with send_orders.py, and run check.py, which reads the queue lengths and the function status from the admin API:

orders_app/function_app.py

import logging
import azure.functions as func

app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS)


@app.route(route="health")
def health(req: func.HttpRequest) -> func.HttpResponse:
    return func.HttpResponse("ok")


@app.queue_trigger(arg_name="msg", queue_name="orders",
                   connection="AzureWebJobsStorage")
def load_orders(msg: func.QueueMessage) -> None:
    order = msg.get_json()
    logging.info("loaded order %s", order["id"])

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",
      "visibilityTimeout": "00:00:01"
    }
  },
  "logging": {
    "logLevel": {
      "default": "Warning",
      "Function": "Information",
      "Azure.Core": "Error",
      "Host.Function.Console": "Information"
    }
  }
}

orders_app/local.settings.json

{
  "IsEncrypted": false,
  "Values": {
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "AzureWebJobs.load_orders.Disabled": "true"
  }
}

orders_app/requirements.txt

azure-functions

send_orders.py

import json, os, sys
from azure.storage.queue import QueueClient, TextBase64EncodePolicy

queue, count = sys.argv[1], int(sys.argv[2])
q = QueueClient.from_connection_string(
    os.environ["AZURE_STORAGE_CONNECTION_STRING"], queue,
    message_encode_policy=TextBase64EncodePolicy())   # base64
try:
    q.create_queue()
except Exception:
    pass                                 # already there
for i in range(1, count + 1):
    q.send_message(json.dumps({"id": f"o-{i}", "amount": 10 * i}))
print("queued", count, "orders on", queue)

check.py

import os, sys
import requests
from azure.core.exceptions import ResourceNotFoundError
from azure.storage.queue import QueueServiceClient

svc = QueueServiceClient.from_connection_string(
    os.environ["AZURE_STORAGE_CONNECTION_STRING"])
queue = sys.argv[1]
for name in [queue, queue + "-poison"]:
    try:
        props = svc.get_queue_client(name).get_queue_properties()
        waiting = props.approximate_message_count
    except ResourceNotFoundError:
        waiting = 0                      # no poison queue yet
    print(f"{name:<15} waiting: {waiting}")
if len(sys.argv) > 2:                    # admin API: master key in Azure
    fn = requests.get("http://localhost:7071/admin/functions/"
                      + sys.argv[2]).json()
    print(f"{sys.argv[2]:<15} disabled: {fn['isDisabled']}")

Terminal

$ cd orders_app && fhost start
host started, functions:
  health:  http://localhost:7071/api/health
  Function load_orders is disabled.
$ python send_orders.py orders 5
queued 5 orders on orders
$ python check.py orders load_orders
orders          waiting: 5
orders-poison   waiting: 0
load_orders     disabled: True

The check shows orders waiting: 5, orders-poison waiting: 0, and load_orders disabled: True. The cause is an AzureWebJobs.load_orders.Disabled setting left over from an incident. We remove it and restart the host:

orders_app/local.settings.json

{
  "IsEncrypted": false,
  "Values": {
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "AzureWebJobsStorage": "UseDevelopmentStorage=true"
  }
}

Terminal

$ cd orders_app && fhost stop && fhost start
host stopped
host started, functions:
  health:  http://localhost:7071/api/health
  load_orders: queueTrigger
$ cd orders_app && fhost wait "Executed 'Functions.load_orders'" 60 5 && fhost log | grep loaded | sort
loaded order o-1
loaded order o-2
loaded order o-3
loaded order o-4
loaded order o-5
$ sleep 2 && python check.py orders load_orders
orders          waiting: 0
orders-poison   waiting: 0
load_orders     disabled: False
$ cd orders_app && fhost stop
host stopped

The host loads o-1 to o-5, and the check now shows waiting: 0 and disabled: False.

If it is enabled, read the host log and Application Insights:

• Host log, Application Insights: failures and exceptions per function
• Poison queue: messages that never succeed
• No runs at all: startup or indexing errors

Azure

The admin API needs no key locally, but in Azure it needs the master key.

Question 2: A timer function ran twice at the same time and loaded data twice. Why, and how do you prevent it?

In one app, a timer is a singleton: instances share a host ID and a storage lock, so scale-out does not double it. Doubles come from somewhere else:

• Second copy: a staging slot or environment on the same storage, with its own host ID
• run_on_startup=True: fires on every restart and scale-out
• Retries: a retry policy re-runs the load

Let's prove the first one. prod and a staging copy share one storage account, staging has its own host ID, and the demo schedule fires every five seconds. We start both hosts and list the timer locks with locks.py:

prod/function_app.py

import logging
import azure.functions as func

app = func.FunctionApp()


# prod schedule: "0 0 2 * * *" (02:00 UTC); every 5 s for the demo
@app.timer_trigger(schedule="*/5 * * * * *", arg_name="timer",
                   run_on_startup=False)
def nightly_load(timer: func.TimerRequest) -> None:
    logging.info("nightly_load: loading yesterday's orders")

prod/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",
      "visibilityTimeout": "00:00:01"
    }
  },
  "logging": {
    "logLevel": {
      "default": "Warning",
      "Function": "Information",
      "Azure.Core": "Error",
      "Host.Function.Console": "Information"
    }
  }
}

prod/local.settings.json

{
  "IsEncrypted": false,
  "Values": {
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "CONTROL_STORAGE": "UseDevelopmentStorage=true"
  }
}

prod/requirements.txt

azure-functions

staging/function_app.py

import logging
import azure.functions as func

app = func.FunctionApp()


# prod schedule: "0 0 2 * * *" (02:00 UTC); every 5 s for the demo
@app.timer_trigger(schedule="*/5 * * * * *", arg_name="timer",
                   run_on_startup=False)
def nightly_load(timer: func.TimerRequest) -> None:
    logging.info("nightly_load: loading yesterday's orders")

staging/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",
      "visibilityTimeout": "00:00:01"
    }
  },
  "logging": {
    "logLevel": {
      "default": "Warning",
      "Function": "Information",
      "Azure.Core": "Error",
      "Host.Function.Console": "Information"
    }
  }
}

staging/local.settings.json

{
  "IsEncrypted": false,
  "Values": {
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "CONTROL_STORAGE": "UseDevelopmentStorage=true"
  }
}

staging/requirements.txt

azure-functions

locks.py

import os
from azure.storage.blob import ContainerClient

hosts = ContainerClient.from_connection_string(
    os.environ["AZURE_STORAGE_CONNECTION_STRING"], "azure-webjobs-hosts")
for b in hosts.list_blobs(name_starts_with="locks/"):
    if b.name.endswith("nightly_load.Listener"):    # the timer's lock
        print(b.name, "| lease:", b.lease.state)

Terminal

$ cd prod && fhost start
host started, functions:
  nightly_load: timerTrigger
$ cd staging && AzureFunctionsWebHost__hostid=tbit-fn-staging fhost start 7072
host started, functions:
  nightly_load: timerTrigger
$ cd prod && fhost wait 'nightly_load: loading' 30 2 && echo 'prod ran the load'
prod ran the load
$ cd staging && fhost wait 'nightly_load: loading' 30 2 && echo 'staging ran the load'
staging ran the load
$ python locks.py
locks/tbit-fn-demo/Host.Functions.nightly_load.Listener | lease: leased
locks/tbit-fn-staging/Host.Functions.nightly_load.Listener | lease: leased
$ cd staging && fhost stop
host stopped
$ cd prod && fhost stop
host stopped

Both print that they ran the load, and there are two leased listener locks, one under tbit-fn-demo and one under tbit-fn-staging. Now both hosts use the same host ID:

Terminal

$ cd prod && fhost start
host started, functions:
  nightly_load: timerTrigger
$ cd prod && fhost wait 'nightly_load: loading' 30 1
$ cd staging && fhost start 7072
host started, functions:
  nightly_load: timerTrigger
$ cd prod && fhost wait 'nightly_load: loading' 30 3 && echo 'prod ran the load'
prod ran the load
$ cd staging && echo "staging runs: $(fhost log | grep -c nightly_load)"
staging runs: 0
$ cd staging && fhost stop
host stopped
$ cd prod && fhost stop
host stopped

Prod keeps running, and staging shows staging runs: 0. That is the lock. But never share host IDs between real apps, or one app's timer never runs.

The fixes: a sticky AzureWebJobs.nightly_load.Disabled setting in staging, no run_on_startup in production, and, from my projects, an idempotent load that claims the date in a loadcontrol table first. Two host IDs again:

prod/function_app.py

import logging
import os
from datetime import datetime, timezone
import azure.functions as func
from azure.core.exceptions import ResourceExistsError
from azure.data.tables import TableServiceClient

logging.getLogger("azure").setLevel(logging.WARNING)  # no SDK chatter
app = func.FunctionApp()
tables = TableServiceClient.from_connection_string(
    os.environ["CONTROL_STORAGE"])
control = tables.create_table_if_not_exists("loadcontrol")


# prod schedule: "0 0 2 * * *" (02:00 UTC); every 5 s for the demo
@app.timer_trigger(schedule="*/5 * * * * *", arg_name="timer",
                   run_on_startup=False)
def nightly_load(timer: func.TimerRequest) -> None:
    day = datetime.now(timezone.utc).date().isoformat()
    try:   # claim the date first: only one insert can win
        control.create_entity({"PartitionKey": "nightly_load",
                               "RowKey": day})
    except ResourceExistsError:
        logging.info("nightly_load: skipped, date already claimed")
        return
    logging.info("nightly_load: claimed the date, loading orders")

staging/function_app.py

import logging
import os
from datetime import datetime, timezone
import azure.functions as func
from azure.core.exceptions import ResourceExistsError
from azure.data.tables import TableServiceClient

logging.getLogger("azure").setLevel(logging.WARNING)  # no SDK chatter
app = func.FunctionApp()
tables = TableServiceClient.from_connection_string(
    os.environ["CONTROL_STORAGE"])
control = tables.create_table_if_not_exists("loadcontrol")


# prod schedule: "0 0 2 * * *" (02:00 UTC); every 5 s for the demo
@app.timer_trigger(schedule="*/5 * * * * *", arg_name="timer",
                   run_on_startup=False)
def nightly_load(timer: func.TimerRequest) -> None:
    day = datetime.now(timezone.utc).date().isoformat()
    try:   # claim the date first: only one insert can win
        control.create_entity({"PartitionKey": "nightly_load",
                               "RowKey": day})
    except ResourceExistsError:
        logging.info("nightly_load: skipped, date already claimed")
        return
    logging.info("nightly_load: claimed the date, loading orders")

loads.py

import os
from azure.data.tables import TableClient

control = TableClient.from_connection_string(
    os.environ["AZURE_STORAGE_CONNECTION_STRING"], "loadcontrol")
rows = list(control.list_entities())
print("loadcontrol rows:", len(rows),
      "| jobs:", sorted({r["PartitionKey"] for r in rows}))

Terminal

$ cd prod && fhost start
host started, functions:
  nightly_load: timerTrigger
$ cd staging && AzureFunctionsWebHost__hostid=tbit-fn-staging fhost start 7072
host started, functions:
  nightly_load: timerTrigger
$ cd prod && fhost wait 'nightly_load: ' 30 3
$ cd staging && fhost wait 'nightly_load: ' 30 3
$ (cd prod && fhost log; cd ../staging && fhost log) | grep -o 'nightly_load: .*' | sort -u
nightly_load: claimed the date, loading orders
nightly_load: skipped, date already claimed
$ python loads.py
loadcontrol rows: 1 | jobs: ['nightly_load']
$ cd staging && fhost stop
host stopped
$ cd prod && fhost stop
host stopped

One host logs claimed the date, loading orders, the other skipped, date already claimed, and loads.py shows loadcontrol rows: 1.

Catch

If the load fails, delete the claim row, or it blocks the retry. That is also how you rerun a date.

Question 3: Someone asks whether to use Functions, Data Factory or Databricks for a new ingestion job. How do you decide?

This is my rule of thumb from real projects, not a Microsoft rule: Functions for event-driven work that is small and short, in code. A file lands, a message arrives, an API is called, and it is done in seconds or minutes.

Let's put the three side by side:

	Azure Functions	Data Factory	Databricks
Best for	event-driven, small, short work in code	managed copies, scheduled orchestration	Spark-scale transforms
Typical job	a file lands, a message arrives, an API call	source tables into the lake on a schedule	big joins, large files, Delta merges
Trade-off	you own the code and the retries	little code, less custom logic	a cluster to run and pay for

Data Factory is for managed copies and scheduled orchestration with many connectors: tables from a source database into the lake, with little code per source. Databricks is for Spark-scale transforms: big joins, large files, Delta merges, anything that needs a cluster's memory.

The Functions limits push you toward that split:

• An HTTP response must come back within 230 seconds on every plan
• The legacy Consumption plan stops at 10 minutes
• A Flex instance has at most 4 GB of memory

They also work together. An event or a schedule starts the flow, from Event Grid or a Data Factory trigger. A function validates and routes. Databricks does the heavy transform.

Pitfall

The mistake I see is a big pandas join in a function, until it runs out of memory. No big joins in a function.

Question 4: The function bill doubled this month. Where do you look?

First, what is metered. On Flex Consumption on demand, you pay for executions and for execution time times instance memory, after a monthly free grant. Always ready instances add a baseline that is billed even when idle.

• More executions? Retries, failing dependencies, tiny batches
• Longer executions? Slow downstream calls, bigger payloads
• More baseline? Always ready, Premium instances

A classic cause is a failing dependency. Let's send three orders to billing_app, where enrich_order calls a partner API that is down, and count what happens:

billing_app/function_app.py

import logging
import azure.functions as func

app = func.FunctionApp()


def call_partner_api(order: dict) -> dict:
    raise ConnectionError("partner API is down")     # the outage


@app.queue_trigger(arg_name="msg", queue_name="enrich",
                   connection="AzureWebJobsStorage")
def enrich_order(msg: func.QueueMessage) -> None:
    order = msg.get_json()
    logging.info("order %s attempt %d", order["id"], msg.dequeue_count)
    call_partner_api(order)

billing_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",
      "visibilityTimeout": "00:00:01"
    }
  },
  "logging": {
    "logLevel": {
      "default": "Warning",
      "Function": "Information",
      "Azure.Core": "Error",
      "Host.Function.Console": "Information"
    }
  }
}

billing_app/local.settings.json

{
  "IsEncrypted": false,
  "Values": {
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "AzureWebJobsStorage": "UseDevelopmentStorage=true"
  }
}

billing_app/requirements.txt

azure-functions

Terminal

$ cd billing_app && fhost start
host started, functions:
  enrich_order: queueTrigger
$ python send_orders.py enrich 3
queued 3 orders on enrich
$ cd billing_app && fhost wait 'Moving message to queue' 120 3 && fhost log > run.log
$ cd billing_app && grep -c "Executed 'Functions.enrich_order' (Failed)" run.log
15
$ cd billing_app && grep -o 'attempt [0-9]' run.log | sort | uniq -c
      3 attempt 1
      3 attempt 2
      3 attempt 3
      3 attempt 4
      3 attempt 5
$ cd billing_app && grep 'Moving message' run.log | sort | uniq -c
      3 Message has reached MaxDequeueCount of 5. Moving message to queue 'enrich-poison'.
$ python check.py enrich
enrich          waiting: 0
enrich-poison   waiting: 3
$ cd billing_app && fhost stop
host stopped

We get 15 failed executions: each of the three orders shows attempts 1 to 5, then MaxDequeueCount of 5 moves it to enrich-poison, which ends with 3 waiting. Three orders, fifteen billed executions. An unlimited retry policy, or a queue that keeps refilling, turns one outage into a big bill, and on Flex every execution is billed for at least one second.

Then the quieter drivers:

Driver	What to check
Executions and GB-seconds	retries, poison counts, tiny batches; at least 1 s billed per execution (Flex)
Always ready / Premium	baseline instances, billed even when idle
Application Insights	sampling turned off, verbose log levels
Storage transactions	Durable Functions, Event Hubs checkpoints, blob polling
Scale-out	no maximum instance count, run_on_startup

My routine:

• Cost Management by meter, then executions and duration per function, month over month
• Budget alert on the resource group
• Fix the cause: a circuit breaker for the failing API, then tune

Real project

Look at executions first. A doubled bill usually means more runs, longer runs or a bigger baseline, and retries are the usual suspect.

Question 5: A function works locally but fails in Azure with missing settings and import errors. How do you troubleshoot it?

Usually it is the environment, not the code. local.settings.json is not published, and only the packages in requirements.txt exist in Azure.

Let's reproduce both. The ingest function writes to the container named in the RAW_CONTAINER setting. We post an order with post_order.py and list the container with list_raw.py:

lake_app/function_app.py

import json
import azure.functions as func

app = func.FunctionApp(http_auth_level=func.AuthLevel.FUNCTION)


@app.route(route="ingest", methods=["POST"])
@app.blob_output(arg_name="out",
                 path="%RAW_CONTAINER%/orders/{id}.json",
                 connection="AzureWebJobsStorage")
def ingest(req: func.HttpRequest,
           out: func.Out[str]) -> func.HttpResponse:
    order = req.get_json()
    out.set(json.dumps(order))
    return func.HttpResponse(f"saved {order['id']}", status_code=201)

lake_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",
      "visibilityTimeout": "00:00:01"
    }
  },
  "logging": {
    "logLevel": {
      "default": "Warning",
      "Function": "Information",
      "Azure.Core": "Error",
      "Host.Function.Console": "Information"
    }
  }
}

lake_app/local.settings.json

{
  "IsEncrypted": false,
  "Values": {
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "RAW_CONTAINER": "raw"
  }
}

lake_app/requirements.txt

azure-functions

post_order.py

import sys
import requests

order = {"id": sys.argv[1], "amount": float(sys.argv[2])}
r = requests.post("http://localhost:7071/api/ingest", json=order)
print(r.status_code, r.text or "(empty body)")

list_raw.py

import os
from azure.storage.blob import ContainerClient

raw = ContainerClient.from_connection_string(
    os.environ["AZURE_STORAGE_CONNECTION_STRING"], "raw")
print("raw:", sorted(raw.list_blob_names()))

Terminal

$ cd lake_app && fhost start
host started, functions:
  ingest: [POST] http://localhost:7071/api/ingest
$ python post_order.py o-1001 25.5
201 saved o-1001
$ python list_raw.py
raw: ['orders/o-1001.json']
$ cd lake_app && fhost stop
host stopped

Locally we get 201 saved o-1001, and the file lands in raw as orders/o-1001.json. Now we start it with only the settings the Azure app has:

lake_app/local.settings.json

{
  "IsEncrypted": false,
  "Values": {
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "AzureWebJobsStorage": "UseDevelopmentStorage=true"
  }
}

Terminal

$ cd lake_app && fhost start
host started, functions:
  ingest: [POST] http://localhost:7071/api/ingest
$ python post_order.py o-1002 7
500 (empty body)
$ cd lake_app && fhost wait "Executed 'Functions.ingest'" 30 && fhost log
Executing 'Functions.ingest' (Reason='This function was programmatically called via the host APIs.')
Executed 'Functions.ingest' (Failed)
System.Private.CoreLib: Exception while executing function: Functions.ingest.
  Microsoft.Azure.WebJobs.Host: '%RAW_CONTAINER%' does not resolve to a value.
$ cd lake_app && fhost stop
host stopped

The call fails with 500 (empty body), and the log says '%RAW_CONTAINER%' does not resolve to a value. Compare the lists: the Azure setting names from the CLI, or fetch them with Core Tools:

compare_settings.sh (Azure, not run)

# which settings does the Azure app really have? names only
az functionapp config appsettings list -g rg-data \
  -n func-lake-prod --query "[].name" -o tsv
# or pull the Azure settings into local.settings.json
func azure functionapp fetch-app-settings func-lake-prod

Next, the import error: a teammate imported polars but did not add it to requirements.txt. As in Azure, it is not installed here:

lake_app/function_app.py

import azure.functions as func
import polars as pl              # new, but not in requirements.txt

app = func.FunctionApp(http_auth_level=func.AuthLevel.FUNCTION)


@app.route(route="ingest", methods=["POST"])
@app.blob_output(arg_name="out",
                 path="%RAW_CONTAINER%/orders/{id}.json",
                 connection="AzureWebJobsStorage")
def ingest(req: func.HttpRequest,
           out: func.Out[str]) -> func.HttpResponse:
    order = req.get_json()
    out.set(pl.DataFrame([order]).write_ndjson())
    return func.HttpResponse(f"saved {order['id']}", status_code=201)

lake_app/local.settings.json

{
  "IsEncrypted": false,
  "Values": {
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "RAW_CONTAINER": "raw"
  }
}

Terminal

$ cat lake_app/requirements.txt
azure-functions
$ cd lake_app && fhost start > start.log 2>&1 || echo "the host did not start"
the host did not start
$ grep -o "ModuleNotFoundError.*missing module" lake_app/start.log | uniq
ModuleNotFoundError: No module named 'polars'. Cannot find module. Please check the requirements.txt
  file for the missing module

The host did not start, with ModuleNotFoundError: No module named 'polars'. In Azure the same mistake looks like a clean deployment with no functions listed.

• AzureWebJobsStorage: not UseDevelopmentStorage=true
• Python version: available on the plan
• Remote build: Linux packages built in Azure
• Key Vault references: resolved, not literal text
• New role assignments: up to 10 minutes to apply

Tip

When a deployment succeeds but no functions are listed, check requirements.txt and the imports first.

Question 6: A team wants to migrate from the Consumption plan to Flex Consumption. How do you plan it?

Linux Consumption is retiring on 30 September 2028 and gets no new features, so Flex Consumption is the target. The documented path is side by side: create a new Flex app, redeploy the code, test, and move traffic.

The CLI's flex migration commands list the eligible apps and create a new Flex app with most of the settings. An in-place upgrade is in preview:

migrate.sh (Azure CLI 2.91, not run)

# 1. which Linux Consumption apps can move, and why not
az functionapp flex-migration list -o table

# 2. side by side (the documented path): a new Flex app
az functionapp flex-migration start \
  --source-name func-orders --source-resource-group rg-data \
  --name func-orders-flex --resource-group rg-data \
  --storage-account stordersflex --maximum-instance-count 100

# 3. redeploy the code, test, move traffic; keep the old
#    app with its triggers disabled as the rollback

# --in-place (same app) and flex-migration revert: preview

Then let's go through what changes on Flex:

On Flex Consumption	What you do
No deployment slots	rolling updates and a separate staging app
Blob trigger: Event Grid source only	source="EventGrid" plus an event subscription
No WEBSITE_TIME_ZONE	timer schedules in UTC
Settings that go away	FUNCTIONS_WORKER_PROCESS_COUNT, WEBSITE_RUN_FROM_PACKAGE, ...

Next, recreate what does not carry over:

• System-assigned identity: a new principal, so redo the role assignments
• Function keys: new keys, so update the callers

Then cut over carefully:

• Benchmark cold starts and latency, before and after
• Move traffic: clients, queues, DNS
• Keep the old app, triggers disabled, for a few weeks: rollback means switching back

Pitfall

Event Grid push cannot reach a private endpoint, so a locked-down app with a blob trigger needs another design, like a storage queue.

Scenario 7: In the final round, the hiring manager asks you to design an event-driven ingestion layer with Functions. What is your design?

My design puts a buffer on every path, thin functions in the middle, and the heavy work somewhere else.

Let's start with the way in. Partner files raise BlobCreated events, which Event Grid delivers into a storage queue. Streams come through Event Hubs, and business messages through Service Bus. Every path has a buffer, so a burst or an outage loses nothing.

Then the functions. They run on Flex Consumption and do the thin work: validate, add metadata, write raw files to the lake, and hand off. Each trigger scales on its own, a maximum instance count protects downstream systems, and multi-step flows run as Durable Functions.

Then reliability:

• At least once: idempotent writes keyed on file name and version, or on message ID
• Failures: poison and dead-letter queues; for Event Hubs, park bad events yourself; alerts and a replay path

Then security and operations:

• Identity: managed identity, identity-based connections, Key Vault references
• Network: VNet integration, private endpoints
• Operations: alerts on failures and missing runs, deploy from a pipeline

Finally, heavy transforms go to Databricks and bulk copies to Data Factory. Functions stay the event-driven front door.

Close with this

That is the split I would defend: buffers in front, idempotent functions in the middle, and Spark for the heavy lifting.

Key points to remember

• Growing queue: is it enabled, is it failing, or is it slow or capped?
• Timer twice: a second host ID on the same storage; fix with disabled staging timers and an idempotent claim
• Functions for small event-driven work; Data Factory for copies; Databricks for Spark-scale transforms
• Bill doubled: executions, retries, baselines, App Insights ingestion, storage transactions
• Local vs Azure: settings aren't published, requirements.txt decides the packages; migrate to Flex side by side

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

Full script - writes the files and runs every step with Azure Functions Core Tools and Azurite (pip install azure-functions azure-functions-durable azure-storage-blob azure-storage-queue azure-data-tables), run it with Python

# ========================================================================
# Azure Functions Interview Q&A #10 - Real-world manager scenarios
# 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_10_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" ...

# Step 1: Five orders wait on the queue and nothing runs: check the queue and the function status
write('orders_app/function_app.py', """\
import logging
import azure.functions as func

app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS)


@app.route(route="health")
def health(req: func.HttpRequest) -> func.HttpResponse:
    return func.HttpResponse("ok")


@app.queue_trigger(arg_name="msg", queue_name="orders",
                   connection="AzureWebJobsStorage")
def load_orders(msg: func.QueueMessage) -> None:
    order = msg.get_json()
    logging.info("loaded order %s", order["id"])
""")
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",
      "visibilityTimeout": "00:00:01"
    }
  },
  "logging": {
    "logLevel": {
      "default": "Warning",
      "Function": "Information",
      "Azure.Core": "Error",
      "Host.Function.Console": "Information"
    }
  }
}
""")
write('orders_app/local.settings.json', """\
{
  "IsEncrypted": false,
  "Values": {
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "AzureWebJobs.load_orders.Disabled": "true"
  }
}
""")
write('orders_app/requirements.txt', """\
azure-functions
""")
write('send_orders.py', """\
import json, os, sys
from azure.storage.queue import QueueClient, TextBase64EncodePolicy

queue, count = sys.argv[1], int(sys.argv[2])
q = QueueClient.from_connection_string(
    os.environ["AZURE_STORAGE_CONNECTION_STRING"], queue,
    message_encode_policy=TextBase64EncodePolicy())   # base64
try:
    q.create_queue()
except Exception:
    pass                                 # already there
for i in range(1, count + 1):
    q.send_message(json.dumps({"id": f"o-{i}", "amount": 10 * i}))
print("queued", count, "orders on", queue)
""")
write('check.py', """\
import os, sys
import requests
from azure.core.exceptions import ResourceNotFoundError
from azure.storage.queue import QueueServiceClient

svc = QueueServiceClient.from_connection_string(
    os.environ["AZURE_STORAGE_CONNECTION_STRING"])
queue = sys.argv[1]
for name in [queue, queue + "-poison"]:
    try:
        props = svc.get_queue_client(name).get_queue_properties()
        waiting = props.approximate_message_count
    except ResourceNotFoundError:
        waiting = 0                      # no poison queue yet
    print(f"{name:<15} waiting: {waiting}")
if len(sys.argv) > 2:                    # admin API: master key in Azure
    fn = requests.get("http://localhost:7071/admin/functions/"
                      + sys.argv[2]).json()
    print(f"{sys.argv[2]:<15} disabled: {fn['isDisabled']}")
""")
run('cd orders_app && python "$FHOST" start')
run('python send_orders.py orders 5')
run('python check.py orders load_orders')

# Step 2: Remove the leftover Disabled setting, restart: the host drains the queue
write('orders_app/local.settings.json', """\
{
  "IsEncrypted": false,
  "Values": {
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "AzureWebJobsStorage": "UseDevelopmentStorage=true"
  }
}
""")
run('cd orders_app && python "$FHOST" stop && python "$FHOST" start')
run('cd orders_app && python "$FHOST" wait "Executed \'Functions.load_orders\'" 60 5 && python "$FHOST" log | grep loaded | sort')
run('sleep 2 && python check.py orders load_orders')
run('cd orders_app && python "$FHOST" stop')

# Step 3: prod and a staging copy with its own host ID on one storage account: both run the timer
write('prod/function_app.py', """\
import logging
import azure.functions as func

app = func.FunctionApp()


# prod schedule: "0 0 2 * * *" (02:00 UTC); every 5 s for the demo
@app.timer_trigger(schedule="*/5 * * * * *", arg_name="timer",
                   run_on_startup=False)
def nightly_load(timer: func.TimerRequest) -> None:
    logging.info("nightly_load: loading yesterday's orders")
""")
write('prod/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",
      "visibilityTimeout": "00:00:01"
    }
  },
  "logging": {
    "logLevel": {
      "default": "Warning",
      "Function": "Information",
      "Azure.Core": "Error",
      "Host.Function.Console": "Information"
    }
  }
}
""")
write('prod/local.settings.json', """\
{
  "IsEncrypted": false,
  "Values": {
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "CONTROL_STORAGE": "UseDevelopmentStorage=true"
  }
}
""")
write('prod/requirements.txt', """\
azure-functions
""")
write('staging/function_app.py', """\
import logging
import azure.functions as func

app = func.FunctionApp()


# prod schedule: "0 0 2 * * *" (02:00 UTC); every 5 s for the demo
@app.timer_trigger(schedule="*/5 * * * * *", arg_name="timer",
                   run_on_startup=False)
def nightly_load(timer: func.TimerRequest) -> None:
    logging.info("nightly_load: loading yesterday's orders")
""")
write('staging/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",
      "visibilityTimeout": "00:00:01"
    }
  },
  "logging": {
    "logLevel": {
      "default": "Warning",
      "Function": "Information",
      "Azure.Core": "Error",
      "Host.Function.Console": "Information"
    }
  }
}
""")
write('staging/local.settings.json', """\
{
  "IsEncrypted": false,
  "Values": {
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "CONTROL_STORAGE": "UseDevelopmentStorage=true"
  }
}
""")
write('staging/requirements.txt', """\
azure-functions
""")
write('locks.py', """\
import os
from azure.storage.blob import ContainerClient

hosts = ContainerClient.from_connection_string(
    os.environ["AZURE_STORAGE_CONNECTION_STRING"], "azure-webjobs-hosts")
for b in hosts.list_blobs(name_starts_with="locks/"):
    if b.name.endswith("nightly_load.Listener"):    # the timer's lock
        print(b.name, "| lease:", b.lease.state)
""")
run('cd prod && python "$FHOST" start')
run('cd staging && AzureFunctionsWebHost__hostid=tbit-fn-staging fhost start 7072')
run('cd prod && python "$FHOST" wait \'nightly_load: loading\' 30 2 && echo \'prod ran the load\'')
run('cd staging && python "$FHOST" wait \'nightly_load: loading\' 30 2 && echo \'staging ran the load\'')
run('python locks.py')
run('cd staging && python "$FHOST" stop')
run('cd prod && python "$FHOST" stop')

# Step 4: The same host ID for both: the timer lock lets only one host fire
run('cd prod && python "$FHOST" start')
run('cd prod && python "$FHOST" wait \'nightly_load: loading\' 30 1')
run('cd staging && python "$FHOST" start 7072')
run('cd prod && python "$FHOST" wait \'nightly_load: loading\' 30 3 && echo \'prod ran the load\'')
run('cd staging && echo "staging runs: $(fhost log | grep -c nightly_load)"')
run('cd staging && python "$FHOST" stop')
run('cd prod && python "$FHOST" stop')

# Step 5: Idempotent load: claim the date in a loadcontrol table first, two host IDs again
write('prod/function_app.py', """\
import logging
import os
from datetime import datetime, timezone
import azure.functions as func
from azure.core.exceptions import ResourceExistsError
from azure.data.tables import TableServiceClient

logging.getLogger("azure").setLevel(logging.WARNING)  # no SDK chatter
app = func.FunctionApp()
tables = TableServiceClient.from_connection_string(
    os.environ["CONTROL_STORAGE"])
control = tables.create_table_if_not_exists("loadcontrol")


# prod schedule: "0 0 2 * * *" (02:00 UTC); every 5 s for the demo
@app.timer_trigger(schedule="*/5 * * * * *", arg_name="timer",
                   run_on_startup=False)
def nightly_load(timer: func.TimerRequest) -> None:
    day = datetime.now(timezone.utc).date().isoformat()
    try:   # claim the date first: only one insert can win
        control.create_entity({"PartitionKey": "nightly_load",
                               "RowKey": day})
    except ResourceExistsError:
        logging.info("nightly_load: skipped, date already claimed")
        return
    logging.info("nightly_load: claimed the date, loading orders")
""")
write('staging/function_app.py', """\
import logging
import os
from datetime import datetime, timezone
import azure.functions as func
from azure.core.exceptions import ResourceExistsError
from azure.data.tables import TableServiceClient

logging.getLogger("azure").setLevel(logging.WARNING)  # no SDK chatter
app = func.FunctionApp()
tables = TableServiceClient.from_connection_string(
    os.environ["CONTROL_STORAGE"])
control = tables.create_table_if_not_exists("loadcontrol")


# prod schedule: "0 0 2 * * *" (02:00 UTC); every 5 s for the demo
@app.timer_trigger(schedule="*/5 * * * * *", arg_name="timer",
                   run_on_startup=False)
def nightly_load(timer: func.TimerRequest) -> None:
    day = datetime.now(timezone.utc).date().isoformat()
    try:   # claim the date first: only one insert can win
        control.create_entity({"PartitionKey": "nightly_load",
                               "RowKey": day})
    except ResourceExistsError:
        logging.info("nightly_load: skipped, date already claimed")
        return
    logging.info("nightly_load: claimed the date, loading orders")
""")
write('loads.py', """\
import os
from azure.data.tables import TableClient

control = TableClient.from_connection_string(
    os.environ["AZURE_STORAGE_CONNECTION_STRING"], "loadcontrol")
rows = list(control.list_entities())
print("loadcontrol rows:", len(rows),
      "| jobs:", sorted({r["PartitionKey"] for r in rows}))
""")
run('cd prod && python "$FHOST" start')
run('cd staging && AzureFunctionsWebHost__hostid=tbit-fn-staging fhost start 7072')
run('cd prod && python "$FHOST" wait \'nightly_load: \' 30 3')
run('cd staging && python "$FHOST" wait \'nightly_load: \' 30 3')
run('(cd prod && python "$FHOST" log; cd ../staging && python "$FHOST" log) | grep -o \'nightly_load: .*\' | sort -u')
run('python loads.py')
run('cd staging && python "$FHOST" stop')
run('cd prod && python "$FHOST" stop')

# Step 6: A partner API is down: every message runs five times, then goes to the poison queue
write('billing_app/function_app.py', """\
import logging
import azure.functions as func

app = func.FunctionApp()


def call_partner_api(order: dict) -> dict:
    raise ConnectionError("partner API is down")     # the outage


@app.queue_trigger(arg_name="msg", queue_name="enrich",
                   connection="AzureWebJobsStorage")
def enrich_order(msg: func.QueueMessage) -> None:
    order = msg.get_json()
    logging.info("order %s attempt %d", order["id"], msg.dequeue_count)
    call_partner_api(order)
""")
write('billing_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",
      "visibilityTimeout": "00:00:01"
    }
  },
  "logging": {
    "logLevel": {
      "default": "Warning",
      "Function": "Information",
      "Azure.Core": "Error",
      "Host.Function.Console": "Information"
    }
  }
}
""")
write('billing_app/local.settings.json', """\
{
  "IsEncrypted": false,
  "Values": {
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "AzureWebJobsStorage": "UseDevelopmentStorage=true"
  }
}
""")
write('billing_app/requirements.txt', """\
azure-functions
""")
run('cd billing_app && python "$FHOST" start')
run('python send_orders.py enrich 3')
run('cd billing_app && python "$FHOST" wait \'Moving message to queue\' 120 3 && python "$FHOST" log > run.log')
run('cd billing_app && grep -c "Executed \'Functions.enrich_order\' (Failed)" run.log')
run("cd billing_app && grep -o 'attempt [0-9]' run.log | sort | uniq -c")
run("cd billing_app && grep 'Moving message' run.log | sort | uniq -c")
run('python check.py enrich')
run('cd billing_app && python "$FHOST" stop')

# Step 7: Locally the ingest function works: the RAW_CONTAINER setting is in local.settings.json
write('lake_app/function_app.py', """\
import json
import azure.functions as func

app = func.FunctionApp(http_auth_level=func.AuthLevel.FUNCTION)


@app.route(route="ingest", methods=["POST"])
@app.blob_output(arg_name="out",
                 path="%RAW_CONTAINER%/orders/{id}.json",
                 connection="AzureWebJobsStorage")
def ingest(req: func.HttpRequest,
           out: func.Out[str]) -> func.HttpResponse:
    order = req.get_json()
    out.set(json.dumps(order))
    return func.HttpResponse(f"saved {order['id']}", status_code=201)
""")
write('lake_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",
      "visibilityTimeout": "00:00:01"
    }
  },
  "logging": {
    "logLevel": {
      "default": "Warning",
      "Function": "Information",
      "Azure.Core": "Error",
      "Host.Function.Console": "Information"
    }
  }
}
""")
write('lake_app/local.settings.json', """\
{
  "IsEncrypted": false,
  "Values": {
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "RAW_CONTAINER": "raw"
  }
}
""")
write('lake_app/requirements.txt', """\
azure-functions
""")
write('post_order.py', """\
import sys
import requests

order = {"id": sys.argv[1], "amount": float(sys.argv[2])}
r = requests.post("http://localhost:7071/api/ingest", json=order)
print(r.status_code, r.text or "(empty body)")
""")
write('list_raw.py', """\
import os
from azure.storage.blob import ContainerClient

raw = ContainerClient.from_connection_string(
    os.environ["AZURE_STORAGE_CONNECTION_STRING"], "raw")
print("raw:", sorted(raw.list_blob_names()))
""")
run('cd lake_app && python "$FHOST" start')
run('python post_order.py o-1001 25.5')
run('python list_raw.py')
run('cd lake_app && python "$FHOST" stop')

# Step 8: With only the settings the Azure app has: RAW_CONTAINER does not resolve
write('lake_app/local.settings.json', """\
{
  "IsEncrypted": false,
  "Values": {
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "AzureWebJobsStorage": "UseDevelopmentStorage=true"
  }
}
""")
run('cd lake_app && python "$FHOST" start')
run('python post_order.py o-1002 7')
run('cd lake_app && python "$FHOST" wait "Executed \'Functions.ingest\'" 30 && python "$FHOST" log')
run('cd lake_app && python "$FHOST" stop')

# Step 9: polars imported but not in requirements.txt: the worker cannot index the app
write('lake_app/function_app.py', """\
import azure.functions as func
import polars as pl              # new, but not in requirements.txt

app = func.FunctionApp(http_auth_level=func.AuthLevel.FUNCTION)


@app.route(route="ingest", methods=["POST"])
@app.blob_output(arg_name="out",
                 path="%RAW_CONTAINER%/orders/{id}.json",
                 connection="AzureWebJobsStorage")
def ingest(req: func.HttpRequest,
           out: func.Out[str]) -> func.HttpResponse:
    order = req.get_json()
    out.set(pl.DataFrame([order]).write_ndjson())
    return func.HttpResponse(f"saved {order['id']}", status_code=201)
""")
write('lake_app/local.settings.json', """\
{
  "IsEncrypted": false,
  "Values": {
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "RAW_CONTAINER": "raw"
  }
}
""")
run('cat lake_app/requirements.txt')
run('cd lake_app && python "$FHOST" start > start.log 2>&1 || echo "the host did not start"')
run('grep -o "ModuleNotFoundError.*missing module" lake_app/start.log | uniq')

That is the whole series

All 10 parts are listed on the Real Time Azure Functions Interview Questions and Answers page on this blog.

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.