第 4 部分：主要應用程式實作範例

上一個部分：第三方 API 實作

這個情境的主要應用程式是一個簡單的 Flask 應用程式，你可以部署到 Azure App 服務。 應用程式會提供名為 /api/v1/getcode 的公用 API 端點，以針對應用程式中的某些其他用途產生程式代碼（例如，為人類使用者使用雙因素驗證）。 主要應用程式也提供簡單的首頁，以顯示 API 端點的連結。

範例的布建腳本會執行下列步驟：

1. 建立 App Service 主機並使用 Azure CLI 指令部署程式碼。 az webapp up

2. 請使用 az storage account create 為主應用程式建立 Azure 儲存體帳戶。

3. 在儲存帳號中建立一個名為「code-requests」的佇列，方法是使用 az storage queue create。

4. 為了確保應用程式可以寫入佇列，請使用 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 部分 - 相依性和環境變數 >>>