手把手带你通过API创建一个loT边缘应用
Posted 华为云开发者社区
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了手把手带你通过API创建一个loT边缘应用相关的知识,希望对你有一定的参考价值。
摘要:使用API Arts&API Explorer调用IoT边缘服务接口创建应用,了解边缘计算在物联网行业的应用。
本文分享自华为云社区《使用API Arts&API Explorer调用IoT边缘服务接口创建应用》,作者:华为IoT云服务。
开始体验前需注册华为云账号并完成实名认证,实验过程中请使用Chrome浏览器完成相关操作。
1、IoT接口介绍(API Arts部分):
1)引言:API Explorer提供了IoT边缘服务,IoT边缘(IoT Edge)是边缘计算在物联网行业的应用。IoT Edge作为物联网边缘“小脑”,在靠近物或数据源头的边缘侧,融合网络、计算、存储、应用核心能力的开放平台,就近提供计算和智能服务,满足行业在实时业务、应用智能、安全与隐私保护等方面的基本需求。
https://support.huaweicloud.com/productdesc-iotedge/iotedge_01_0001.html
我们先通过API Arts介绍一下如何通过API创建一个loT边缘应用。
2) 进入API Arts并登录
https://console.huaweicloud.com/apiarts/?region=cn-north-4#/home
3)新建一个名为“loT边缘”的项目:
3.1)点击首页右上角的创建项目按钮
3.2)在新建弹窗内输入“loT边缘应用”,点击确定
3.3)新建成功之后,自动跳转到项目详情页面,点击“新建API接口”卡片,我们通过设计API的形式介绍创建边缘应用的接口。
4)介绍:创建边缘应用接口是一个使用POST方法进行请求的接口,新建一个POST接口主要有四个部分:接口基本信息、请求头、请求体和返回响应。
4.1)接口基本信息
接口的名称为“创建应用”,路径为“/v2/project_id/edge-apps”。
4.2)请求参数
在路径中输入“/v2/project_id/edge-apps”后,请求参数会自动识别路径中的请求参数。
project_id是调用API接口时需要的一个凭证,可以通过以下方式获得:
https://support.huaweicloud.com/api-iotedge/iotedge_api_0032.html
我们可以向project_id参数增加描述信息:
“项目ID:https://support.huaweicloud.com/api-iotedge/iotedge_api_0032.html ”
4.3)请求体
创建项目的配置项在请求体中发送。
主要有5个string类型的参数:
• edge_app_id:应用ID,必填,最小长度4个字符,最大长度为32
• edge_app_name:应用名称,非必填,最小程度4个字符,最大长度为64
• description:应用描述,非必填,最小长度0,最大长度为255
• function_type:功能类型,非必填,分为数据处理(DATA_PROCESSING)、协议解析(PROTOCOL_PARSING)、IT集成(ON_PREMISE_INTEGRATION)和混合应用(COMPOSITE_APPLICATION),数据默认为DATA_PROCESSING,数据处理模块可以传输消息,协议解析为驱动类型,IT集成为部署南向3rdIA使用,混合应用可同时实现数据处理和协议解析。最小长度为0,最大长度为255
• protocol:驱动协议类型OPCUA|Modbus-TCP,非必填,最小长度0,最大长度为64
将这些内容放进请求体中得到:
4.4)返回响应
新建成功时的返回码为201,参数结构如下所示:
4.5)保存
点击保存按钮后,API Arts会根据设计内容生成文档,在日常开发中可以通过文档进行协作。创建应用的接口的文档如下所示:
5)结言:IoT边缘应用创建接口的主要内容已经介绍完了,可以在API Explorer上开通服务体验到API能力。
二、IoT边缘服务调用(API Explorer部分):
1、调用诗歌生成服务前,需要开通IOT边缘服务:
https://console.huaweicloud.com/iotedge/?region=cn-north-4#/console/authorizationEdge
注:IoT边缘收费说明(有一定的额度,可以支持应用的创建):
https://support.huaweicloud.com/productdesc-iotedge/iotedge_01_0008.html
2、登录API Explorer IOT边缘服务调试界面(需要华为云账号登录):
1)查询应用列表
https://console.huaweicloud.com/apiexplorer/#/openapi/IoTEdge/debug?api=BatchListEdgeApps
2)创建应用,以创建HarmonyOS应用为例:
https://console.huaweicloud.com/apiexplorer/#/openapi/IoTEdge/debug?api=CreateEdgeApp
3)查询应用
https://console.huaweicloud.com/apiexplorer/#/openapi/IoTEdge/debug?api=ShowEdgeApp
4)删除应用:
https://console.huaweicloud.com/apiexplorer/#/openapi/IoTEdge/debug?api=DeleteEdgeApp
5)创建边缘节点:
https://console.huaweicloud.com/apiexplorer/#/openapi/IoTEdge/debug?api=CreateEdgeNode
6)查询边缘节点列表
https://console.huaweicloud.com/apiexplorer/#/openapi/IoTEdge/debug?api=ListEdgeNodes
7)查询边缘节点详情
https://console.huaweicloud.com/apiexplorer/#/openapi/IoTEdge/debug?api=ShowEdgeNode
注:调试参数中输入第6步查询到的ID
8)删除边缘节点
https://console.huaweicloud.com/apiexplorer/#/openapi/IoTEdge/debug?api=DeleteEdgeNode
注:调试参数中输入第6步查询到的ID
三、改造诗歌(CodeArts IDE Online + API插件部分):
打开CodeArts IDE Online,将刚刚响应体复制到IDE中,启动调试,可以将刚刚生成的诗词合成一段mp3音频,双击音频可以打开播放。
手把手带你入门 API 开发
引言
在本文中,您将学习如何使用 Flask、SQLite 3(轻易数据库)和 JSON 创建用于数据通信的 REST API。
本文使用 4 个最常用的 HTTP 动词:GET、POST、PUT 和 DELETE,对应数据库的 CRUD 操作。
比如管理的是一个游戏数据库 games.db
,其中包含名称(name)、价格(price) 和等级(rate)。
我们还将通过使用 Flask 创建的 API 公开几个操作:
- 获取所有游戏
- 创建一个新游戏
- 更新游戏
- 删除游戏
- 通过 ID 获取游戏
首先,我们将使用 Python 创建与数据库相关的 CRUD,然后我们将在 API 中使用 Flask 公开所有这些函数,编码格式为 JSON。
安装 SQLite
- 点击此处,下载你的系统对应的 SQLite 版本,本文以 Windows 为例:
- 下载后将这两个压缩包内的文件 解压到 C 盘的某个目录下:
- 并将该目录添加至环境变量:
- 查看 SQLite3 版本:
λ sqlite3
SQLite version 3.38.5 2022-05-06 15:25:27
Enter ".help" for usage hints.
Connected to a transient in-memory database.
Use ".open FILENAME" to reopen on a persistent database.
sqlite>
创建数据库
使用 sqlite3 databaseName.db
命令来创建一个 SQLite 数据库,本文创建一个 games.db
数据库:
λ sqlite3 games.db
SQLite version 3.38.5 2022-05-06 15:25:27
Enter ".help" for usage hints.
sqlite> .databases
main: C:\\Program Files\\cmder\\games.db r/w
创建表
CREATE TABLE IF NOT EXISTS games(
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
price REAL NOT NULL,
rate INTEGER NOT NULL
)
创建 db.py
我们上个步骤中看到数据库将被称为 games.db
。新建 Python 的 SQLite3 连接文件 db.py
:
import sqlite3
DATABASE_NAME = "games.db"
# 获取数据库连接
def get_db():
conn = sqlite3.connect(DATABASE_NAME)
return conn
# 创建数据库表
def create_tables():
tables = [
"""
CREATE TABLE IF NOT EXISTS games (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
price REAL NOT NULL,
rate INTEGER NOT NULL
)
"""
]
db = get_db()
cursor = db.cursor()
for table in tables:
cursor.execute(table)
中,此外,我们有两个功能:
- 其中之一是
get_db()
:用于获得数据库连接 - 另一个功能
create_tables()
:是当games
数据库表不存在的情况下创建数据库表。
现在我们已经定义了数据库,让我们看看使用 SQLite3 数据库的游戏的 CRUD 操作。
创建 game_controller.py
在 API 中公开数据库之前,我们将创建一个游戏控制器,它将负责保存、更新、删除和获取游戏数据的所有操作。
所有这些函数都在一个名为 game_controller.py
的文件中,它看起来像这样:
from db import get_db
def insert_game(name, price, rate):
db = get_db()
cursor = db.cursor()
statement = "INSERT INTO games(name, price, rate) VALUES (?, ?, ?)"
cursor.execute(statement, [name, price, rate])
db.commit()
return True
def update_game(id, name, price, rate):
db = get_db()
cursor = db.cursor()
statement = "UPDATE games SET name = ?, price = ?, rate = ? WHERE id = ?"
cursor.execute(statement, [name, price, rate, id])
db.commit()
return True
def delete_game(id):
db = get_db()
cursor = db.cursor()
statement = "DELETE FROM games WHERE id = ?"
cursor.execute(statement, [id])
db.commit()
return True
def get_by_id(id):
db = get_db()
cursor = db.cursor()
statement = "SELECT id, name, price, rate FROM games WHERE id = ?"
cursor.execute(statement, [id])
return cursor.fetchone()
def get_games():
db = get_db()
cursor = db.cursor()
query = "SELECT id, name, price, rate FROM games"
cursor.execute(query)
return cursor.fetchall()
在文件中,我们看到了几个函数。 insert_game
函数接收游戏数据并将其插入数据库(INSERT);所有这些都使用准备好的语句来避免我们使用 Python 和 Flask 创建的这个 API 中的 SQL 注入。
我们还看到其他方法,例如 update_game
执行 UPDATE 操作以更新游戏,delete_game
从其 id 删除游戏 (DELETE),get_by_id
从其 id 返回游戏(使用 SELECT 操作)。
最后我们看看返回所有现有游戏的 get_games
函数。
请注意,所有函数都使用数据库和游标来执行所有操作。 现在我们有了数据库操作的 CRUD,是时候用 Flask 公开 API 中的所有内容了
创建 main.py
安装 flask
可以查看官方安装帮助:
$ pip install Flask
我们在 API 中做的第一件事是创建 Flask 应用程序并导入游戏控制器。我们还从数据库中导入了一个函数,因为我们需要在启动应用程序时创建表:
from flask import Flask, jsonify, request
import game_controller
from db import create_tables
app = Flask(__name__)
然后使用 GET、PUT、POST 和 DELETE http 动词定义路由:
@app.route(/games, methods=["GET"])
def get_games():
games = game_controller.get_games()
return jsonify(games)
@app.route("/game", methods=["POST"])
def insert_game():
game_details = request.get_json()
name = game_details["name"]
price = game_details["price"]
rate = game_details["rate"]
result = game_controller.insert_game(name, price, rate)
return jsonify(result)
@app.route("/game", methods=["PUT"])
def update_game():
game_details = request.get_json()
id = game_details["id"]
name = game_details["name"]
price = game_details["price"]
rate = game_details["rate"]
result = game_controller.update_game(id, name, price, rate)
return jsonify(result)
@app.route("/game/<id>", methods=["DELETE"])
def delete_game(id):
result = game_controller.delete_game(id)
return jsonify(result)
@app.route("/game/<id>", methods=["GET"])
def get_game_by_id(id):
game = game_controller.get_by_id(id)
return jsonify(game)
每条路由都对应了我们之前创建的游戏控制器功能,从而与 SQLite3 数据库进行交互。
- 在更新和插入游戏时,我们使用
get_json
阅读了请求的 JSON ,然后访问字典:
game_details = request.get_json()
- 在删除或通过 ID 获取的情况下,我们从路由中读取 id 变量作为 并在方法中接收它:
game = game_controller.get_by_id(id)
- 此 Python API 通过 JSON 进行通信,因此所有响应都是根据 jsonify 函数返回的内容进行的:
return jsonify(result)
- 最后,我们创建 Flask 应用程序来启动服务器并监听请求:
if __name__ == "__main__":
create_tables()
"""
Here you can change debug and port
Remember that, in order to make this API functional, you must set debug in False
"""
app.run(host=0.0.0.0, port=8000, debug=False)
至此,我们完整的 main.py
函数如下:
from flask import Flask, jsonify, request, json
import game_controller
from db import create_tables
app = Flask(__name__)
@app.route(/games, methods=["GET"])
def get_games():
games = game_controller.get_games()
return jsonify(games)
@app.route("/game", methods=["POST"])
def insert_game():
game_details = json.loads(request.get_data())
name = game_details["name"]
price = game_details["price"]
rate = game_details["rate"]
result = game_controller.insert_game(name, price, rate)
return jsonify(result)
@app.route("/game", methods=["PUT"])
def update_game():
game_details = request.get_json()
id = game_details["id"]
name = game_details["name"]
price = game_details["price"]
rate = game_details["rate"]
result = game_controller.update_game(id, name, price, rate)
return jsonify(result)
@app.route("/game/<id>", methods=["DELETE"])
def delete_game(id):
result = game_controller.delete_game(id)
return jsonify(result)
@app.route("/game/<id>", methods=["GET"])
def get_game_by_id(id):
game = game_controller.get_by_id(id)
return jsonify(game)
"""
Enable CORS. Disable it if you dont need CORS
"""
@app.after_request
def after_request(response):
response.headers["Access-Control-Allow-Origin"] = "*" # <- You can change "*" for a domain for example "http://localhost"
response.headers["Access-Control-Allow-Credentials"] = "true"
response.headers["Access-Control-Allow-Methods"] = "POST, GET, OPTIONS, PUT, DELETE"
response.headers["Access-Control-Allow-Headers"] = "Accept, Content-Type, Content-Length, Accept-Encoding, X-CSRF-Token, Authorization"
return response
if __name__ == "__main__":
create_tables()
"""
Here you can change debug and port
Remember that, in order to make this API functional, you must set debug in False
"""
app.run(host=0.0.0.0, port=8000, debug=False)
值得注意的是,这里我们添加 CORS,如果您要从与 API 本身不同的域使用此 API,则需要启用 CORS。只需将以下代码片段添加到 API(在存储库中,您会发现已添加的代码,您可以根据需要将其删除):
"""
Enable CORS. Disable it if you dont need CORS
"""
@app.after_request
def after_request(response):
response.headers["Access-Control-Allow-Origin"] = "*" # <- You can change "*" for a domain for example "http://localhost"
response.headers["Access-Control-Allow-Credentials"] = "true"
response.headers["Access-Control-Allow-Methods"] = "POST, GET, OPTIONS, PUT, DELETE"
response.headers["Access-Control-Allow-Headers"] = "Accept, Content-Type, Content-Length, Accept-Encoding, X-CSRF-Token, Authorization"
return response
启动程序并测试
使用 python3 main.py
,运行我们的服务器和 API :
* Serving Flask app main (lazy loading)
* Environment: production
WARNING: This is a development server. Do not use it in a production deployment.
Use a production WSGI server instead.
* Debug mode: off
* Running on all addresses (0.0.0.0)
WARNING: This is a development server. Do not use it in a production deployment.
* Running on http://127.0.0.1:8000
* Running on http://10.26.4.188:8000 (Press CTRL+C to quit)
为了方便本文创建的所有 API 的测试,使用到 Apifox 作为接口测试工具:
GET 获取所有游戏测试
访问 http://127.0.0.1:8000/games
。返回 200 OK,但是目前数据库中没有游戏,所以返回空 JSON 字符串:
POST 新增一个游戏测试
因为游戏 id 是自增的,所以只需要往外面的服务器传入 name, price 和 rate:
服务器后台显示如下:
127.0.0.1 - - [16/Jun/2022 14:51:05] "POST /game HTTP/1.1" 200 -
PUT 修改游戏内容测试
查看我们当前的所有的游戏内容:
然后将 LOL 的名字改为 英雄联盟, 此时我们需要传入四个参数:id,name, price, rate,因为要根据 id 来找到我们想要更改的那个游戏:
"id": 2,
"name": "英雄联盟",
"price": 1,
"rate": 99
测试成功如下:
GET 获取单个游戏内容测试
为了检查刚刚针对 id 为 2 的游戏是否更名成功,我们可以传入http://127.0.0.1:8000/game/2
路径:
可以看到,PUT 和当前的 GET 请求都是 OK 的。
DELETE 删除游戏测试
最后,就来到我们的删除环节,把 id 为 3 的游戏内容删除:
可以看到删除成功,我们的 API 测试是完全 OK 的
总结
最后,又到了本文需要做总结的时候了。
本篇教程利用 Python、SQLite3、Flask 实现了一个简单的**游戏 Rest API **功能,通过在 Flask 中处理传入的 HTTP 的四大请求:GET、POST、PUT、DELETE,实现了最基本的增删改查功能。
最后利用 Apifox 作为我们的接口测试工具,完整的体验了一个简易 API 开发的流程。
希望本文能对你有所帮助,如果喜欢本文,可以点个关注.下一篇文章见!
宇宙古今无有穷期,一生不过须臾,当思奋争。
以上是关于手把手带你通过API创建一个loT边缘应用的主要内容,如果未能解决你的问题,请参考以下文章