构建 class 以使用 `with` 建立独占 sql 连接
Building a class to use `with` to establish exclusive sql connections
我正在研究一个简单的便利 class 与 with
运算符一起使用,以便我可以建立对 sqlite3 数据库的独占访问,以便在并发系统中进行扩展的写入会话。
这是class:
import sqlite3 as sql
class ExclusiveSqlConnection(object):
"""meant to be used with a with statement to ensure proper closure"""
def __enter__(self, path):
self.con = sql.connect(path, isolation_level = "EXCLUSIVE")
self.con.execute("BEGIN EXCLUSIVE")
self.cur = self.con.cursor()
return self
def __exit__(self):
self.con.commit()
self.con.close()
但是,当我运行它时出现错误:
with sql_lib.ExclusiveSqlConnection(self.output) as c:
TypeError: object.__new__() takes no parameters
您的 ExclusiveSqlConnection
的构造函数(__init__
方法)需要采用 path
参数。
另一方面,__enter__
除了 self
之外没有其他参数。
import sqlite3 as sql
class ExclusiveSqlConnection(object):
"""meant to be used with a with statement to ensure proper closure"""
def __init__(self, path):
self.path = path
def __enter__(self):
self.con = sql.connect(self.path, isolation_level = "EXCLUSIVE")
self.con.execute("BEGIN EXCLUSIVE")
self.cur = self.con.cursor()
return self
def __exit__(self, exception_type, exception_val, trace):
self.con.commit()
self.con.close()
我正在研究一个简单的便利 class 与 with
运算符一起使用,以便我可以建立对 sqlite3 数据库的独占访问,以便在并发系统中进行扩展的写入会话。
这是class:
import sqlite3 as sql
class ExclusiveSqlConnection(object):
"""meant to be used with a with statement to ensure proper closure"""
def __enter__(self, path):
self.con = sql.connect(path, isolation_level = "EXCLUSIVE")
self.con.execute("BEGIN EXCLUSIVE")
self.cur = self.con.cursor()
return self
def __exit__(self):
self.con.commit()
self.con.close()
但是,当我运行它时出现错误:
with sql_lib.ExclusiveSqlConnection(self.output) as c:
TypeError: object.__new__() takes no parameters
您的 ExclusiveSqlConnection
的构造函数(__init__
方法)需要采用 path
参数。
另一方面,__enter__
除了 self
之外没有其他参数。
import sqlite3 as sql
class ExclusiveSqlConnection(object):
"""meant to be used with a with statement to ensure proper closure"""
def __init__(self, path):
self.path = path
def __enter__(self):
self.con = sql.connect(self.path, isolation_level = "EXCLUSIVE")
self.con.execute("BEGIN EXCLUSIVE")
self.cur = self.con.cursor()
return self
def __exit__(self, exception_type, exception_val, trace):
self.con.commit()
self.con.close()