江苏龙网

  1. 主页 > 生活百科 > >

基于 Serverless Framework 的人工智能小程序开发( 四 )

Serverless F

公共组件的编写

  • 如果这个函数要作为小程序的一个接口，那么就要接 APIGW，那么如何才能实现本地测试呢？每次都发到线上配置 APIGW 触发器才能测试，这显然是不现实的 。
  • 这个函数需要数据库的连接，需要获取用户的信息等，难道别的函数不需要吗？如果需要，那么是否每个函数都要重复写这部分代码？或者说，代码的复用应该如何处理呢？是否可以提取公共组件呢？
所以，我将这个函数进行了规范化和完整化：
?复制代码
# -*- coding: utf8 -*- import json try:import returnCommonfrom mysqlCommon import mysqlCommonexcept:import common.testCommoncommon.testCommon.setEnv()import common.returnCommon as returnCommonfrom common.mysqlCommon import mysqlCommonmysql = mysqlCommon()def main_handler(event, context):try:print(event)body = json.loads(event['body'])wecaht = body['wechat']nickname = body['nickname']remark = str(body['remark'])if not wecaht:return returnCommon.return_msg(True, "请使用微信小程序登陆本页面 。")if not mysql.getUserInfor(wecaht):if not nickname:return returnCommon.return_msg(True, "参数异常，请重试 。")if mysql.addUserInfor(wecaht, nickname, remark):return returnCommon.return_msg(False, "注册成功")return returnCommon.return_msg(True, "注册失败，请重试 。")return returnCommon.return_msg(False, "登录成功")except Exception as e:print(e)return returnCommon.return_msg(True, "用户信息异常，请联系管理员处理") def test():event = {"requestContext": {"serviceId": "service-f94sy04v","path": "/test/{path}","httpMethod": "POST","requestId": "c6af9ac6-7b61-11e6-9a41-93e8deadbeef","identity": {"secretId": "abdcdxxxxxxxsdfs"},"sourceIp": "14.17.22.34","stage": "release"},"headers": {"Accept-Language": "en-US,en,cn","Accept": "text/html,application/xml,application/json","Host": "service-3ei3tii4-251000691.ap-guangzhou.apigateway.myqloud.com","User-Agent": "User Agent String"},"body": json.dumps({"wechat": "12345","nickname": "test","remark": "",}),"pathParameters": {"path": "value"},"queryStringParameters": {"foo": "bar"},"headerParameters": {"Refer": "10.0.2.14"},"stageVariables": {"stage": "release"},"path": "/test/value","queryString": {"foo": "bar","bob": "alice"},"httpMethod": "POST"}print(main_handler(event, None))if __name__ == "__main__":test()数据库等一些公共组件，统一放在common目录下，例如mysqlCommon.py(部分):
?复制代码
# -*- coding: utf8 -*- import osimport randomimport pymysqlimport datetime try:import cosClientexcept:import common.cosClient as cosClientclass mysqlCommon:def __init__(self):self.getConnection({"host": os.environ.get('mysql_host'),"user": os.environ.get('mysql_user'),"port": int(os.environ.get('mysql_port')),"db": os.environ.get('mysql_db'),"password": os.environ.get('mysql_password')})def getConnection(self, conf):self.connection = pymysql.connect(host=conf['host'],user=conf['user'],password=conf['password'],port=int(conf['port']),db=conf['db'],charset='utf8',cursorclass=pymysql.cursors.DictCursor,autocommit=1)def doAction(self, stmt, data):try:self.connection.ping(reconnect=True)cursor = self.connection.cursor()cursor.execute(stmt, data)result = cursorcursor.close()return resultexcept Exception as e:print(e)try:cursor.close()except:passreturn Falsedef addUserInfor(self, wecaht, nickname, remark):insert_stmt = ("INSERT INTO users(wechat, nickname, remark) ""VALUES (%s,%s,%s)")data = https://www.isolves.com/it/ai/2020-07-24/(wecaht, nickname, remark)result = self.doAction(insert_stmt, data)return False if result == False else True这样做的好处是：
  • 将数据库提取出一个公共组件，便于维护
  • 在 login 函数中，根据不同的时期（本地开发和线上）可以导入不同的模块
便于开发与测试的方法由于云函数的测试非常不友好，所以为了在编写代码时可以更快地模拟线上环境，我选择通过增加test()方法来模拟触发器情况，进行简单的测试 。
?复制代码
【基于 Serverless Framework 的人工智能小程序开发】try:import cosClientexcept:import common.cosClient as cosClient


推荐阅读


上一篇：罗布泊探险失踪的人 罗布泊失踪的人

下一篇：从病毒到“基础软件污染”，Linux 真的安全吗？