简单封装一下pymysql库

简单封装一下pymysql库

my_pymysql.py

#!/bin/python
# -*- coding:utf-8 -*-
import pymysql
import numpy

def get_connect(**kwargs):
    mysqldb=pymysql.connect(host=kwargs.get('host'),
               user=kwargs.get('user'),
               password=kwargs.get('password'),
               port=kwargs.get('port'),
               database=kwargs.get('database'),
               charset=kwargs.get('charset'))
    return mysqldb

def get_exec_return_list(mysqldb, sql):
    mysqldb=mysqldb
    mysqldb_cur=mysqldb.cursor()
    mysqldb_cur.execute(sql)
    mysqldb.commit()
    mysqldb_data = mysqldb_cur.fetchall()
    #mysqldb_cur.close()
    #mysqldb.close()
    mysqldb_data = numpy.array(mysqldb_data)
    return mysqldb_data
def get_exec_return_list_and_close(mysqldb, sql):
    mysqldb=mysqldb
    mysqldb_cur=mysqldb.cursor()
    mysqldb_cur.execute(sql)
    mysqldb.commit()
    mysqldb_data = mysqldb_cur.fetchall()
    mysqldb_cur.close()
    mysqldb.close()
    mysqldb_data = numpy.array(mysqldb_data)
    return mysqldb_data
def get_exec(mysqldb, sql):
    mysqldb=mysqldb
    mysqldb_cur=mysqldb.cursor()
    mysqldb_cur.execute(sql)
    mysqldb.commit()
    mysqldb_data = mysqldb_cur.fetchall()
    #mysqldb_cur.close()
    #mysqldb.close()
    return mysqldb_data

def get_exec_and_close(mysqldb, sql):
    mysqldb=mysqldb
    mysqldb_cur=mysqldb.cursor()
    mysqldb_cur.execute(sql)
    mysqldb.commit()
    mysqldb_data = mysqldb_cur.fetchall()
    mysqldb_cur.close()
    mysqldb.close()
    return mysqldb_data

def data_format(sql_data):
    pass

__version__ = '1.0'

将my_pymysql.py   放到/home/user/pylib/目录

test.py简单使用页

#!/bin/python

import sys
sys.path.append("/home/user/pylib/")

import my_pymysql
import time


DB_CONFIG = {
    "host": '192.168.10.201',
    "port": 3306,
    "user": 'root',
    "password": 'root__',
    "database": 'testdatabase',
    "charset": 'utf8'
}


sql='''
select * from testtable 
'''
sql1='''
select * from testtable where id = '1'
'''
sql2='''
select * from testtable where id = '2'
'''

def get_mysql_con():
    return my_pymysql.get_connect(host=DB_CONFIG['host'], user=DB_CONFIG['user'], password=DB_CONFIG['password'], database=DB_CONFIG['database'], charset=DB_CONFIG['charset']);

def get_mysql_data():
    #mysqldb = get_mysql_con();
    mysqldb = my_pymysql.get_connect(**DB_CONFIG);
    a = my_pymysql.get_exec(mysqldb, sql)
    return a;


def alarm_data(res):
    strlen = len(res);
    if res :
        print("查询到对应数据，需要告警！");
    else :
        print("无数据，查询状态正常，无需告警！");


res = get_mysql_data()
print(res)
cur_date = time.strftime("%Y-%m-%d",time.localtime())
print(cur_date)
mysqldb = my_pymysql.get_connect(**DB_CONFIG)
print(mysqldb)
print(sql1)
print(sql2)
res1 = my_pymysql.get_exec(mysqldb, sql1)
res2 = my_pymysql.get_exec(mysqldb, sql2)
print(res1)
print(res2)

附：

import pymysql
conn = pymysql.connect(
    host='127.0.0.1',
    user='root',
    password='root__',
    database='sqlmonitor',
    port=3306,
    charset='utf8'
)
sql = 'select * from pay_get_out order by id desc limit 10;'
cur = conn.cursor()
cur.execute(sql)
results = cur.fetchall()
path = results[0][1]
print(path)
cur.close()

/home/daixiongwei/lib/my_sendsms.py

#!/bin/python
# -*- coding:utf-8 -*-
import requests
import json
import datetime
import os
import sys
import logging

#alarm_v5 = "http://10.192.219.13:58080/alarm/v5"
alarm_v5 = "http://10.192.219.12:58080/alarm/v5"

def send_smsv5(token, users, sendChannel, ttsVoice, smsMsg, wechatMsg, emailTitle, emailMsg):
    headers = {'Content-Type': "application/json;charset=utf-8"}
    batch_info = {"token": token,
                  "alertSystem":"自定义监控系统",
                  "businessLine":"业务平台部",
                  "type":2,
                  "ruleType":2,
                  "grade":2,
                  "sendChannel":sendChannel,
                  "userAccounts":users,
                  "ttsVoice":ttsVoice,
                  "smsMsg":smsMsg,
                  "wechatMsg":wechatMsg,
                  "emailTitle":emailTitle,
                  "emailMsg":emailMsg }
    print(batch_info)
    batch_info = json.dumps(batch_info)
    response=requests.post(alarm_v5,data=batch_info,headers=headers)
    print(response.text)

def send_smsv5_1(token, users, sendChannel, **kwargs):
    headers = {'Content-Type': "application/json;charset=utf-8"}
    batch_info = {"token": token,
                  "alertSystem":"自定义监控系统",
                  "businessLine":"业务平台部",
                  "type":2,
                  "ruleType":2,
                  "grade":2,
                  "sendChannel":sendChannel,
                  "userAccounts":users,
                  "ttsVoice":kwargs.get('ttsVoice'),
                  "smsMsg":kwargs.get('smsMsg'),
                  "wechatMsg":kwargs.get('wechatMsg'),
                  "emailTitle":kwargs.get('emailTitle'),
                  "emailMsg":kwargs.get('emailMsg') }
    print(batch_info)
    batch_info = json.dumps(batch_info)
    response=requests.post(alarm_v5,data=batch_info,headers=headers)
    print(response.text)

def send_smsv5_2(**kwargs):
    headers = {'Content-Type': "application/json;charset=utf-8"}
    batch_info = {"token": kwargs.get('token'),
                  "alertSystem":"自定义监控系统",
                  "businessLine":"业务平台部",
                  "type":2,
                  "ruleType":2,
                  "grade":2,
                  "sendChannel":kwargs.get('sendChannel'),
                  "userAccounts":kwargs.get('users'),
                  "ttsVoice":kwargs.get('ttsVoice'),
                  "smsMsg":kwargs.get('smsMsg'),
                  "wechatMsg":kwargs.get('wechatMsg'),
                  "emailTitle":kwargs.get('emailTitle'),
                  "emailMsg":kwargs.get('emailMsg')  }
    print(batch_info)
    batch_info = json.dumps(batch_info)
    response=requests.post(alarm_v5,data=batch_info,headers=headers)
    print(response.text)