快速上手 Pythond 采集器的最佳实践

    try:        self.conn = pymysql.Connect(**self.config)        self.logger.info("pymysql.Connect() ok, {0}".format(id(self.conn)))    except Exception as e:        raise e
def __del__(self):    self.close()
def close(self):    if self.conn:        self.logger.info("conn.close() {0}".format(id(self.conn)))        self.conn.close()

def execute_query(self, sql_str, sql_params=(), first=True):    res_list = None    cur = None    try:        cur = self.conn.cursor()        cur.execute(sql_str, sql_params)        res_list = cur.fetchall()    except Exception as e:        err = str(e)        self.logger.error('execute_query: {0}'.format(err))        if first:            retry = self._deal_with_network_exception(err)            if retry:                return self.execute_query(sql_str, sql_params, False)    finally:        if cur is not None:            cur.close()    return res_list
def execute_write(self, sql_str, sql_params=(), first=True):    cur = None    n = None    err = None    try:        cur = self.conn.cursor()        n = cur.execute(sql_str, sql_params)    except Exception as e:        err = str(e)        self.logger.error('execute_query: {0}'.format(err))        if first:            retry = self._deal_with_network_exception(err)            if retry:                return self.execute_write(sql_str, sql_params, False)    finally:        if cur is not None:            cur.close()    return n, err
def _deal_with_network_exception(self, stre):    errno_str = self._get_errorno_str(stre)    if errno_str != '2006' and errno_str != '2013' and errno_str != '0':        return False    try:        self.conn.ping()    except Exception as e:        return False    return True
def _get_errorno_str(self, stre):    searchObj = self.re_errno.search(stre)    if searchObj:        errno_str = searchObj.group(1)    else:        errno_str = '-1'    return errno_str
def _is_duplicated(self, stre):    errno_str = self._get_errorno_str(stre)    # 1062：字段值重复，入库失败    # 1169：字段值重复，更新记录失败    if errno_str == "1062" or errno_str == "1169":        return True    return False

# if your datakit ip is 127.0.0.1 and port is 9529, you won't need use this,# just comment it.# def __init__(self, **kwargs):#     super().__init__(ip = '127.0.0.1', port = 9529)
def run(self):    config = {        "host": "172.16.2.203",        "port": 30080,        "user": "root",        "password": "Kx2ADer7",        "db": "df_core",        "autocommit": True,        # "cursorclass": pymysql.cursors.DictCursor,        "charset": "utf8mb4"    }
    mysql_conn = MysqlConn(logging.getLogger(''), config)    query_str = "select count(1) from customers where last_logined_time>=(unix_timestamp()-%s)"    sql_params = ('3600')    n = mysql_conn.execute_query(query_str, sql_params)
    data = [      {          "measurement": "hour_logined_customers_count", # 指标名称。          "tags": {            "tag_name": "tag_value", # 自定义 tag，根据自己想要标记的填写，我这里是随便写的          },          "fields": {            "count": n[0][0], # 指标，这里是每个小时登录的用户数          },      },    ]
    in_data = {        'M':data,        'input': "pyfromgit"    }
    return self.report(in_data) # you must call self.report here

# Python 采集器名称name = 'some-python-inputs'  # required
# 运行 Python 采集器所需的环境变量#envs = ['LD_LIBRARY_PATH=/path/to/lib:$LD_LIBRARY_PATH',]
# Python 采集器可执行程序路径(尽可能写绝对路径)cmd = "python3" # required. python3 is recommended.
# 用户脚本的相对路径(填写文件夹，填好后该文件夹下一级目录的模块和 py 文件都将得到应用)dirs = ["hellopythond"] # 这里填的是文件夹名，即类名