0

I'm trying to patch sqlite3.Cursor to accept any random kwargs and ignore them.

My attempt is below. Essentially this is trying to patch sqlite3.Cursor with TestCursor which accepts **kwargs, and ignores them.

# 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() 

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

Where am I going wrong?

I also tried patching conn directly, but it complains that conn.cursor is read only.

1
  • always put FULL error message (starting at word "Traceback") in question (not in comments) as text (not screenshot, not link to external portal). There are other useful information in the full error/traceback. Commented Dec 19, 2023 at 8:23

1 Answer 1

0

I came across this answer, which gave me the clues I needed.

Working solution was to have a inherit sqlite3.Connection and have a custom cursor function that accepted kwargs. This can also be passed directly to sqlite3 as the connection factory.

# 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() 
Sign up to request clarification or add additional context in comments.

Comments

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.