论坛 › Python交流 › 数据库连接问题 pymysql

从入门到富豪 发表于 2020-8-31 14:35:21

数据库连接问题 pymysql

现在在写数据库模板，有一个问题一直困扰，在非商用环境中(也就是不需要用连接池的情况)，我想实现这个功能：当这个链接self.connection如果超过1个小时没有被使用，就关闭这个链接 self.connection.close(), 自己摸索3天无果，特来求高手指点迷津，因为是小白入门不久， （如果能写出详细的代码更佳。） 我的源代码如下：
如果代码有其他的优化空间， 大佬也可以一并指出，万分感谢！

import traceback# 这个模块可以用来打印完整的错误信息
import pymysql


class Connect_DB(object):
    # 调用创建数据库链接函数，创建表函数
    def __init__(self, host, user, password, db, charset):
      self.host = host
      self.user = user
      self.password = password
      self.db = db
      self.charset = charset
      self.connect()# 连接数据库函数得到了self.connection
      self.create_table()# 创建表函数
      print(1)

    # 连接数据库函数
    def connect(self):
      try:
            self.connection = pymysql.connect(host=self.host, user=self.user, password=self.password, db=self.db,charset=self.charset)
      except:
            self.write_log()

    # 创建表函数(如果没有表的话就调用)
    def create_table(self):
      try:
            # 这里通过self.cursor把cursor变成类属性，全局可用
            with self.connection.cursor() as cursor:
                create_table = """
            create table if not exists student (
            id int(8) not null auto_increment primary key,
            name varchar(255) not null,
            chinese float(4) not null default '0' ,
            math float(4) not null default '0',
            english float(4) default '0' not null)
            auto_increment =10000;   
            """
                # 上面的auto_increment =10000是为了设定ID初始值
                cursor.execute(create_table)
                self.connection.commit()
                print('数据库创建表')
      except Exception as e:
            self.write_log()

    # 增删改查函数
    def change_data(self, sql):
      # sql：sql语句
      # 返回：操作的记录条数
      try:
            with self.connection.cursor() as cursor:
                result = cursor.execute(sql)
                self.connection.commit()
            return result
      except:
            self.write_log()
            self.connection.rollback()# 回滚操作

    # 查询数据库函数
    def query(self, sql, fetch_one):
      # sql：sql语句
      # fetch_one:布尔值，判断查询一条还是多条记录。
      # 返回值：查询结果
      try:
            with self.connection.cursor() as cursor:
                cursor.execute(sql)
                if fetch_one:
                  result = cursor.fetchone()
                else:
                  result = cursor.fetchall()
                return result
      except:
            self.write_log()

    # 失败后写日志函数
    def write_log(self):
      with open('connect_db_log.txt', 'a+') as file:
            traceback.print_exc(file=file)# 这个模块可以用来打印完整的错误信息
            file.flush()

    # 删除函数，函数执行完毕触发，这样就不用手动关闭链接了。
    def __del__(self):
      self.connection.close()



if __name__ == '__main__':
    connect = Connect_DB('localhost', 'root', 'masike', 'student_sys', 'utf8')

查看完整版本: 数据库连接问题 pymysql