pecan API调用

Posted 2020-12-29 李成石

1.在PasteDeploy部署caesar-api服务时，setup_app中建立app，app中设置hooks，hooks负责加载conf配置文件和数据库连接 self.storage_connection。

从而在请求访问时可以调用对应钩子对象的方法。

def setup_app(pecan_config=None, extra_hooks=None):
    # FIXME: Replace DBHook with a hooks.TransactionHook
    app_hooks = [hooks.ConfigHook(CONF),
                 hooks.DBHook(CONF)]
    # 配置文件钩子ConfigHook和数据库连接DBHook钩子。
    if extra_hooks:
        app_hooks.extend(extra_hooks)

    if not pecan_config:
        pecan_config = get_pecan_config()

    pecan.configuration.set_config(dict(pecan_config), overwrite=True)

    # NOTE(sileht): pecan debug won\'t work in multi-process environment
    pecan_debug = CONF.api.pecan_debug
    if CONF.api.workers and CONF.api.workers != 1 and pecan_debug:
        pecan_debug = False

    app = pecan.make_app(
        pecan_config.app.root,
        debug=pecan_debug,
        force_canonical=getattr(pecan_config.app, \'force_canonical\', True),
        hooks=app_hooks, #添加请求钩子
        wrap_app=middleware.ParsableErrorMiddleware,
        guess_content_type_from_ext=False
    )

    return app　　

 在DBHook中，before方法RoutingState中添加数据库连接属性storage_conn：从而在pecan.requst.storage_conn获取数据库连接

class DBHook(hooks.PecanHook):

    def __init__(self, conf):
        self.storage_connection = self.get_connection(conf)

        if not self.storage_connection:
            raise Exception(
                "API failed to start. Failed to connect to database")

    def before(self, state):
        # 添加数据库连接
        state.request.storage_conn = self.storage_connection

    @staticmethod
    def get_connection(conf):
        try:
            return storage.get_connection_from_config(conf)
        except Exception as err:
            LOG.exception("Failed to connect to db" "retry later: %s",
                          err)        

通过debug：

              在请求时无法使用pdb进行debug断点测试，需要在代码中设置import pdb;pdb.set_trace()同时，使用gdb attach 服务进程，然后再发送请求时，才会自动到断点位置。

             

2.请求访问（pecan对象路由框架）

 在根目录root.py中定义v2对象，路由到V2Controller

class RootController(object):
    def __init__(self):
        self.v2 = v2.V2Controller()

在V2Controller中定义_look_up方法，对/v2/后的字段判断，如果为meters则路由到 meters.MetersController()对象

class V2Controller(rest.RestController):

    @pecan.expose()
    def _lookup(self, kind, *remainder):
        if kind == \'meters\':
            return meters.MetersController(), remainder

meters.MetersController()又路由到 MeterController(meter_name)，其中的meter_name作为meters后的另一个参数。如果懂得springmvc的controller的requestMapping中路径映射，其类似于

@RequestMapping(value = "/xxx/{xxx}",method=RequestMethod.GET)中{xxx}动态获取请求中的变量

class MetersController(rest.RestController):
    """Works on meters."""

    @pecan.expose()
    def _lookup(self, meter_name, *remainder):
        return MeterController(meter_name), remainder

在 MeterController(meter_name)中 GET请求被路由到get_all方法中，@wsme_pecan.wsexpose([OldSample], [base.Query], int)对请求参数和相应返回进行转化，返回值为list的OldSample对象，第一个参数为list的base.Query对象可以为None，limit 是int类型，可为None

class MeterController(rest.RestController):
    """Manages operations on a single meter."""
    _custom_actions = {
        \'statistics\': [\'GET\'],
    }

    def __init__(self, meter_name):
        pecan.request.context[\'meter_name\'] = meter_name
        self.meter_name = meter_name

    @wsme_pecan.wsexpose([OldSample], [base.Query], int)
    def get_all(self, q=None, limit=None):
        """Return samples for the meter.

        :param q: Filter rules for the data to be returned.
        :param limit: Maximum number of samples to return.
        """

        q = q or []
        limit = v2_utils.enforce_limit(limit)
        kwargs = v2_utils.query_to_kwargs(q, storage.SampleFilter.__init__)
        kwargs[\'meter\'] = self.meter_name
        f = storage.SampleFilter(**kwargs)
        return [OldSample.from_db_model(e)
                for e in pecan.request.storage_conn.get_samples(f, limit=limit) # 数据库连接对象，调用其get_samples获取数据库数据
                ]

最后调用接口，查询数据库数据：

1.启动caesar-api服务：

2.调用/v2/meters接口获取数据

以上只是个人感悟，具体可参考我的github: https://github.com/CaesarLinsa/ceilometer_TestCase