python微服务设计

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.

疾风先生

6193人浏览 · 2017-12-21 15:05:27

疾风先生 · 2017-12-21 15:05:27 发布

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))

向您推荐>>Eolink开发者社区

权威｜前沿｜技术｜干货｜国内首个API全生命周期开发者社区

更多推荐

ELK实现containerd的容器日志采集展示【基于logging的全栈监测】

企业级ELK Stack构建介绍

云原生

深入理解 Mocha 测试框架：从零实现一个 Mocha

前言什么是自动化测试自动化测试在很多团队中都是Devops环节中很难执行起来的一个环节，主要原因在于测试代码的编写工作很难抽象，99%的场景都需要和业务强绑定，而且写测试代码的编写工作量往往比编写实际业务代码的工作量更多。在一些很多业务场景中投入产出比很低，适合写自动化测试的应该是那些中长期业务以及一些诸如组件一样的基础库。自动化测试是个比较大的概念，其中分类也比较多，比如单元测试，端对端测试，集

云原生

(20200916 Solved)docker-compose up创建容器自动退出

问题描述如题，创建容器后自动退出了。并且docker start container无效解决方案原因是缺失了控制终端的配置，需要在docker-compose.yml中增加tty:true ，有时候这样也不行，需要再增加一个command:/bin/bash，命令不一定是这个，需要是一个不会退出的命令，然后用-d后台启动容器。Referencesdocker-compose启动容器后自动退出...

云原生

所有评论(0)

查看更多评论

疾风先生

@wind_602

已为社区贡献1条内容