在处理文件或数据库的时候,程序遵循的是 打开(文件/连接) -> 处理 -> 释放 的完整流程,而且处理过程中还有可能出现一些意外导致程序崩溃,若无try…except…finally则可能导致一些连接无法关闭。

    在python中有一个很有效的语句,即with…as,可以一步到位的做到打开+关闭的操作,针对文件的处理的简单操作如下:

    1. with open("test.txt", r) as fh:
    2. line = fh.readlines()

    上述操作等同于

    1. fh = open("test.txt", r)
    2. lines = fh.readlines()
    3. fh.close()

    当然,with…as语句也可以进行扩展,如下,使用with…as进行数据库文件的操作。

    在python中实现了enterexit方法,即支持上下文管理器协议。上下文管理器就是支持上下文管理器协议的对象,它是为了with而生。当with语句在开始运行时,会在上下文管理器对象上调用 enter 方法。with语句运行结束后,会在上下文管理器对象上调用 exit 方法。

    1. # -*- coding:utf-8 -*-
    2. import os
    3. import sqlite3
    4. class ParameterProcess(object):
    5. def __init__(self, _logger):
    6. self.logger = _logger
    7. _parent_path = os.path.abspath(os.path.join(os.path.abspath(__file__), os.pardir, os.pardir))
    8. _path = os.path.join(_parent_path, "Configuration")
    9. self.db_name = os.path.join(_path, "polygons.db")
    10. self.conn = None
    11. self.cursor = None
    12. def __enter__(self):
    13. return self
    14. def __exit__(self, e_t, e_v, t_b):
    15. self.close_sqlite()
    16. def delete_file(self):
    17. """ 删除已存在的数据库文件 """
    18. if os.path.isfile(self.db_name):
    19. os.remove(self.db_name)
    20. def create_table(self):
    21. """ 创建数据表 """
    22. sql = "CREATE TABLE POLYGONS( " \
    23. "ID INTEGER, " \
    24. "CATEGORY VARCHAR(20), " \
    25. "AREA DOUBLE(12,15)," \
    26. "POINTS VARCHAR(8));"
    27. if not self.is_table_exists("POLYGONS"):
    28. self.cursor.execute(sql)
    29. def is_table_exists(self, table):
    30. """ 查看某个表是否存在
    31. :param table: str
    32. :return: Bool
    33. """
    34. sql = "select count(*) as cnt from sqlite_master where type='table' and name = '%s'" % table.upper()
    35. self.logger.info(sql)
    36. try:
    37. rec = self.cursor.execute(sql)
    38. for row in rec:
    39. if row[0] == 1:
    40. return True
    41. except Exception as e:
    42. self.logger.error("执行sql语句 %s 的时候出错,错误信息:%s " % (sql, str(e)))
    43. return False
    44. def init_sqlite(self):
    45. """ 创建数据库连接 """
    46. try:
    47. self.conn = sqlite3.connect(self.db_name)
    48. self.cursor = self.conn.cursor()
    49. return True
    50. except Exception as e:
    51. self.logger.error("初始化数据库 %s 的时候失败,错误信息:%s" % (self.db_name, str(e)))
    52. return False
    53. def close_sqlite(self):
    54. """ 关闭数据库连接 """
    55. try:
    56. self.cursor.close()
    57. self.conn.close()
    58. self.logger.info("断开数据库连接!")
    59. except Exception as e:
    60. self.logger.error("关闭数据库 %s 的时候出错,错误信息:%s" % (self.db_name, str(e)))
    61. finally:
    62. self.cursor = None
    63. self.conn = None
    64. def insert_single_data(self, data):
    65. """ 向数据库内插入单条数据
    66. :param data: list
    67. """
    68. sql = "insert into polygons values(%s);" % (','.join(['?'] * len(data)))
    69. self.cursor.execute(sql, data)
    70. self.conn.commit()
    71. def select_data(self, sql):
    72. """ 执行查询语句
    73. :param sql: str
    74. :return: tuple
    75. """
    76. rec = self.cursor.execute(sql)
    77. for row in rec:
    78. yield row
    79. if __name__ == '__main__':
    80. from Common.logger import logger
    81. _logger = logger("ParameterProcess")
    82. with ParameterProcess(_logger) as para_process:
    83. para_process.init_sqlite()
    84. para_process.create_table()

    引用:
    [1]. https://www.toutiao.com/a6781415731801096715/?tt_from=copy_link&utm_campaign=client_share&timestamp=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