Python：修补 sqlite3.Cursor 以接受 kwargs

Question

我正在尝试修补

sqlite3.Cursor

以接受任何随机 kwargs 并忽略它们。

我的尝试如下。本质上，这是试图用

sqlite3.Cursor

来修补

TestCursor

，它接受

**kwargs

，并忽略它们。

# in my test code
import sqlite3
from contextlib import contextmanager
from unittest.mock import patch

class TestCursor(sqlite3.Cursor):
    def __init__(self, **kwargs):
        super().__init__()

@contextmanager
def get_connection():
    with patch('sqlite3.Cursor', TestCursor):
        conn = sqlite3.connect(':memory:')
        # do some database setup
        try:
            yield conn
        finally:
            conn.close()
        

# The function I'm trying to test
def send_query():
    with get_connection() as conn:
        
        #offending line
        cur = conn.cursor(row_factory='foo')
        
        cur.execute("SELECT * FROM table")
        data = cur.fetchall()
        return data
    
send_query() 

这给了我

TypeError: 'row_factory' is an invalid keyword argument for this function

我哪里出错了？

我也尝试直接修补

conn

，但它抱怨

conn.cursor

是只读的。

Answer 1

我遇到了this答案，它给了我所需的线索。

工作解决方案是继承

sqlite3.Connection

并拥有接受 kwargs 的自定义光标函数。这也可以直接传递给

sqlite3

作为连接工厂。

# in my test code
import sqlite3
from contextlib import contextmanager

class TestConnect(sqlite3.Connection):
    def cursor(self, **kwargs):
        return super(TestConnect, self).cursor()

@contextmanager
def get_connection():
    conn = sqlite3.connect(':memory:', factory=TestConnect)
    # do some database setup
    try:
        yield conn
    finally:
        conn.close()
        

# The function I'm trying to test
def send_query():
    with get_connection() as conn:
        cur = conn.cursor(row_factory='foo')
        cur.execute("CREATE TABLE scores_view(foo, bar, baz)")
        data = cur.fetchall()
        return data

send_query()