Azure Functions Interview Questions and Answers Part 1 - Azure Functions Fundamentals

Azure Functions Interview Q&A · Part 1 of 10

In this post we are going to cover Azure Functions fundamentals. These are 7 real Azure Functions interview questions for data engineers, asked the way they come up on a real project, and the last one is a scenario: the team wants to replace a cron server that runs ten small Python scripts, and you have to move them to Functions.

Azure Functions is the small, event-driven piece of an Azure data platform. A file lands, a message arrives or a timer fires, and a short piece of your code runs. Interviewers start here because every later question, about scaling, retries, security or cost, builds on these basics: what a trigger is, what a binding is, which hosting plan you pick and what goes in which config file.

Part 1 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: Your manager asks what Azure Functions is and where it fits in a data platform. What do you say?
4. Question 2: A new teammate asks what triggers and bindings are. How do you explain them?
5. Question 3: Someone asks how a Python function app is structured in the v2 programming model. What goes where?
6. Question 4: Leadership asks which hosting plan to pick: Flex Consumption, Premium or Dedicated. How do you decide?
7. Question 5: A teammate asks what the host.json and local.settings.json files are for. How do you explain them?
8. Question 6: Someone asks when you would NOT use Azure Functions for a data job. What do you say?
9. Scenario 7: The team wants to replace a cron server that runs ten small Python scripts. How do you move them to Functions?
10. The full script

The short answer

Say this first

Azure Functions is serverless, event-driven compute: one trigger per function, optional bindings, a Python v2 app of decorators and blueprints. Pick Flex Consumption for new serverless apps, keep host.json for runtime settings and local.settings.json on your machine, send heavy or long jobs elsewhere, and stateful workflows to Durable Functions.

Let's set up the project

Everything below runs on the real Azure Functions host running locally with Core Tools, with Azurite, Microsoft's open-source storage emulator, for blobs and queues. Each step writes a small Python function app and a client script or two. fhost is a small helper that runs func start in the background, prints the functions the host indexed, waits for log lines and prints the cleaned host log. What only Azure can show, such as hosting plans and scaling, is shown as code.

Question 1: Your manager asks what Azure Functions is and where it fits in a data platform. What do you say?

Azure Functions is serverless, event-driven compute. An event runs your small function, the platform scales it, and on the serverless plans it can scale to zero.

The events are the usual ones in a data platform: a file, a message, a timer or an HTTP call. Our code is a small Python function, and its work usually ends in the lake or on a queue for the next step.

In a data platform, Functions is the glue around the big engines. It does three kinds of jobs:

• Ingest: APIs and webhooks into the raw zone
• React: to files landing in the lake
• Glue: move messages, start and check pipelines

It is not where the heavy lifting goes. Let's put the three services side by side.

Job	Best fit
small, short, event-driven jobs: APIs, files, messages	Azure Functions
scheduled bulk copies with many connectors	Data Factory
big joins and Spark-scale transforms	Databricks

So Functions takes the small, short, frequent jobs. For Python, the v2 programming model, with decorators, is the recommended model for new apps, and runtime 4.x is the only supported version.

Real project

Keep each function small and stateless, and hand work to the next step through a queue, not through memory.

Question 2: A new teammate asks what triggers and bindings are. How do you explain them?

A trigger is what starts a function, and every function has exactly one. Bindings connect the function to data: an input binding passes data in, an output binding writes data out. Bindings are optional, and a function can have as many as you need.

Let's build one. Our on_order function has a queue trigger on the orders queue and a blob output binding on processed/{id}.json, both as decorators. There is no storage client code: the function reads the message, logs it and sets the output. We run fhost start, then python send.py orders to send two orders, wait for both runs, and list the container with python blobs.py processed.

orders_app/function_app.py

import logging
import azure.functions as func

app = func.FunctionApp()


@app.queue_trigger(arg_name="msg", queue_name="orders",
                   connection="AzureWebJobsStorage")         # trigger
@app.blob_output(arg_name="out", path="processed/{id}.json",
                 connection="AzureWebJobsStorage")         # output
def on_order(msg: func.QueueMessage, out: func.Out[str]) -> None:
    order = msg.get_json()
    logging.info("order %s amount %s", order["id"], order["amount"])
    out.set(msg.get_body().decode())     # the binding writes it

orders_app/host.json

{
  "version": "2.0",
  "logging": {
    "logLevel": {
      "default": "Warning",
      "Function": "Information",
      "Azure.Core": "Error",
      "Host.Function.Console": "Information"
    }
  },
  "extensions": {
    "queues": {
      "batchSize": 1,
      "newBatchThreshold": 0,
      "maxPollingInterval": "00:00:01"
    }
  },
  "extensionBundle": {
    "id": "Microsoft.Azure.Functions.ExtensionBundle",
    "version": "[4.*, 5.0.0)"
  }
}

orders_app/local.settings.json

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

orders_app/requirements.txt

azure-functions

send.py

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

queue = QueueClient.from_connection_string(
    os.environ["AZURE_STORAGE_CONNECTION_STRING"], sys.argv[1],
    message_encode_policy=TextBase64EncodePolicy())  # host: base64
try:
    queue.create_queue()
except ResourceExistsError:
    pass
for oid, amount in [("o-1001", 25.5), ("o-1002", 14.0)]:
    queue.send_message(json.dumps({"id": oid, "amount": amount}))
print("sent 2 orders to", sys.argv[1])

blobs.py

import os, sys
from azure.storage.blob import ContainerClient

box = ContainerClient.from_connection_string(
    os.environ["AZURE_STORAGE_CONNECTION_STRING"], sys.argv[1])
for blob in box.list_blobs():
    text = box.download_blob(blob.name).readall().decode()
    print(f"{sys.argv[1]}/{blob.name}: {text}")

Terminal

$ cd orders_app && fhost start
host started, functions:
  on_order: queueTrigger
$ python send.py orders
sent 2 orders to orders
$ cd orders_app && fhost wait "Executed 'Functions.on_order'" 60 2 && fhost log
Executing 'Functions.on_order' (Reason='New queue message detected on 'orders'.')
order o-1001 amount 25.5
Executed 'Functions.on_order' (Succeeded)
Executing 'Functions.on_order' (Reason='New queue message detected on 'orders'.')
order o-1002 amount 14.0
Executed 'Functions.on_order' (Succeeded)
$ python blobs.py processed
processed/o-1001.json: {"id": "o-1001", "amount": 25.5}
processed/o-1002.json: {"id": "o-1002", "amount": 14.0}
$ cd orders_app && fhost stop
host stopped

The host indexes on_order: queueTrigger, and the log shows one run per message: order o-1001 amount 25.5 and order o-1002 amount 14.0, both Succeeded. The {id} in the blob path comes from the message JSON, so processed ends up with processed/o-1001.json and processed/o-1002.json.

Not every service has every direction. Timer is a trigger only, Table Storage has input and output but no trigger, and Blob and Cosmos DB have all three.

Service	Trigger	Input	Output
Blob Storage	yes	yes	yes
Queue Storage	yes	-	yes
Table Storage	-	yes	yes
Timer	yes	-	-
Cosmos DB	yes	yes	yes

Tip

connection names an app setting, not the secret. The same code reads Azurite here and a real storage account in Azure.

Question 3: Someone asks how a Python function app is structured in the v2 programming model. What goes where?

The entry point is function_app.py. It creates the app object, and decorators on plain Python functions declare the triggers and bindings. There is no function.json per function anymore. That is the v1 model, a folder per function, kept for legacy apps.

As the app grows, we use blueprints. Each business area gets its own file with a Blueprint object, and function_app.py registers them. Shared code, like business rules, goes in a normal package, here called shared.

Next to the code sit host.json for runtime settings, local.settings.json for your machine only, requirements.txt listing every package, azure-functions included, and .funcignore for files you don't publish, like tests. Let's write the files and list them with find.

shop_app/function_app.py

import azure.functions as func
from bp_orders import bp as orders_bp
from bp_payments import bp as payments_bp

app = func.FunctionApp(http_auth_level=func.AuthLevel.FUNCTION)
app.register_functions(orders_bp)       # one blueprint per area
app.register_functions(payments_bp)

shop_app/host.json

{
  "version": "2.0",
  "logging": {
    "logLevel": {
      "default": "Warning",
      "Function": "Information",
      "Azure.Core": "Error",
      "Host.Function.Console": "Information"
    }
  },
  "extensions": {
    "queues": {
      "batchSize": 1,
      "newBatchThreshold": 0,
      "maxPollingInterval": "00:00:01"
    }
  },
  "extensionBundle": {
    "id": "Microsoft.Azure.Functions.ExtensionBundle",
    "version": "[4.*, 5.0.0)"
  }
}

shop_app/local.settings.json

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

shop_app/requirements.txt

azure-functions

shop_app/bp_orders.py

import azure.functions as func
from shared.rules import is_valid

bp = func.Blueprint()


@bp.route(route="orders", methods=["POST"],
          auth_level=func.AuthLevel.FUNCTION)
def add_order(req: func.HttpRequest) -> func.HttpResponse:
    order = req.get_json()
    if not is_valid(order):
        return func.HttpResponse("rejected", status_code=400)
    return func.HttpResponse(f"accepted {order['id']}",
                             status_code=202)

shop_app/bp_payments.py

import logging
import azure.functions as func

bp = func.Blueprint()


@bp.timer_trigger(schedule="0 0 2 * * *", arg_name="timer",
                  run_on_startup=False)
def reconcile_payments(timer: func.TimerRequest) -> None:
    logging.info("reconcile yesterday's payments")

shop_app/shared/__init__.py

shop_app/shared/rules.py

def is_valid(order: dict) -> bool:
    """A business rule shared by several functions."""
    return bool(order.get("id")) and order.get("amount", 0) > 0

shop_app/tests/test_rules.py

from shared.rules import is_valid


def test_rejects_zero_amount():
    assert not is_valid({"id": "o-1", "amount": 0})

shop_app/.funcignore

.venv
.vscode
local.settings.json
tests

Terminal

$ cd shop_app && find . -type f | sort
./.funcignore
./bp_orders.py
./bp_payments.py
./function_app.py
./host.json
./local.settings.json
./requirements.txt
./shared/__init__.py
./shared/rules.py
./tests/test_rules.py

We get ten files: the two blueprints bp_orders.py and bp_payments.py, shared/rules.py, tests/test_rules.py and the config files. Now we run fhost start to see what the host indexes.

Terminal

$ cd shop_app && fhost start
host started, functions:
  add_order: [POST] http://localhost:7071/api/orders
  reconcile_payments: timerTrigger
$ cd shop_app && fhost stop
host stopped

It found both blueprints: add_order: [POST] http://localhost:7071/api/orders from orders and reconcile_payments: timerTrigger from payments.

Pitfall

Never name a folder after a standard library module, like email/ or json/, and keep top-level imports light, because the host imports the whole app every time it starts.

Question 4: Leadership asks which hosting plan to pick: Flex Consumption, Premium or Dedicated. How do you decide?

For new serverless apps, Flex Consumption is my default. It scales on events, down to zero and per function, it supports virtual network integration, and optional always ready instances cut cold starts.

Let's go through the three plans:

Plan	Best for	Scaling	Memory per instance	Slots
Flex Consumption	new serverless apps (Linux)	events, to zero, per function; always ready optional	512 MB, 2 GB (default), 4 GB	none: rolling updates
Premium	always warm, Windows, containers	events; at least 1 instance always warm	3.5 to 14 GB	3 per app
Dedicated	an App Service plan you already run	you scale it: no event scaling	1.75 to 256 GB	1 to 20

First, Premium. It fits when you need instances that are always warm, Windows, containers, deployment slots, or bigger instances, up to 14 GB. At least one instance is always running, and billed.

Then Dedicated. This is an App Service plan you already pay for. Cost is predictable, but there is no event-driven scaling: you scale it yourself, and long runs need Always On.

On timeouts, all three default to 30 minutes and can run unbounded. But no HTTP response can take longer than 230 seconds, on any plan.

The legacy Consumption plan on Linux retires on 30 September 2028, gets no new features, and Python 3.12 is its last Python. So new apps go to Flex Consumption. Here is how we create one with the Azure CLI: Linux, Python 3.13 and 2 GB instances.

create_app.sh (Azure CLI, not run)

# Flex Consumption app: Linux, Python 3.13, 2 GB instances
az functionapp create --name fn-orders-prod \
  --resource-group rg-data \
  --storage-account stfnordersprod \
  --flexconsumption-location eastus \
  --runtime python --runtime-version 3.13 \
  --instance-memory 2048 --maximum-instance-count 100

Real project

Don't pick Premium just for cold starts. Try Flex Consumption with a few always ready instances first, and keep Premium for the cases that need it, like Windows or slots.

Question 5: A teammate asks what the host.json and local.settings.json files are for. How do you explain them?

host.json configures the runtime for the whole app, and it is deployed with your code. local.settings.json holds the app settings for your machine only, and it is not published by default.

Let's start with host.json. It has log levels, trigger settings like the queue batch size, the extension bundle and the function timeout, here 00:10:00. Batch size 1 keeps our log in order.

Then local.settings.json. It holds local app settings, like the storage connection, which here points at Azurite. The .gitignore that func init generates leaves it out of source control.

Its Values become environment variables, just like app settings in Azure. We start the host and call our settings endpoint with curl.

cfg_app/function_app.py

import json, logging, os
import azure.functions as func

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


@app.route(route="settings", methods=["GET"])
def show_settings(req: func.HttpRequest) -> func.HttpResponse:
    names = ["SOURCE_SYSTEM", "BATCH_SIZE", "NOT_SET"]
    values = {n: os.getenv(n) for n in names}      # app settings
    values["BATCH_SIZE type"] = type(os.getenv("BATCH_SIZE")).__name__
    return func.HttpResponse(json.dumps(values),
                             mimetype="application/json")


@app.queue_trigger(arg_name="msg", queue_name="%ORDERS_QUEUE%",
                   connection="AzureWebJobsStorage")
def load_orders(msg: func.QueueMessage) -> None:
    logging.info("load_orders read %s from %s",
                 msg.get_json()["id"], os.getenv("ORDERS_QUEUE"))

cfg_app/host.json

{
  "version": "2.0",
  "functionTimeout": "00:10:00",
  "logging": {
    "logLevel": {
      "default": "Warning",
      "Function": "Information",
      "Azure.Core": "Error",
      "Host.Function.Console": "Information"
    }
  },
  "extensions": {
    "queues": {
      "batchSize": 1,
      "newBatchThreshold": 0,
      "maxPollingInterval": "00:00:01"
    }
  },
  "extensionBundle": {
    "id": "Microsoft.Azure.Functions.ExtensionBundle",
    "version": "[4.*, 5.0.0)"
  }
}

cfg_app/local.settings.json

{
  "IsEncrypted": false,
  "Values": {
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "SOURCE_SYSTEM": "oracle-fusion",
    "BATCH_SIZE": "500",
    "ORDERS_QUEUE": "orders-dev"
  }
}

cfg_app/requirements.txt

azure-functions

Terminal

$ cd cfg_app && fhost start
host started, functions:
  show_settings: [GET] http://localhost:7071/api/settings
  load_orders: queueTrigger
$ curl -s http://localhost:7071/api/settings | python -m json.tool
{
    "SOURCE_SYSTEM": "oracle-fusion",
    "BATCH_SIZE": "500",
    "NOT_SET": null,
    "BATCH_SIZE type": "str"
}

SOURCE_SYSTEM comes back as oracle-fusion, BATCH_SIZE is the string "500", not a number (the type is str), and the missing NOT_SET is null.

Settings also work inside bindings. The queue trigger names its queue %ORDERS_QUEUE%, an app setting in percent signs. We send two orders with python send.py orders-dev and check the log.

Terminal

$ python send.py orders-dev
sent 2 orders to orders-dev
$ cd cfg_app && fhost wait "Executed 'Functions.load_orders'" 60 2
$ cd cfg_app && fhost log | grep load_orders
Executing 'Functions.load_orders' (Reason='New queue message detected on 'orders-dev'.')
load_orders read o-1001 from orders-dev
Executed 'Functions.load_orders' (Succeeded)
Executing 'Functions.load_orders' (Reason='New queue message detected on 'orders-dev'.')
load_orders read o-1002 from orders-dev
Executed 'Functions.load_orders' (Succeeded)
$ cd cfg_app && fhost stop
host stopped

load_orders read o-1001 and o-1002 from orders-dev. In production the same code reads another queue.

Tip

To change a host.json value per environment without editing the file, set an app setting like AzureFunctionsJobHost__functionTimeout: the prefix, two underscores, then the path to the value.

Question 6: Someone asks when you would NOT use Azure Functions for a data job. What do you say?

When the job is heavy, long or stateful. Functions is built for small, short, event-driven work, and its limits show where it stops.

Let's look at each case.

Avoid Functions for	Because	Use instead
big joins, Spark-scale transforms	memory: 1.5 GB (Consumption), 4 GB max (Flex); Flex temp disk 0.8 GB	Databricks
scheduled bulk copies, many sources	connectors, monitoring and retries are built in there	Data Factory
one long run	Consumption: 10 min max; Flex, Premium: 60 min grace on scale-in; HTTP: 230 s	split it via a queue, or a batch job
long workflows with state	a plain function keeps no state between runs	Durable Functions

First, big joins and Spark-scale transforms go to Databricks. One instance has limited memory: 1.5 GB on the legacy Consumption plan, 4 GB at most on Flex Consumption, and Flex has less than 1 GB of temporary disk.

Then, scheduled bulk copies from many sources go to Data Factory, which has the connectors, monitoring and retries built in. Rewriting copy activities as functions is code you then maintain.

Long single runs are the third case. Legacy Consumption stops a run at 10 minutes, Flex and Premium allow only 60 minutes of grace on scale-in, and any HTTP response is cut at 230 seconds. Split the work through a queue, or use a batch job.

And a plain function keeps no state between runs, so long workflows with state need Durable Functions.

Rule of thumb

If one run needs more than a few minutes, gigabytes of memory or a cluster, it is not a function. A function can still start that job and check on it.

Scenario 7: The team wants to replace a cron server that runs ten small Python scripts. How do you move them to Functions?

One function app, one timer function per script. The script's logic moves into a plain module the timer calls, and each cron line becomes an NCRONTAB schedule, with a seconds field in front.

Let's map three of the ten jobs:

Old cron script	crontab	Timer function	NCRONTAB (seconds first)
export_orders.py	30 1 * * *	export_orders	0 30 1 * * *
refresh_fx_rates.py	0 */4 * * *	refresh_fx_rates	0 0 */4 * * *
purge_temp.py	0 3 * * 0	purge_temp	0 0 3 * * 0

Timers run in UTC. WEBSITE_TIME_ZONE works on Windows and on Linux Premium or Dedicated, but not on Flex Consumption, so we write the UTC time and plan for daylight saving. Each schedule lives in an app setting, like %ORDERS_SCHEDULE%, so a new time is a settings change, not a deploy.

Now let's run it. Locally, the admin API runs each timer on demand. We run fhost start, then bash run_now.sh, which posts to /admin/functions/<name> for each job, and print the log.

jobs_app/function_app.py

import logging
import azure.functions as func
import jobs              # the old cron scripts, as plain functions

app = func.FunctionApp()


@app.timer_trigger(schedule="%ORDERS_SCHEDULE%", arg_name="timer",
                   run_on_startup=False)
def export_orders(timer: func.TimerRequest) -> None:
    logging.info(jobs.export_orders())


@app.timer_trigger(schedule="%FX_SCHEDULE%", arg_name="timer",
                   run_on_startup=False)
def refresh_fx_rates(timer: func.TimerRequest) -> None:
    logging.info(jobs.refresh_fx_rates())


@app.timer_trigger(schedule="%PURGE_SCHEDULE%", arg_name="timer",
                   run_on_startup=False)
def purge_temp(timer: func.TimerRequest) -> None:
    logging.info(jobs.purge_temp())

jobs_app/host.json

{
  "version": "2.0",
  "logging": {
    "logLevel": {
      "default": "Warning",
      "Function": "Information",
      "Azure.Core": "Error",
      "Host.Function.Console": "Information"
    }
  },
  "extensions": {
    "queues": {
      "batchSize": 1,
      "newBatchThreshold": 0,
      "maxPollingInterval": "00:00:01"
    }
  },
  "extensionBundle": {
    "id": "Microsoft.Azure.Functions.ExtensionBundle",
    "version": "[4.*, 5.0.0)"
  }
}

jobs_app/local.settings.json

{
  "IsEncrypted": false,
  "Values": {
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "ORDERS_SCHEDULE": "0 30 1 * * *",
    "FX_SCHEDULE": "0 0 */4 * * *",
    "PURGE_SCHEDULE": "0 0 3 * * 0"
  }
}

jobs_app/requirements.txt

azure-functions

jobs_app/jobs.py

"""The old cron scripts as plain functions: easy to unit test."""


def export_orders() -> str:
    # ... the old export_orders.py code ...
    return "export_orders: yesterday's orders written to raw/orders"


def refresh_fx_rates() -> str:
    # ... the old refresh_fx_rates.py code ...
    return "refresh_fx_rates: currency rates refreshed"


def purge_temp() -> str:
    # ... the old purge_temp.py code ...
    return "purge_temp: temp files older than 7 days removed"

jobs_app/run_now.sh

# Run each timer now through the local admin API (Azure: master key)
for job in export_orders refresh_fx_rates purge_temp; do
  curl -s -o /dev/null -w "$job %{http_code}\n" -X POST \
    -H "Content-Type: application/json" -d '{"input": ""}' \
    "http://localhost:7071/admin/functions/$job"
  fhost wait "Executed 'Functions.$job'" 30
done

Terminal

$ cd jobs_app && fhost start
host started, functions:
  export_orders: timerTrigger
  purge_temp: timerTrigger
  refresh_fx_rates: timerTrigger
$ cd jobs_app && bash run_now.sh && fhost log
export_orders 202
refresh_fx_rates 202
purge_temp 202
Executing 'Functions.export_orders' (Reason='This function was programmatically called via the host
  APIs.')
export_orders: yesterday's orders written to raw/orders
Executed 'Functions.export_orders' (Succeeded)
Executing 'Functions.refresh_fx_rates' (Reason='This function was programmatically called via the host
  APIs.')
refresh_fx_rates: currency rates refreshed
Executed 'Functions.refresh_fx_rates' (Succeeded)
Executing 'Functions.purge_temp' (Reason='This function was programmatically called via the host APIs.')
purge_temp: temp files older than 7 days removed
Executed 'Functions.purge_temp' (Succeeded)
$ cd jobs_app && fhost stop
host stopped

The host indexes three timerTrigger functions, each admin call returns 202, and each job logs its work, for example export_orders: yesterday's orders written to raw/orders. Locally the admin API needs no key; in Azure it needs the master key.

Before go-live, remember how timers behave:

• Singleton: a storage lock, one run at a time
• No retry: by default, so make jobs safe to rerun
• run_on_startup: never in production

Then add monitoring:

• Application Insights: failures and run history
• Alert: when a job did not run
• Key Vault: secrets out of config files

Real project

The alert for a job that did not run is the one people forget, and it is the one that catches a silent failure.

Key points to remember

• Serverless, event-driven glue for small, short jobs; heavy work goes to Data Factory or Databricks
• One trigger per function; bindings are optional, input or output, and name an app setting
• A Python v2 app: function_app.py, blueprints, a shared package, requirements.txt
• Flex Consumption for new serverless apps; Premium for always warm; Dedicated for existing App Service
• host.json = runtime settings, deployed; local.settings.json = local only, not published by default

The full script

Here is a Python script that 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 #1 - Azure Functions fundamentals
# 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_01_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: A queue trigger and a blob output binding: two orders in, two blobs out
write('orders_app/function_app.py', """\
import logging
import azure.functions as func

app = func.FunctionApp()


@app.queue_trigger(arg_name="msg", queue_name="orders",
                   connection="AzureWebJobsStorage")         # trigger
@app.blob_output(arg_name="out", path="processed/{id}.json",
                 connection="AzureWebJobsStorage")         # output
def on_order(msg: func.QueueMessage, out: func.Out[str]) -> None:
    order = msg.get_json()
    logging.info("order %s amount %s", order["id"], order["amount"])
    out.set(msg.get_body().decode())     # the binding writes it
""")
write('orders_app/host.json', """\
{
  "version": "2.0",
  "logging": {
    "logLevel": {
      "default": "Warning",
      "Function": "Information",
      "Azure.Core": "Error",
      "Host.Function.Console": "Information"
    }
  },
  "extensions": {
    "queues": {
      "batchSize": 1,
      "newBatchThreshold": 0,
      "maxPollingInterval": "00:00:01"
    }
  },
  "extensionBundle": {
    "id": "Microsoft.Azure.Functions.ExtensionBundle",
    "version": "[4.*, 5.0.0)"
  }
}
""")
write('orders_app/local.settings.json', """\
{
  "IsEncrypted": false,
  "Values": {
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "AzureWebJobsStorage": "UseDevelopmentStorage=true"
  }
}
""")
write('orders_app/requirements.txt', """\
azure-functions
""")
write('send.py', """\
import json, os, sys
from azure.core.exceptions import ResourceExistsError
from azure.storage.queue import QueueClient, TextBase64EncodePolicy

queue = QueueClient.from_connection_string(
    os.environ["AZURE_STORAGE_CONNECTION_STRING"], sys.argv[1],
    message_encode_policy=TextBase64EncodePolicy())  # host: base64
try:
    queue.create_queue()
except ResourceExistsError:
    pass
for oid, amount in [("o-1001", 25.5), ("o-1002", 14.0)]:
    queue.send_message(json.dumps({"id": oid, "amount": amount}))
print("sent 2 orders to", sys.argv[1])
""")
write('blobs.py', """\
import os, sys
from azure.storage.blob import ContainerClient

box = ContainerClient.from_connection_string(
    os.environ["AZURE_STORAGE_CONNECTION_STRING"], sys.argv[1])
for blob in box.list_blobs():
    text = box.download_blob(blob.name).readall().decode()
    print(f"{sys.argv[1]}/{blob.name}: {text}")
""")
run('cd orders_app && python "$FHOST" start')
run('python send.py orders')
run('cd orders_app && python "$FHOST" wait "Executed \'Functions.on_order\'" 60 2 && python "$FHOST" log')
run('python blobs.py processed')
run('cd orders_app && python "$FHOST" stop')

# Step 2: A Python v2 app with two blueprints and a shared package: the files
write('shop_app/function_app.py', """\
import azure.functions as func
from bp_orders import bp as orders_bp
from bp_payments import bp as payments_bp

app = func.FunctionApp(http_auth_level=func.AuthLevel.FUNCTION)
app.register_functions(orders_bp)       # one blueprint per area
app.register_functions(payments_bp)
""")
write('shop_app/host.json', """\
{
  "version": "2.0",
  "logging": {
    "logLevel": {
      "default": "Warning",
      "Function": "Information",
      "Azure.Core": "Error",
      "Host.Function.Console": "Information"
    }
  },
  "extensions": {
    "queues": {
      "batchSize": 1,
      "newBatchThreshold": 0,
      "maxPollingInterval": "00:00:01"
    }
  },
  "extensionBundle": {
    "id": "Microsoft.Azure.Functions.ExtensionBundle",
    "version": "[4.*, 5.0.0)"
  }
}
""")
write('shop_app/local.settings.json', """\
{
  "IsEncrypted": false,
  "Values": {
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "AzureWebJobsStorage": "UseDevelopmentStorage=true"
  }
}
""")
write('shop_app/requirements.txt', """\
azure-functions
""")
write('shop_app/bp_orders.py', """\
import azure.functions as func
from shared.rules import is_valid

bp = func.Blueprint()


@bp.route(route="orders", methods=["POST"],
          auth_level=func.AuthLevel.FUNCTION)
def add_order(req: func.HttpRequest) -> func.HttpResponse:
    order = req.get_json()
    if not is_valid(order):
        return func.HttpResponse("rejected", status_code=400)
    return func.HttpResponse(f"accepted {order['id']}",
                             status_code=202)
""")
write('shop_app/bp_payments.py', """\
import logging
import azure.functions as func

bp = func.Blueprint()


@bp.timer_trigger(schedule="0 0 2 * * *", arg_name="timer",
                  run_on_startup=False)
def reconcile_payments(timer: func.TimerRequest) -> None:
    logging.info("reconcile yesterday's payments")
""")
write('shop_app/shared/__init__.py', """\

""")
write('shop_app/shared/rules.py', 'def is_valid(order: dict) -> bool:\n    """A business rule shared by several functions."""\n    return bool(order.get("id")) and order.get("amount", 0) > 0\n')
write('shop_app/tests/test_rules.py', """\
from shared.rules import is_valid


def test_rejects_zero_amount():
    assert not is_valid({"id": "o-1", "amount": 0})
""")
write('shop_app/.funcignore', """\
.venv
.vscode
local.settings.json
tests
""")
run('cd shop_app && find . -type f | sort')

# Step 3: The host indexes the functions from both blueprints
run('cd shop_app && python "$FHOST" start')
run('cd shop_app && python "$FHOST" stop')

# Step 4: Values in local.settings.json are environment variables (always strings)
write('cfg_app/function_app.py', """\
import json, logging, os
import azure.functions as func

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


@app.route(route="settings", methods=["GET"])
def show_settings(req: func.HttpRequest) -> func.HttpResponse:
    names = ["SOURCE_SYSTEM", "BATCH_SIZE", "NOT_SET"]
    values = {n: os.getenv(n) for n in names}      # app settings
    values["BATCH_SIZE type"] = type(os.getenv("BATCH_SIZE")).__name__
    return func.HttpResponse(json.dumps(values),
                             mimetype="application/json")


@app.queue_trigger(arg_name="msg", queue_name="%ORDERS_QUEUE%",
                   connection="AzureWebJobsStorage")
def load_orders(msg: func.QueueMessage) -> None:
    logging.info("load_orders read %s from %s",
                 msg.get_json()["id"], os.getenv("ORDERS_QUEUE"))
""")
write('cfg_app/host.json', """\
{
  "version": "2.0",
  "functionTimeout": "00:10:00",
  "logging": {
    "logLevel": {
      "default": "Warning",
      "Function": "Information",
      "Azure.Core": "Error",
      "Host.Function.Console": "Information"
    }
  },
  "extensions": {
    "queues": {
      "batchSize": 1,
      "newBatchThreshold": 0,
      "maxPollingInterval": "00:00:01"
    }
  },
  "extensionBundle": {
    "id": "Microsoft.Azure.Functions.ExtensionBundle",
    "version": "[4.*, 5.0.0)"
  }
}
""")
write('cfg_app/local.settings.json', """\
{
  "IsEncrypted": false,
  "Values": {
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "SOURCE_SYSTEM": "oracle-fusion",
    "BATCH_SIZE": "500",
    "ORDERS_QUEUE": "orders-dev"
  }
}
""")
write('cfg_app/requirements.txt', """\
azure-functions
""")
run('cd cfg_app && python "$FHOST" start')
run('curl -s http://localhost:7071/api/settings | python -m json.tool')

# Step 5: A queue name from an app setting: %ORDERS_QUEUE% resolves to orders-dev
run('python send.py orders-dev')
run('cd cfg_app && python "$FHOST" wait "Executed \'Functions.load_orders\'" 60 2')
run('cd cfg_app && python "$FHOST" log | grep load_orders')
run('cd cfg_app && python "$FHOST" stop')

# Step 6: Three cron scripts as timer functions, schedules in app settings, run on demand
write('jobs_app/function_app.py', """\
import logging
import azure.functions as func
import jobs              # the old cron scripts, as plain functions

app = func.FunctionApp()


@app.timer_trigger(schedule="%ORDERS_SCHEDULE%", arg_name="timer",
                   run_on_startup=False)
def export_orders(timer: func.TimerRequest) -> None:
    logging.info(jobs.export_orders())


@app.timer_trigger(schedule="%FX_SCHEDULE%", arg_name="timer",
                   run_on_startup=False)
def refresh_fx_rates(timer: func.TimerRequest) -> None:
    logging.info(jobs.refresh_fx_rates())


@app.timer_trigger(schedule="%PURGE_SCHEDULE%", arg_name="timer",
                   run_on_startup=False)
def purge_temp(timer: func.TimerRequest) -> None:
    logging.info(jobs.purge_temp())
""")
write('jobs_app/host.json', """\
{
  "version": "2.0",
  "logging": {
    "logLevel": {
      "default": "Warning",
      "Function": "Information",
      "Azure.Core": "Error",
      "Host.Function.Console": "Information"
    }
  },
  "extensions": {
    "queues": {
      "batchSize": 1,
      "newBatchThreshold": 0,
      "maxPollingInterval": "00:00:01"
    }
  },
  "extensionBundle": {
    "id": "Microsoft.Azure.Functions.ExtensionBundle",
    "version": "[4.*, 5.0.0)"
  }
}
""")
write('jobs_app/local.settings.json', """\
{
  "IsEncrypted": false,
  "Values": {
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "ORDERS_SCHEDULE": "0 30 1 * * *",
    "FX_SCHEDULE": "0 0 */4 * * *",
    "PURGE_SCHEDULE": "0 0 3 * * 0"
  }
}
""")
write('jobs_app/requirements.txt', """\
azure-functions
""")
write('jobs_app/jobs.py', '"""The old cron scripts as plain functions: easy to unit test."""\n\n\ndef export_orders() -> str:\n    # ... the old export_orders.py code ...\n    return "export_orders: yesterday\'s orders written to raw/orders"\n\n\ndef refresh_fx_rates() -> str:\n    # ... the old refresh_fx_rates.py code ...\n    return "refresh_fx_rates: currency rates refreshed"\n\n\ndef purge_temp() -> str:\n    # ... the old purge_temp.py code ...\n    return "purge_temp: temp files older than 7 days removed"\n')
write('jobs_app/run_now.sh', '# Run each timer now through the local admin API (Azure: master key)\nfor job in export_orders refresh_fx_rates purge_temp; do\n  curl -s -o /dev/null -w "$job %{http_code}\\n" -X POST \\\n    -H "Content-Type: application/json" -d \'{"input": ""}\' \\\n    "http://localhost:7071/admin/functions/$job"\n  fhost wait "Executed \'Functions.$job\'" 30\ndone\n')
run('cd jobs_app && python "$FHOST" start')
run('cd jobs_app && bash run_now.sh && python "$FHOST" log')
run('cd jobs_app && python "$FHOST" stop')

Next in this series

Part 2: Triggers and Bindings in Practice

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.