在处理文件或数据库的时候,程序遵循的是 打开(文件/连接) -> 处理 -> 释放 的完整流程,而且处理过程中还有可能出现一些意外导致程序崩溃,若无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 os
import sqlite3
class 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 = None
self.cursor = None
def __enter__(self):
return self
def __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 True
except Exception as e:
self.logger.error("执行sql语句 %s 的时候出错,错误信息:%s " % (sql, str(e)))
return False
def init_sqlite(self):
""" 创建数据库连接 """
try:
self.conn = sqlite3.connect(self.db_name)
self.cursor = self.conn.cursor()
return True
except Exception as e:
self.logger.error("初始化数据库 %s 的时候失败,错误信息:%s" % (self.db_name, str(e)))
return False
def 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 = None
self.conn = None
def 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 row
if __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