J6Me1tHZ

1import sqlite3
2from functools import wraps
3
4
5def sqlite_conn(func):
6    @wraps(func)
7    def wrap(self, *args, **kwargs):
8        conn = getattr(self, 'conn', None)
9        if not conn:
10            raise NotImplementedError(
11                'This class has no "conn" object and cannot '
12                'implement this method'
13            )
14        cursor = conn.cursor()
15        try:
16            res = func(self, cursor, *args, **kwargs)
17            conn.commit()
18            return res
19        finally:
20            if kwargs.get('close_conn'):
21                conn.close()
22    return wrap
23
24
25class DbOperations(object):
26    def __init__(self, conn_str):
27        self._conn_str = conn_str
28        self.conn = sqlite3.connect(conn_str)
29
30    @sqlite_conn
31    def create_tables(self, cursor, **kwargs):
32        cursor.execute(
33            'create table if not exists test (id integer, desc text)')
34
35    @sqlite_conn
36    def insert_data(self, cursor, table, data, **kwargs):
37        cursor.execute('insert into {} values (?, ?)'.format(table), data)
38
39    @sqlite_conn
40    def read_data(self, cursor, table, **kwargs):
41        cursor.execute('select * from {}'.format(table))
42        return cursor.fetchall()
43
44
45class NotDbOperations(object):
46    @sqlite_conn
47    def get_something():
48        return 'here you go!'
49
50
51db = DbOperations('/tmp/this-is-your-database-on-tmp.db')
52
53db.create_tables()
54
55db.insert_data('test', (1, 'hai'))
56db.insert_data('test', (2, 'there'))
57
58print(db.read_data('test', conn_close=True))
59
60
61not_db = NotDbOperations()
62
63try:
64    not_db.get_something()
65except NotImplementedError as exc:
66    print(exc)