创建 Python 云原生 Web 应用，第 2 部分：添加 Azure Cosmos DB 数据库
在本文中，我们将继续深入，为我们的 Flask Web 应用添加一个 Azure Cosmos DB 数据库。
在上一篇文章中，我们在 Azure 中创建了一个托管 Python Flask 任务管理应用的 Web 应用服务。然而，我们当前的应用程序还没有任何功能，例如保存或检索数据。
由于我们是 Python 开发者而不是数据库管理员，我们希望避免设置和管理任何服务器或其他基础设施。因此，在本系列的最后一篇文章中，我们将使用 Azure Cosmos DB 服务为我们的应用程序添加一个数据库。然后，我们将终于拥有一个功能齐全的基本任务管理 Web 应用程序。
创建 Cosmos DB 服务
在创建数据库和保存数据之前，我们需要一种方法让 Visual Studio Code (VS Code) 打开和管理我们的服务。我们将使用 Azure Database 扩展来实现这一点。
首先，我们在 VS Code 中打开上一篇文章中的项目文件夹，然后转到左侧的扩展菜单 (CTRL+SHIFT+X)。我们搜索 Azure Databases for VS Code 扩展并进行安装。安装此扩展后，我们打开左侧的 Azure (CTRL+SHIFT+A) 菜单。现在应该有一个数据库部分。
接下来，我们需要为我们的应用程序创建一个新数据库。所以，我们点击数据库部分旁边的加号 (+) 图标开始创建过程。要创建数据库，我们需要
- 指定要在其中创建资源的订阅。
- 选择数据库服务器类型。对于 Python，我们使用 Azure Cosmos DB for MongoDB API。
- 提供一个帐户名称，例如“flasktutorialdbaccount”。
- 指定容量模型。我们在此处选择无服务器，但如果您需要保证的性能，可以选择预配吞吐量。
- 选择或创建资源组。我们之前的应用服务创建了一个名为“appsvc_linux_centralus”（或类似名称）的资源组，所以我们在此处使用它。
- 选择一个位置，例如“US East”。
完成此过程后，我们现在应该在Azure菜单的数据库部分下的订阅中看到一个 Cosmos DB 服务。我们右键单击此服务并选择复制连接字符串选项。
我们将连接字符串存储为 Azure App Service 中的环境变量。所以，我们打开Azure选项卡下的应用服务部分，选择订阅，然后选择我们的 Web 应用。在我们的 Web 应用下方有一个应用程序设置区域。我们右键单击应用程序设置标题并选择添加新设置。我们给它命名，例如“MONGODB”，然后粘贴连接字符串。
我们还将此数据库添加到我们的本地应用程序中，以便在本地测试保存和输入数据。更好的开发实践是运行不同的数据库服务或在本地运行服务（这很有挑战性，因为存储管理器目前不支持 MongoDB API）。
为了简单起见，我们将重用当前的数据库连接，方法是转到运行菜单，选择打开配置，然后将以下行添加到env部分
"MONGODB": "<your_connection_string>"
使用 PyMongo 连接到 Cosmos DB
我们将使用PyMongo 客户端将我们的应用程序连接到数据库。我们可以使用以下命令安装此客户端
pip install pymongo
我们还需要手动更新我们的 requirements.txt 文件，方法是添加行 pymongo==3.12.1 或使用命令 pip freeze > requirements.txt。一旦我们安装了 PyMongo，我们就可以像 Flask 一样将其导入到我们的应用程序中。
现在我们在基目录中创建一个名为“database.py”的新文件。此文件将包含我们所有的数据库代码，用于保存、更新和检索任务。然后，我们将以下代码添加到文件顶部
from pymongo import MongoClient
import os, datetime, uuid
mongo_client = MongoClient(os.environ["MONGODB"])
mongo_db = mongo_client['TaskManagerDB']
mongo_collection_tasks = mongo_db['Tasks']
前两行导入了 Mongo 客户端和一些我们将稍后需要的系统库，用于访问环境变量、使用日期和时间函数以及生成唯一标识符。
接下来，我们通过使用 MONGODB 环境变量作为连接字符串来加载 Mongo 客户端。最后，我们创建一个指向名为“TaskManagerDB”的数据库的变量，并将其集合设置为“Tasks”。
Cosmos DB 将数据存储在文档中，而不是像 Python 开发者熟悉的 JSON 文件那样存储在表中的行。但是，我们需要用一些临时数据初始化我们的数据库，因为数据库和集合都不存在。
所以，让我们通过在一个名为“initialize_db”的函数中生成一个任务 JSON 来创建我们的初始数据集。我们将以下代码输入到我们的 database.py 文件中
def initialize_db():
if "Tasks" not in mongo_db.list_collection_names():
mongo_collection_tasks.insert_one({
"taskid": str(uuid.uuid4()),
"name": "Test Task 1",
"Description": "Description of Test Task 1",
"Due": datetime.datetime.utcnow() + datetime.timedelta(days=20),
"Completed": False,
"Created": datetime.datetime.utcnow(),
"Modified": datetime.datetime.utcnow()
})
此函数将检查以确保 Tasks 不存在（只有当集合包含数据时才存在）。如果 Tasks 不存在，该函数将插入一个任务并生成一个唯一 ID。
如果您想要更多初始数据，您可以重复此命令几次以插入多个任务（我们创建了四个，但简化了上面的代码）。
在我们离开数据库脚本之前，让我们也创建另外两个函数来检索数据，方法是输入以下代码
def find_tasks(starting_point):
tasks = []
for task in mongo_collection_tasks.find(skip=starting_point, limit=10):
tasks.append(task)
return tasks
def find_task(task_id):
return mongo_collection_tasks.find_one({"taskid": task_id})
这两个函数将从一个起始点检索十个任务（以便我们可以分页数据），或者检索基于内部 ID 的特定任务。我们切换回我们的app.py并在文件顶部使用代码 from .database import database 添加一个新的导入。
接下来，在定义了 app 变量后，我们还使用代码 database.initialize_db() 初始化我们的数据库。
最后，让我们也修改我们的 home 函数，使其接受来自 find_tasks 函数的数据，如下所示
@app.route("/")
def home():
return render_template(
"home.html", data=database.find_tasks(0)
)
此更改将一个名为“data”的变量推送到模板中，该变量包含一个任务对象的数组，我们可以使用它。
我们的最后一步是更新我们的 home.html 模板，删除大部分占位符数据，并使用以下代码循环遍历我们数据库数据的 data 元素
{% extends "layout.html" %}
{% block body %}
<div class="container px-4 py-5">
<h2 class="pb-2 border-bottom">Current Tasks</h2>
<div class="row row-cols-1 row-cols-sm-1 row-cols-md-2 row-cols-lg-3 g-4 py-5">
<!-- Repeat Block for Each Task-->
{% for task in data %}
<div class="col d-flex align-items-center card">
<div class="card-body">
<h4 class="fw-bold mb-0">{{task.name}}</h4>
<p>{{task.description}}</p>
<a href="/viewtask/{{task._id}}" class="btn btn-primary">View Task</a>
<a href="/edittask/{{task._id}}" class="btn btn-primary">Edit Task</a>
</div>
</div>
{% endfor %}
<!-- End Repeat -->
</div>
</div>
{% endblock %}
我们代码中的 {% for task in data %} 行现在会遍历我们传入的数组。然后，它会生成包含来自我们数据库的所有数据的 card div。如果我们本地运行此代码并打开地址 http://127.0.0.1:5000/，我们现在应该能看到我们创建的所有任务。
此时，我们也可以将 Web 应用程序部署到我们的 Azure App Service。我们打开左侧的Azure选项卡，展开应用服务选项，选择我们的订阅，右键单击我们的Web 应用，然后单击部署到 Web 应用。部署过程运行一段时间后，我们应该能通过我们的外部 URL 看到相同的任务列表。
查看和更新任务
现在我们的应用程序可以从数据库中读取任务列表，让我们也为我们的查看和编辑任务函数做同样的事情。首先，我们需要像处理 home 方法一样，更新 view_task 和 edit_task 方法的 GET 路由，使用以下代码
@app.route("/viewtask/<string:task_id>", methods=['GET'])
def view_task(task_id):
return render_template(
"viewtask.html", data=database.find_task(task_id)
)
@app.route("/edittask/<string:task_id>", methods=['GET'])
def edit_task(task_id):
return render_template(
"edittask.html", data=database.find_task(task_id)
)
与我们的 home 路由一样，此代码将一个包含来自数据库的任务详细信息的 data 对象传递进来。现在，让我们更新 view_task.html 文件，删除占位符值，并放入数据库数据值
{% extends "layout.html" %}
{% block body %}
<div class="container container-md px-4 py-5">
<h2 class="pb-2 border-bottom">View Task</h2>
<div class="row">
<table class="table table-hover">
<tr><td>Name: </td><td>{{data.name}}</td></tr>
<tr><td>Description: </td><td>{{data.description}}</td></tr>
<tr><td>Due On: </td><td>{{data.due}}</td></tr>
<tr><td>Completed: </td><td>{{data.completed}}</td></tr>
<tr><td>Created On: </td><td>{{data.created}}</td></tr>
<tr><td>Modified On: </td><td>{{data.modified}}</td></tr>
</table>
</div>
</div>
{% endblock %}
We also need to do something similar to our edit screen by updating the edittask.html using the following code:
{% extends "layout.html" %}
{% block body %}
<div class="container container-md px-4 py-5">
<h2 class="pb-2 border-bottom">Edit Task</h2>
<form method="POST" enctype="multipart/form-data">
<div class="mb-3">
<label for="name" class="form-label">Task Name</label>
<input type="text" class="form-control" name="name" id="name" value="{{ data.name }}">
</div>
<div class="mb-3">
<label for="description" class="form-label">Description</label>
<textarea class="form-control" name="description" id="description" rows="5">{{ data.description }}</textarea>
</div>
<div class="mb-3">
<label for="due" class="form-label">Due On</label>
<input type="text" class="form-control" name="due" id="due" value="{{ data.due }}">
</div>
<div class="form-check mb-3">
<input type="checkbox" class="form-check-input" name="completed" id="completed" value="{{ data.completed }}">
<label for="completed" class="form-check-label">Completed</label>
</div>
<div class="container"><p></p></div>
<div class="mb-3">
<div class="row">
<div class="col-md-4">
<label for="created" class="form-label">Created</label>
<input type="text" class="form-control" name="created" id="created" value="{{ data.created }}" disabled>
</div>
<div class="col-md-4">
<label for="modified" class="form-label">Modified</label>
<input type="text" class="form-control" name="modified" id="modified" value="{{ data.modified }}" disabled>
</div>
</div>
</div>
<input class="btn btn-primary" type="submit" value="Update Task" />
<a href="/" class="btn btn-secondary" type="button">Home</a>
</form>
</div>
{% endblock %}
现在运行我们的应用程序，我们会看到各种字段显示来自我们数据库的数据。
我们可以做得更好一些，改进日期和时间字段以及完成字段，但核心功能正在工作，所以让我们继续更新数据库中的数据。
为了处理表单的 POST 响应，我们将进一步扩展我们的 edittask 路由，使用以下代码
@app.route("/edittask/<string:task_id>", methods=['GET', 'POST'])
def edit_task(task_id):
if request.method == 'GET':
return render_template(
"edittask.html", data=database.find_task(task_id)
)
if request.method == 'POST':
data = database.find_task(request.view_args['task_id'])
data['name'] = request.form['name']
data['description'] = request.form['description']
data['due'] = request.form['due']
if 'completed' in request.form.keys():
data['completed'] = True
database.update_task(data)
return view_task(request.view_args['task_id'])
我们的 edit 函数现在已经改变了很多。此路由现在将接受 GET 和 POST 方法。如果收到 GET 方法，该函数将运行之前的 GET 功能。如果代码使用 POST 方法，我们会使用 URL 中的初始任务 ID 从数据库中查找任务。然后，我们使用表单中的数据更新该任务的名称、描述、截止日期和完成状态。
最后，我们调用一个新的函数来使用我们组合的数据更新任务，然后将用户重定向到“查看任务”页面。为了使此操作正常工作，我们还需要在我们的数据库文件中创建 update_task 方法，方法是输入以下代码
def update_task(data):
data['modified'] = datetime.datetime.utcnow()
return mongo_collection_tasks.update({"taskid": data['taskid']}, data)
此方法使用当前日期和时间更新传入的 JSON 文件中的 modified 字段，然后使用 MongoDB update 函数更新我们的文档。
现在运行我们的应用程序并编辑任务，我们应该能够修改我们的任务并将修改保存回我们的数据库。我们还可以将应用程序重新部署到我们的应用服务，并看到相同的结果。
后续步骤
我们现在已经使用很少的 Python 代码编写了一个最小化的任务管理应用程序。该应用程序可以读写 Azure Cosmos DB 服务的数据，并将其部署到 Azure App Service，而无需离开我们的代码编辑器。
您可以通过添加一个“创建新任务”功能来扩展此应用程序，类似于当前的编辑任务页面，它使用我们用来初始化数据库的相同“insert_one”方法。您还可以添加更多数据元素，在表单中构建更好的类型处理，并添加验证来完成项目。
现在您已经了解了使用 Azure 资源快速轻松地构建 Python Web 应用程序的可能性，您可能会受到启发来创建独特的应用程序。注册免费试用，亲身体验在 Azure 上运行和扩展 Python Web 应用程序有多么容易。
要了解有关如何将 Python Web 应用部署到 Linux 上的 App Service 的更多信息，请查看 快速入门：创建 Python 应用。