mirror of https://gitee.com/a529548204/apitest.git
69 lines
2.0 KiB
Python
69 lines
2.0 KiB
Python
# coding:utf-8
|
|
|
|
import warnings
|
|
|
|
warnings.simplefilter('ignore', DeprecationWarning)
|
|
import pymysql
|
|
from scripts.log import Log
|
|
import logging
|
|
from config.confManage import db_manage
|
|
|
|
Log()
|
|
|
|
|
|
class MYSQL(object):
|
|
|
|
def __init__(self):
|
|
self.host = db_manage("${host}$")
|
|
self.user = db_manage("${user}$")
|
|
self.password = db_manage("${password}$")
|
|
self.database = db_manage("${database}$")
|
|
self.charset = db_manage("${charset}$")
|
|
self.port = int(db_manage("${port}$"))
|
|
try:
|
|
logging.debug("正在连接数据库..")
|
|
self.conn = pymysql.connect(host=self.host, user=self.user, password=self.password, database=self.database,
|
|
port=self.port, charset=self.charset)
|
|
self.cursor = self.conn.cursor(cursor=pymysql.cursors.DictCursor)
|
|
logging.debug("数据库连接成功..")
|
|
except Exception as e:
|
|
logging.error("连接数据库失败..{}".format(e))
|
|
raise e
|
|
|
|
def run_sql(self, sql):
|
|
"""
|
|
执行sql语句
|
|
:param sql:
|
|
:return:
|
|
"""
|
|
try:
|
|
logging.debug("准备执行SQL语句..")
|
|
logging.debug("sql语句:{}".format(sql))
|
|
self.cursor.execute(sql)
|
|
rs = self.cursor.fetchall()
|
|
return rs
|
|
except Exception as e:
|
|
logging.error("执行SQL失败..{}".format(e))
|
|
raise
|
|
|
|
def commit(self):
|
|
self.conn.commit()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
DB = MYSQL()
|
|
datas = DB.run_sql(
|
|
"""
|
|
select count(*) count
|
|
from (
|
|
select device.id, device.qr_code, device.region_id, region.enterprise_id, device.status, device.category_id
|
|
from device
|
|
inner join region
|
|
on device.region_id = region.id
|
|
and region.enterprise_id = 88
|
|
)
|
|
as dr
|
|
where status = 3 or status = 1 or status = 2 or status=0 or status IS NULL
|
|
"""
|
|
)
|