Enjoy the good life everyday!
关闭
欢迎来PyGo个人空间 ^_^
Python模块之watcher打点 | PyGo²

Python模块之watcher打点

Python Python模块系列
Python模块系列

Python项目常用的模块代码。

版本

名称 版本
Python 3

描述

用于API计时打点,写入request表。

设计

装饰器。

使用

装饰器的方式使用。

1
2
3
4
from flask import request

@watcher(watcher_args=request)
watcher_args参数为request对象,里面包含blueprint, endpoint, method, path, header等请求参数

源码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
# -*- coding: utf-8 -*-

"""
------------------------------------------------

describe:
watcher
用于API计时打点,写入request表

base_info:
__author__ = "PyGo"
__time__ = "2022/8/26 23:02"
__version__ = "v.1.0.0"
__mail__ = "gaoming971366@163.com"
__blog__ = "www.pygo2.top"
__project__ = "open2lisapi"

usage:
from flask import request
@watcher(watcher_args=request)
watcher_args参数为request对象,里面包含blueprint, endpoint, method, path, header等请求参数

design:

reference urls:

python version:
python3


Enjoy the good life everyday!!!
Life is short, I use python.

------------------------------------------------
"""

# ------------------------------------------------------------
# usage: /usr/bin/python watcher.py
# ------------------------------------------------------------
from datetime import datetime
from functools import wraps
from pprint import pprint

from deploy.utils.logger import logger as LOG
from deploy.utils.utils import get_rtx_id
from deploy.services.request import RequestService


# 声明一个全局RequestService对象
request_service = RequestService()
GLOBAL_NEW_ENDPOINR = [
'dashboard.pan',
'dashboard.pan_chart',
'dashboard.index',
'auth.user_list',
'info.dict_list'
]


# method desc: API request to write database table [request]
def __add_request(request, cost, rtx=None):
"""
API request information to insert into request table
:param request: API request object parameters
:param cost: API run time, unit is second
:param rtx: request user rtx-id
- 1.request get "X-Rtx-Id"
- 2.no request by manual set
"""
rtx_id = get_rtx_id(request) or rtx # not rtx_id, no insert
if not rtx_id:
return False
method = getattr(request, 'method') # method allow only get or post
if method and str(method).upper() not in ['GET', 'POST']:
return False
if hasattr(request, 'endpoint') and getattr(request, 'endpoint') in GLOBAL_NEW_ENDPOINR:
RequestService().add_request(request=request, cost=cost, rtx=rtx_id)
else:
request_service.add_request(request=request, cost=cost, rtx=rtx_id)


# >>>>>>>>>>>>>>>>>>>>>>>>>>>>> API打点计时器 <<<<<<<<<<<<<<<<<<<<<<<<<<<<<
def watcher(watcher_args):
def _watcher(fn):
@wraps(fn)
def _wrapper(*args, **kwargs):
start = datetime.now()
res = fn(*args, **kwargs)
end = datetime.now()
cost = round((end-start).microseconds * pow(0.1, 6), 4) # API run time, unit is second
if watcher_args:
__add_request(request=watcher_args, cost=cost) # API request to write database table [request]
LOG.info('@Watcher [%s] is run: %s' % (getattr(watcher_args, 'endpoint') or fn.__name__, cost))
return res

return _wrapper
return _watcher

系列

Python模块之command系统命令
Python模块之excel模块
Python模块之logger日志
Python模块之utils公共方法
Python模块之watcher打点
Python模块之config配置解析
Python模块之dtalk钉钉消息
Python模块之企业微信

更多模块请参考文章TAG进行查看。

Python模块系列,持续更新中。。。。。。
  • 本文作者:mingliang.gao【一个爱老婆Python程序猿。。。。。。】
  • 本文链接: http://pygo2.top/articles/31232/
  • 版权声明: 本博客所有文章欢迎转载,转载请注明出处!
觉得有帮助 请偶坐个公交车
0%