第 4 部分：主要应用程序实现示例

2025-06-17

我们方案中的主要应用是部署到 Azure 应用服务的简单 Flask 应用。该应用提供一个名为 /api/v1/getcode 的公共 API 终结点，该终结点为应用中的一些其他用途生成代码（例如，为人类用户提供双重身份验证）。主应用还提供一个简单的主页，用于显示指向 API 终结点的链接。

示例的预配脚本执行以下步骤：

使用 Azure CLI 命令 az webapp up创建应用服务主机并部署代码。
为主应用创建 Azure 存储帐户（使用 az storage account create）。
在名为“code-requests”的存储帐户中创建队列（使用 az storage queue create）。
若要确保允许应用写入队列，请使用 az role assignment create 向应用分配“存储队列数据参与者”角色。有关角色的详细信息，请参阅如何使用 Azure CLI 分配角色权限。

主要应用代码如下所示：本系列下一部分提供了重要详细信息的说明。

from flask import Flask, request, jsonify
import requests, random, string, os
from datetime import datetime
from azure.keyvault.secrets import SecretClient
from azure.identity import DefaultAzureCredential
from azure.storage.queue import QueueClient

app = Flask(__name__)
app.config["DEBUG"] = True

number_url = os.environ["THIRD_PARTY_API_ENDPOINT"]

# Authenticate with Azure. First, obtain the DefaultAzureCredential
credential = DefaultAzureCredential()

# Next, get the client for the Key Vault. You must have first enabled managed identity
# on the App Service for the credential to authenticate with Key Vault.
key_vault_url = os.environ["KEY_VAULT_URL"]
keyvault_client = SecretClient(vault_url=key_vault_url, credential=credential)

# Obtain the secret: for this step to work you must add the app's service principal to
# the key vault's access policies for secret management.
api_secret_name = os.environ["THIRD_PARTY_API_SECRET_NAME"]
vault_secret = keyvault_client.get_secret(api_secret_name)

# The "secret" from Key Vault is an object with multiple properties. The key we
# want for the third-party API is in the value property. 
access_key = vault_secret.value

# Set up the Storage queue client to which we write messages
queue_url = os.environ["STORAGE_QUEUE_URL"]
queue_client = QueueClient.from_queue_url(queue_url=queue_url, credential=credential)


@app.route('/', methods=['GET'])
def home():
    return f'Home page of the main app. Make a request to <a href="./api/v1/getcode">/api/v1/getcode</a>.'


def random_char(num):
       return ''.join(random.choice(string.ascii_letters) for x in range(num))


@app.route('/api/v1/getcode', methods=['GET'])
def get_code():
    headers = {
        'Content-Type': 'application/json',
        'x-functions-key': access_key
        }

    r = requests.get(url = number_url, headers = headers)
    
    if (r.status_code != 200):       
        return "Could not get you a code.", r.status_code

    data = r.json()
    chars1 = random_char(3)
    chars2 = random_char(3)
    code_value = f"{chars1}-{data['value']}-{chars2}"
    code = { "code": code_value, "timestamp" : str(datetime.utcnow()) }

    # Log a queue message with the code for, say, a process that invalidates
    # the code after a certain period of time.
    queue_client.send_message(code)

    return jsonify(code)


if __name__ == '__main__':
    app.run()

第 5 部分 - 依赖项和环境变量 >>>

通过

第 4 部分：主要应用程序实现示例

反馈

其他资源