python微服務設計
- 2020 年 1 月 10 日
- 筆記
Nameko + API Swagger
創建項目
## 安裝微服務框架 pip install nameko==2.5.4.4 ## 安裝api框架 pip install nameko-swagger==1.2.7 ## 創建項目 nameko-admin createproject demo
項目目錄結構
demo/ .tox/ bin/ run.sh conf/ config.yaml logs/ service/ demo.py spec/ v1/ api.yaml test/ ... ##
默認配置
## api.yaml paths: /demo/health: get: summary: 獲取服務健康狀態 tags: - demo operationId: get_health responses: default: description: | 返回結果：{"code": "xxx", "msg": "xxx", "data": {}, "time": xxx, "why": "xxx"} 其中`code`欄位可能返回的錯誤碼包括： * "OK" 操作成功，結果見`data`欄位 * "INTERNAL_SERVER_ERROR" 內部錯誤，具體原因見`why`欄位 schema: $ref: '#/definitions/HealthResponse'
## demo.py @swagger.unmarshal_request @http('GET', '/v1/demo/health') def http_get_health(self, request): return dict(code='OK', msg='', data={}, time=int(time.time()))
配置在路徑path的作用域中
## api.yaml添加path的路徑參數配置 /demo/path/index/{uid}: get: summary: 根據請求參數在path中進行查詢 tags: - demo operationId: index parameters: - name: uid in: path # Note the name is the same as in the path description: 用戶uid required: true type: integer responses: default: description: | 返回結果：{"code": "xxx", "msg": "xxx", "data": {}, "time": xxx, "why": "xxx"} 其中`code`欄位可能返回的錯誤碼包括： * "OK" 操作成功，結果見`data`欄位 * "INTERNAL_SERVER_ERROR" 內部錯誤，具體原因見`why`欄位 schema: $ref: '#/definitions/HealthResponse'
## demo.py @swagger.unmarshal_request @http('GET', '/v1/demo/path/index/<int:uid>') def path_index(self, request, uid): logger.info(request) logger.info(uid) return dict(code='OK', msg='', data={r"msg": "path index success..."}
配置在路徑query作用域中
## api.yaml /demo/query/index: get: summary: 根據請求參數在query中查詢 operationId: query_index parameters: - name: cid in: query description: 用戶頻道Id required: true type: integer responses: default: description: | 返回結果：{"code": "xxx", "msg": "xxx", "data": {}, "time": xxx, "why": "xxx"} 其中`code`欄位可能返回的錯誤碼包括： * "OK" 操作成功，結果見`data`欄位 * "INTERNAL_SERVER_ERROR" 內部錯誤，具體原因見`why`欄位 schema: $ref: '#/definitions/HealthResponse'
## demo.py @swagger.unmarshal_request @http('GET', '/v1/demo/query/index') def query_index(self, request, cid): logger.info(request) logger.info(cid) return dict(code='OK', msg='', data={"cid": cid}, time=int(time.time()))
配置在query中查詢多個
## api.yaml /demo/query/many: get: summary: 根據請求參數在query中查詢 operationId: query_many parameters: - name: cid in: query description: 用戶頻道Id required: true type: integer minimum: 1 - name: uid in: query description: 用戶id required: false type: integer minimum: 1 - name: channelid in: query required: false type: string responses: default: description: | 返回結果：{"code": "xxx", "msg": "xxx", "data": {}, "time": xxx, "why": "xxx"} 其中`code`欄位可能返回的錯誤碼包括： * "OK" 操作成功，結果見`data`欄位 * "INTERNAL_SERVER_ERROR" 內部錯誤，具體原因見`why`欄位 schema: $ref: '#/definitions/HealthResponse'
## demo.py @swagger.unmarshal_request @http('GET', '/v1/demo/query/many') def query_many(self, request, cid, uid=0, channelid=0): logger.info(request) logger.info(cid) logger.info(uid) logger.info(channelid) return dict(code='OK', msg='', data={"cid": cid, "uid": uid, "channelid": channelid}, time=int(time.time()))
在post的表單中查詢
## api.yaml /demo/post: post: summary: 根據請求參數在post的表單中查詢 operationId: do_post parameters: - name: cid in: formData description: 用戶頻道Id required: true type: integer minimum: 1 - name: uid in: formData description: 用戶id required: false type: integer minimum: 1 - name: channelid in: formData required: false type: string responses: default: description: | 返回結果：{"code": "xxx", "msg": "xxx", "data": {}, "time": xxx, "why": "xxx"} 其中`code`欄位可能返回的錯誤碼包括： * "OK" 操作成功，結果見`data`欄位 * "INTERNAL_SERVER_ERROR" 內部錯誤，具體原因見`why`欄位 schema: $ref: '#/definitions/HealthResponse'
## demo.py @swagger.unmarshal_request @http('POST', '/v1/demo/post') def do_post(self, request, cid, uid=0, channelid=""): logger.info(request) logger.info(cid) logger.info(uid) logger.info(channelid) return dict(code='OK', msg='do post message', data={"cid": cid, "uid": uid, "channelid": channelid}, time=int(time.time()*1000))
配置在post中查詢多個
## api.yaml /demo/post/query: post: summary: 根據請求參數在post的表單以及query的路徑中查詢 operationId: query_post parameters: - name: cid in: query description: 用戶頻道Id required: true type: integer minimum: 1 - name: uid in: query description: 用戶id required: false type: integer minimum: 1 - name: channelid in: formData required: false type: string responses: default: description: | 返回結果：{"code": "xxx", "msg": "xxx", "data": {}, "time": xxx, "why": "xxx"} 其中`code`欄位可能返回的錯誤碼包括： * "OK" 操作成功，結果見`data`欄位 * "INTERNAL_SERVER_ERROR" 內部錯誤，具體原因見`why`欄位 schema: $ref: '#/definitions/HealthResponse'
## demo.py @swagger.unmarshal_request @http('POST', '/v1/demo/post/query') def query_post(self, request, cid, uid=0, channelid=""): logger.info(request) logger.info(cid) logger.info(uid) logger.info(channelid) return dict(code='OK', msg='query post message', data={"cid": cid, "uid": uid, "channelid": channelid}, time=int(time.time() * 1000))
配置在body中
## api.yaml /demo/post/body: post: summary: 根據請求參數在body中查詢 operationId: query_body parameters: - in: body name: jsonParameters schema: $ref: '#/definitions/RequestJson' responses: default: description: | 返回結果：{"code": "xxx", "msg": "xxx", "data": {}, "time": xxx, "why": "xxx"} 其中`code`欄位可能返回的錯誤碼包括： * "OK" 操作成功，結果見`data`欄位 * "INTERNAL_SERVER_ERROR" 內部錯誤，具體原因見`why`欄位 schema: $ref: '#/definitions/HealthResponse' ## 定義的requestJson格式 definitions: RequestJson: description: 請求參數在body中的json格式 type: object properties: cid: description: 用戶的cid type: integer uid: description: 用戶的uid type: integer channelid: description: 頻道Id type: string required: - cid
## demo.py @swagger.unmarshal_request @http('POST', '/v1/demo/post/body') def query_body(self, request, jsonParameters): u""" jsonParameters名稱要和api.yaml上聲明的name一致,請求的body的json格式為 { cid:"", uid:"", channelid:"" } ## 使用這種可以傳遞一個對象的json數據而避免使用過長的參數列表來進行request請求 """ return dict(code='OK', msg='query post message', data=jsonParameters, time=int(time.time() * 1000))
