sqlite_conn_decorator_with_classes.py

import sqlite3
from functools import wraps


def sqlite_conn(func):
    @wraps(func)
    def wrap(self, *args, **kwargs):
        conn = getattr(self, 'conn', None)
        if not conn:
            raise NotImplementedError(
                'This class has no "conn" object and cannot '
                'implement this method'
            )
        cursor = conn.cursor()
        try:
            res = func(self, cursor, *args, **kwargs)
            conn.commit()
            return res
        finally:
            if kwargs.get('close_conn'):
                conn.close()
    return wrap


class DbOperations(object):
    def __init__(self, conn_str):
        self._conn_str = conn_str
        self.conn = sqlite3.connect(conn_str)

    @sqlite_conn
    def create_tables(self, cursor, **kwargs):
        cursor.execute(
            'create table if not exists test (id integer, desc text)')

    @sqlite_conn
    def insert_data(self, cursor, table, data, **kwargs):
        cursor.execute('insert into {} values (?, ?)'.format(table), data)

    @sqlite_conn
    def read_data(self, cursor, table, **kwargs):
        cursor.execute('select * from {}'.format(table))
        return cursor.fetchall()


class NotDbOperations(object):
    @sqlite_conn
    def get_something():
        return 'here you go!'


db = DbOperations('/tmp/this-is-your-database-on-tmp.db')

db.create_tables()

db.insert_data('test', (1, 'hai'))
db.insert_data('test', (2, 'there'))

print(db.read_data('test', conn_close=True))


not_db = NotDbOperations()

try:
    not_db.get_something()
except NotImplementedError as exc:
    print(exc)