使用Azure Functions和Graphene操作CosmosDB数据

2 年 ago

新, 韵

5 minutes

首先

我們這次開始在Azure Functions中實施GraphQL API，所以我想將其作為備忘錄整理起來。
上一次，作為Azure Functions的準備工作，我們從Python中準備了與Azure CosmosDB的數據操作。下一步，我們想要在Azure Functions中準備CosmosDB的數據操作，同時使其能夠作為Azure Functions的HTTP請求來接收GraphQL查詢。

本文介绍了对AzureFunctions本地函数的确认（由于在运行时出现了模块导入错误）。AzureFunctions的部署和确认将在另外的时间进行。

关于Azure Functions，官方文档提供了丰富的概述和详细信息。此外，官方文档中有关于“在Azure上创建第一个Python函数”和“面向Azure Functions Python开发者的指南”，提供了配置、实现和教程等内容。可以参考这些文档以加深理解。

关于GraphQL的实现部分，我们使用了一个名为Graphene的库，它在GraphQL的文档中也有介绍。对于GraphQL，我希望能学习术语和规范等方面的知识。

2. 预先准备

该章节涵盖了在本地开发Azure Functions所需的工具安装以及创建函数项目的模板。如果您已经进行了这些步骤，可以跳过本章节。在官方文档中也有相关说明。

2.1. 工具的安装

为了在本地开发Azure Functions，需要先安装.Net Core SDK和Azure Functions Core Tools。在开发环境中，假设使用macOS系统。

请从以下位置下载并安装.NET Core SDK。
.NET Core 2.2 构建应用程序 – SDK

按照官方文件的说明，可以通过brew命令安装Auzre Functions Core Tools。

brew tap azure/functions
brew install azure-functions-core-tools

Azure CLI也可以通过brew命令来安装，按照官方文档的说明进行操作。

brew install azure-cli

2.2. 函数的初始设置

项目名称为MyFunctionProj，并将函数名设定为MyFunction。
根据实际使用环境确定相应的名称并进行创建。

创建虚拟环境并激活。

python3.6 -m venv .env
source .env/bin/activate

创建本地函数项目

func init MyFunctionProj

Select a worker runtime: 
1. dotnet
2. node
3. python (preview)
4. java
5. powershell
Choose option: 3

建立函数

cd MyFunctionProj
func new

Select a template:
1. Blob trigger
2. Cosmos DB trigger
3. Event Grid trigger
4. Event Hub trigger
5. HTTP trigger
6. Queue trigger
7. Service Bus Queue trigger
8. Service Bus Topic trigger
9. Timer trigger

Choose option: 5
Function name: MyFunction

4. 执行本地函数

pip install -r requirements.txt
func host start

创建Azure Functions

假设资源组和存储帐户已创建。
我们要创建 Azure Functions 的 Function App。
（以下命令未部署 Application Insights，如果需要检查日志，请单独部署。）

az login
az functionapp create --resource-group <resource_group_name> --os-type Linux \
--consumption-plan-location westeurope  --runtime python \
--name <azurefunctions_app_name> --storage-account  <storage_name>

6. 将代码部署至Azure Functions。

az login
cd MyFunctionProj
func azure functionapp publish <azurefunctions_app_name> --build-native-deps

如果出现以下内容，则表示部署已完成。

Upload completed successfully.
Deployment completed successfully.

在执行部署命令时，根据包的不同可能需要添加”–build-native-deps”选项进行执行（将会显示以下错误信息，请参考相关信息）。
添加”–build-native-deps”选项会在Docker容器（mcr.microsoft.com/azure-functions/python镜像）内部进行Python代码的构建，因此需要安装Docker环境。

binary dependencies without wheels are not supported.  
Use the --build-native-deps option to automatically build and configure the dependencies using a Docker container. 
More information at https://aka.ms/func-python-publish

3. 函数的实现

3.1. 实施概述

我想要添加之前创建的CosmosDB相关代码和GraphQL代码，实现一个样例，当收到HTTP POST的GraphQL查询时，返回相应的响应。

按照Azure Functions的Python开发指南，将共享代码保存在另一个文件夹中，并改用相对导入的方式来导入模块。

文件夹结构如下所示。

MyFunctionProj
 | - cosmosdb
 | | - __init__.py
 | | - config.py
 | | - cosmosdb.py
 | - graphqllib
 | | - __init__.py
 | | - graphql.py
 | - MyFunction
 | | - __init__.py
 | | - function.json
 | - .funcignore
 | - .gitignore
 | - host.json
 | - requirements.txt

在 requirements.txt 文件中添加以下两个要求。

azure-cosmos
graphene>=2.0

cosmosdb文件夹中存放着用于操作CosmosDB数据的代码，这是上次创建的。
graphql文件夹中存放着定义GraphQL模式解析器以及执行实际处理的代码。我们使用Graphene来实现GraphQL。关于此使用情况，我们将另外整理。

3.2. 运行GraphQL查询

我們在Python中使用Graphene來實現GraphQL，就如上所述。Graphene是針對Python/JavaScript的GraphQL庫。
在實施的過程中，我認為參考官方文件和GitHub也將很有幫助。

在推进GraphQL的实现时，首先要创建模式定义文件（SDL），然后定义解析器，最后在解析器内部实现与数据源的交互处理，按照这个流程进行实现。

在Graphene中，我们将使用Python代码来实现定义架构解析器的过程。与其他语言的GraphQL库（如express-graphql、apollo-server和Prisma）不同之处在于，在Graphene中，我们使用代码而不是架构定义语言来描述GraphQL架构。

对象类型的定义

要定义GraphQL的对象类型，需要准备一个继承自ObjectType的类。
对象内部的字段需要声明字段名和GraphQL类型作为表达式。

#idとmessageというフィールとを持つItem型オブジェクトの定義
class Item(graphene.ObjectType): #GraphQLオブジェクトの定義
    id = graphene.String(required=True) #オブジェクト内部のフィールドの名称と型の宣言
    message = graphene.String()

查询类型的定义

GraphQL的Query类型也会创建一个继承自ObjectType的类。
Query内部的字段（主要是解析器）会声明字段名、参数和返回值的GraphQL类型表达式，然后实现解析器函数（resolve_方法名的方法）。
解析器是一个函数，用于调用数据源或返回某个值（如单个记录、记录列表等），当GraphQL查询调用对应字段名的解析器时，会按照指定的流程执行。
（CosmosDB的数据操作将在解析器内部执行。）

#getItemというリゾルバー関数を持つQuery型の定義
class Query(graphene.ObjectType): #クエリ型オブジェクトの用意
    getItem = graphene.Field(Item) #リゾルバーの関数名、引数・返り値の型の宣言

    #リゾルバー関数getItemと対応づけのうえ、内部処理を実装する
    def resolve_getItem(self, info): 
        return Item(id="1", message="SampleMessage")

查询声明和查询执行

在执行GraphQL查询时，需要在Schema类中将query与Query类型对应起来（即声明根查询类型），这样，在实际执行查询时（调用execute方法），对应的查询解析器函数将被调用并执行处理。

schema = graphene.Schema(query=Query)
schema.execute("GraphQLのクエリ")

GraphQL相关的代码

我这次准备了与GraphQL相关的代码，请参见以下内容。
＃代码已经推送到GitHub，但还在整理中。

import graphene
import json
from datetime import datetime, timezone
from ..cosmosdb.cosmosdb import DatabaseConnection
from ..cosmosdb.cosmosdb import getItem, getReplacedItem
from logging import getLogger
logger = getLogger(__name__)

class DbItem(graphene.ObjectType):
    id = graphene.String(required=True)
    partitionKey = graphene.ID(required=True)
    message = graphene.String()
    addition = graphene.String()
    rid = graphene.String() # comosdb column name _rid
    link = graphene.String() # comosdb column name _self
    etag = graphene.String() # comosdb column name _etag
    attachments = graphene.String() # comosdb column name _attachments
    ts = graphene.Int() # comosdb column name _ts
    datetime = graphene.types.datetime.DateTime()  

    def resolve_rid(self, info):
        return self._rid
    def resolve_link(self, info):
        return self._self
    def resolve_etag(self, info):
        return self._etag
    def resolve_attachments(self, info):
        return self._attachments
    def resolve_ts(self, info):
        return self._ts
    def resolve_datetime(self, info):
        return datetime.fromtimestamp(self._ts, timezone.utc)

class Query(graphene.ObjectType):
    getSampleItem = graphene.Field(DbItem)
    readItem = graphene.Field(DbItem, argument=graphene.String())
    readItems = graphene.List(DbItem)

    def resolve_getSampleItem(self, info):
        item = DbItem(id="1", partitionKey=1,
            message="SampleMessage", addition="SampleAddtionMessage")
        return item

    def resolve_readItem(self, info, argument):
        results = DatabaseConnection().read_item(argument)
        if results.__len__() > 0:
            item = DbItem.__new__(DbItem)
            item.__dict__.update(results[0])
            return item
        else:
            return {}

    def resolve_readItems(self, info):
        results = []
        for item in DatabaseConnection().read_items():
            i = DbItem.__new__(DbItem)
            i.__dict__.update(item)
            results.append(i)
        return results

class GraphQL:
    def __init__(self):
        self.schema = graphene.Schema(
            query=Query,
        )

    def query(self, query):
        logger.info(query)
        results = self.schema.execute(query)
        return json.dumps(results.data)

如果用GraphQL模式定义语言来描述上述模式，会如下所示。

type DbItem {
  id: String!
  partitionKey: ID!
  message: String
  addition: String
  rid: String
  link: String
  etag: String
  attachments: String
  ts: Int
  datetime: DateTime
}

type Query {
  getSampleItem: DbItem
  readItem(argument:String): DbItem
  readItems: [DbItem]     
}

schema {
  query: Query
}

从CosmosDB的数据检索结果中返回的是一个字典类型的数据，但在传递给Graphene时，我会进行一些略微复杂的操作。

item = DbItem.__new__(DbItem)
item.__dict__.update(CosmosDBデータ取得結果[dict型])

由于不能在字段名和解析器中使用(或者更确切地说，不能使用符号)，所以我们选择了一种方法来处理包含要获取的原始数据字段名的情况，即准备一个resolve_XXX函数来解决。

处理HTTP POST请求

Azure Functions处理请求的部分有一个简单的结构。
从请求中获取JSON数据，接收GraphQL查询字符串，将处理投递给GraphQL查询执行部分，然后获取结果。这是一个简单的结构。

import azure.functions as func
from ..graphqllib.graphql import GraphQL, Query
from logging import getLogger
logger = getLogger(__name__)

def main(req: func.HttpRequest) -> func.HttpResponse:
    logger.info('Python HTTP trigger function processed a request.')

    query = req.params.get('query')
    if not query:
        try:
            req_body = req.get_json()
            # query = str(req.get_body().decode(encoding='utf-8'))
            logger.info(req_body)
        except ValueError:
            pass
        else:
            query = req_body.get('query')

    try:
        results = GraphQL().query(query)
    except Exception as e:
        logger.error(e)
        return func.HttpResponse(
             "Internal Server Error",
             status_code=500
        )

    if results:
        return func.HttpResponse(f"{results.data}")
    else:
        return func.HttpResponse(
             "Please pass a name on the query string or in the request body",
             status_code=400
        )

如果代码准备好了，我们将再次部署。

az login
cd MyFunctionProj
func azure functionapp publish <app_name> --build-native-deps

4. 进行 GraphQL 查询的确认执行

我想要使用已实施的函数来发出GraphQL查询并查看响应。
发出查询的方法可以使用curl命令，也可以使用GraphiQL或Insomnia。
如果使用curl命令，您可以使用以下命令来发出查询。

curl -X POST -H "Content-Type: application/json" -d '{"query": "query { readItems { id message etag ts} }"}' http://localhost:7071/api/MyFunction

以下是本次发布的GraphQL查询。对于Insomnia，请指定URL并发出以下GraphQL查询以获取响应。

query { 
  readItems { 
    id
    message
    etag
    ts
  } 
}

所以，回应如下。
使用Graphene执行的GraphQL查询结果似乎以OrderedDict类型返回，但如果不进行任何转换，那么这样的结果不易于阅读，并且在从外部API调用时需要进行各种转换，因此需要进行修正。
→将响应数据从OrderedDict类型转换为JSON似乎可以解决问题。

OrderedDict([('readItems', [OrderedDict([('id', 'id1'), ('message', 'Hello World CosmosDB!'), ('etag', '"00001946-0000-2300-0000-5cc96e910000"'), ('ts', 1556704913)]), OrderedDict([('id', 'id3'), ('message', 'Hello World CosmosDB!'), ('etag', '"00008a47-0000-2300-0000-5cca65bc0000"'), ('ts', 1556768188)]), OrderedDict([('id', 'id4'), ('message', 'Hello World CosmosDB!'), ('etag', '"00008b47-0000-2300-0000-5cca65bc0000"'), ('ts', 1556768188)])])])

如果已经部署了Azure Functions，则URL将变为以下内容：
https://.azurewebsites.net/api/myfunction

经过遵循本文的步骤，会导致500错误。~ 查看ApplicationInsights的日志，似乎出现了导入错误的情况。我打算另外确认一下这个问题。~ 看起来这里发生了Graphene和模块的冲突，所以我已经修正了与graphql相关的代码文件夹名称。

5. 总结

这一次我们在Azure Functions中实现了对CosmosDB数据的操作和GraphQL API。目前，Azure Functions的Python版本仍处于预览阶段。所使用的工具和规范可能会发生变化，可能需要参考文档并进行试错。(在本地函数运行正常，但部署后却无法正常工作也是有可能的…)

首先，要确认Azure Functions部署后的错误，并进行响应数据转换，实现Mutation，添加认证，使代码能够在CI/CD中进行部署，还要使Azure Functions相关环境能够从ARM模板中进行部署。这是否是未来的工作？

请提供相关信息。

Azure Functions 文档
创建首个 Python 函数的 Azure (预览版)
Azure 函数适用于 Python 开发者的指南
操作 Azure Functions Core Tools
GraphQL
Graphene