欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  IT编程

Python连接MYSQL数据库类

程序员文章站 2022-12-08 22:48:24
Python连接MYSQL数据库类 DBHelper.py: import pymysql import logging import sys from mySQL.db_config import db_config # 加入日志 # 获取logger实例 logger = logging.get ......

python连接mysql数据库类

dbhelper.py:

import pymysql
import logging
import sys
from mysql.db_config import db_config

# 加入日志
# 获取logger实例
logger = logging.getlogger("basespider")
# 指定输出格式
formatter = logging.formatter('%(asctime)s\
              %(levelname)-8s:%(message)s')
# 文件日志
file_handler = logging.filehandler("operation_database.log")
file_handler.setformatter(formatter)
# 控制台日志
console_handler = logging.streamhandler(sys.stdout)
console_handler.setformatter(formatter)

# 为logge添加具体的日志处理器
logger.addhandler(file_handler)
logger.addhandler(console_handler)

logger.setlevel(logging.info)


class dbhelper():
    # 构造函数,初始化数据库连接
    def __init__(self,sql,params=none):
        self.sql = sql
        self.params = params
        self.conn = none
        self.cur = none

    def connectiondatabase(self):
        print(db_config['host'],db_config['username'],db_config['password'],db_config['database'],db_config['charset'])
        try:
            self.conn = pymysql.connect(db_config['host'],db_config['username'],
                                    db_config['password'],db_config['database'],charset=db_config['charset'])
        except:
            logger.error("connectdatabase failed")
            return false
        self.cur = self.conn.cursor()
        return true



    # 关闭数据库
    def closedatabase(self):
        # 如果数据打开,则关闭;否则没有操作
        if self.conn and self.cur:
            self.cur.close()
            self.conn.close()
        return true

    # 执行数据库的sq语句,主要用来做插入操作
    def execute(self):
        self.connectiondatabase()
        try:
            if self.conn and self.cur:
                # 正常逻辑,执行sql,提交操作
                self.cur.execute(self.sql,self.params)
                self.conn.commit()
        except:
            logger.error("execute failed: " + self.sql)
            logger.error("params: " + self.params)
            self.closedatabase()
            return false
        return true

    # 用来查询表数据
    def select(self):
        self.connectiondatabase()
        self.cur.execute(self.sql,self.params)
        result = self.cur.fetchall()
        print(result)
        return result

配置信息 db_config.py

 

db_config = {
    'host':'127.0.0.1',
    'port':3306,
    'username':'root',
    'password':'admin',
    'database':'database',
    'charset':'utf8'
}

 测试代码:

from mysql import  dbhelper as mdb


def main():
    # 连接 mysql,获取连接的对象
    sql=r'select * from tusers'
    db=mdb.dbhelper(sql)
    rows=db.select()
    # 依次遍历结果集,发现每个元素,就是表中的一条记录,用一个元组来显示
    for row in rows:
        print(row)


if __name__ == '__main__':
    main()