0

I was writing a few simple CRUD operations to try out sqlite3 with Python, and then I saw a nice function that executes queries and closes connection in this answer:

from contextlib import closing
import sqlite3

def query(self, db_name, sql):
    with closing(sqlite3.connect(db_name)) as con, con,  \
            closing(con.cursor()) as cur:
        cur.execute(sql)
        return cur.fetchall()

I thought it would be nice to have something like this and call this function with whatever sql sentence I need whenever I want to query the database.

However, when I'm running an insert I'd need to return cur.lastrowid instead or cur.fetchall() and when deleting I'd like to know the cursor.rowcount instead. Also, sometimes I need to add parameters to the query, for instance sometimes I want to run select * from [some_table] and some other times I need select * from [some_table] where [some_column] = ?. So the function needs some tweaks depending on what kind of operation is being executed.

I could write one function for each kind of operation, with the same basic structure and the tweaks each query needs. But that sounds a bit repetitive since there would be duplicate chunks of code and these functions would look pretty similar to each other. So I'm not sure it's the right approach.

Is there another alternative to make this function a bit more "generic" to fit all cases?

Floella
  • 1,279
  • 1
  • 22
  • 41

1 Answers1

1

One option is to have callouts in the with clause that let you customize program actions. There are many ways to do this. One is to write a class that calls methods to allow specialization. In this example, a class has pre and post processers. It does its work in __init__ and leaves its result in an instance variable which allows for terse usage.

from contextlib import closing
import sqlite3
            
class SqlExec:

    def __init__(self, db_name, sql, parameters=()):
        self.sql = sql
        self.parameters = parameters
        with closing(sqlite3.connect(db_name)) as self.con,  \
                closing(con.cursor()) as self.cur:
            self.pre_process()
            self.cur.execute(self.sql, parameters=self.parameters)
            self.retval = self.post_process()
        
    def pre_process(self):
        return

    def post_process_fetchall(self):
        self.retval = self.cur.fetchall

    post_process = post_process_fetchall

class SqlExecLastRowId(SqlExec):

    def post_process(self):
        self.retval = cur.lastrowid    

last_row = SqlExecLastRowId("mydb.db", "DELETE FROM FOO WHERE BAR='{}'",
    paramters=("baz",)).retval
tdelaney
  • 73,364
  • 6
  • 83
  • 116