在处理文件或数据库的时候,程序遵循的是 打开(文件/连接) -> 处理 -> 释放 的完整流程,而且处理过程中还有可能出现一些意外导致程序崩溃,若无try…except…finally则可能导致一些连接无法关闭。
在python中有一个很有效的语句,即with…as,可以一步到位的做到打开+关闭的操作,针对文件的处理的简单操作如下:
with open("test.txt", r) as fh:line = fh.readlines()
上述操作等同于
fh = open("test.txt", r)lines = fh.readlines()fh.close()
当然,with…as语句也可以进行扩展,如下,使用with…as进行数据库文件的操作。
在python中实现了enter和exit方法,即支持上下文管理器协议。上下文管理器就是支持上下文管理器协议的对象,它是为了with而生。当with语句在开始运行时,会在上下文管理器对象上调用 enter 方法。with语句运行结束后,会在上下文管理器对象上调用 exit 方法。
# -*- coding:utf-8 -*-import osimport sqlite3class ParameterProcess(object):def __init__(self, _logger):self.logger = _logger_parent_path = os.path.abspath(os.path.join(os.path.abspath(__file__), os.pardir, os.pardir))_path = os.path.join(_parent_path, "Configuration")self.db_name = os.path.join(_path, "polygons.db")self.conn = Noneself.cursor = Nonedef __enter__(self):return selfdef __exit__(self, e_t, e_v, t_b):self.close_sqlite()def delete_file(self):""" 删除已存在的数据库文件 """if os.path.isfile(self.db_name):os.remove(self.db_name)def create_table(self):""" 创建数据表 """sql = "CREATE TABLE POLYGONS( " \"ID INTEGER, " \"CATEGORY VARCHAR(20), " \"AREA DOUBLE(12,15)," \"POINTS VARCHAR(8));"if not self.is_table_exists("POLYGONS"):self.cursor.execute(sql)def is_table_exists(self, table):""" 查看某个表是否存在:param table: str:return: Bool"""sql = "select count(*) as cnt from sqlite_master where type='table' and name = '%s'" % table.upper()self.logger.info(sql)try:rec = self.cursor.execute(sql)for row in rec:if row[0] == 1:return Trueexcept Exception as e:self.logger.error("执行sql语句 %s 的时候出错,错误信息:%s " % (sql, str(e)))return Falsedef init_sqlite(self):""" 创建数据库连接 """try:self.conn = sqlite3.connect(self.db_name)self.cursor = self.conn.cursor()return Trueexcept Exception as e:self.logger.error("初始化数据库 %s 的时候失败,错误信息:%s" % (self.db_name, str(e)))return Falsedef close_sqlite(self):""" 关闭数据库连接 """try:self.cursor.close()self.conn.close()self.logger.info("断开数据库连接!")except Exception as e:self.logger.error("关闭数据库 %s 的时候出错,错误信息:%s" % (self.db_name, str(e)))finally:self.cursor = Noneself.conn = Nonedef insert_single_data(self, data):""" 向数据库内插入单条数据:param data: list"""sql = "insert into polygons values(%s);" % (','.join(['?'] * len(data)))self.cursor.execute(sql, data)self.conn.commit()def select_data(self, sql):""" 执行查询语句:param sql: str:return: tuple"""rec = self.cursor.execute(sql)for row in rec:yield rowif __name__ == '__main__':from Common.logger import logger_logger = logger("ParameterProcess")with ParameterProcess(_logger) as para_process:para_process.init_sqlite()para_process.create_table()
引用:
[1]. https://www.toutiao.com/a6781415731801096715/?tt_from=copy_link&utm_campaign=client_share×tamp=1578969549&app=news_article&utm_source=copy_link&utm_medium=toutiao_ios&req_id=202001141039090100160230160C01611C&group_id=6781415731801096715
[2]. https://www.cnblogs.com/flashBoxer/p/9664813.html
