Оба метода sqlite3.connect()
и sqlite3.onnection.cursor()
позволяют вам указать аргумент factory
для замены обычного соединения или класса курсора вашим собственным подклассом.Вы можете использовать эти пути для предоставления своего собственного cursor.chronology()
метода.
Таким образом, вы бы подклассом класса sqlite3.Cursor
добавили свой собственный метод:
import sqlite3
class ChronologyCursor(sqlite3.Cursor):
def chronology(self):
print("Hello World")
# ...
Затем вы можете использовать этот методclass в качестве аргумента factory
для вызова cursor()
:
>>> db = sqlite3.connect(':memory:')
>>> cursor = db.cursor(factory=ChronologyCursor)
>>> type(cursor)
<class '__main__.ChronologyCursor'>
>>> cursor.chronology()
Hello World
Вы также можете использовать фабрику соединений (подкласс sqlite3.connection()
), чтобы всегда использовать класс курсора:
class ChronologyConnection(sqlite3.Connection):
def cursor(self, *args, **kwargs):
if kwargs.get('factory') is None:
kwargs['factory'] = ChronologyCursor
return super().cursor(*args, **kwargs)
, затем используйте db = sqlite3.connect(':memory:', factory=ChronologyConnection)
, чтобы использовать новый класс подключения:
>>> db = sqlite3.connect(':memory:', factory=ChronologyConnection)
>>> type(db)
<class '__main__.ChronologyConnection'>
>>> cursor = db.cursor()
>>> cursor.chronology()
Hello World
Я настоятельно рекомендую против исправления функции sqlite3.connect()
, чтобы сделать вышеуказанную фабрику настройкой по умолчанию, но если вы должен сделать это прозрачным, вы можете поставить свою собственную функцию connect()
на модуль sqlite3
.Базовый пакет - чистый Python, так что вы можете обезьяна-патч добавить в него альтернативную connect()
функцию.
Итак, в вашем модуле scheduling
вы поместите вышеупомянутые классы, а затемиспользуйте:
_sqlite3_connect = sqlite3.connect
def chronology_connect(*args, **kwargs):
if kwargs.get('factory') is None:
kwargs['factory'] = ChronologyConnection
return _sqlite3_connect(*args, **kwargs)
sqlite3.connect = chronology_connect
Теперь достаточно импортировать scheduling
, чтобы совершать вызовы на sqlite3.connect()
. Использовать альтернативную фабрику соединений:
>>> import sqlite3
>>> import scheduling
>>> db = sqlite3.connect(':memory:')
>>> cursor = db.cursor()
>>> cursor.chronology()
Hello World
>>> type(db)
<class 'scheduling.ChronologyConnection'>
>>> type(cursor)
<class 'scheduling.ChronologyCursor'>
Причина, по которой вы не хочу сделать вышеизложенное, это то, что sqlite3.connect()
вызовы отклоняются от нормы, действуют так, что противоречат тому, что говорится в документации.Это делает ваш код сложнее поддерживать в долгосрочной перспективе.Вместо этого я бы придерживался явных sqlite3.connect(...., factory=ChronologyConnection)
вызовов.