ยปCore Development>Code coverage>Doc/includes/dbpickle.py

Python code coverage for Doc/includes/dbpickle.py

#countcontent
1n/a# Simple example presenting how persistent ID can be used to pickle
2n/a# external objects by reference.
3n/a
4n/aimport pickle
5n/aimport sqlite3
6n/afrom collections import namedtuple
7n/a
8n/a# Simple class representing a record in our database.
9n/aMemoRecord = namedtuple("MemoRecord", "key, task")
10n/a
11n/aclass DBPickler(pickle.Pickler):
12n/a
13n/a def persistent_id(self, obj):
14n/a # Instead of pickling MemoRecord as a regular class instance, we emit a
15n/a # persistent ID.
16n/a if isinstance(obj, MemoRecord):
17n/a # Here, our persistent ID is simply a tuple, containing a tag and a
18n/a # key, which refers to a specific record in the database.
19n/a return ("MemoRecord", obj.key)
20n/a else:
21n/a # If obj does not have a persistent ID, return None. This means obj
22n/a # needs to be pickled as usual.
23n/a return None
24n/a
25n/a
26n/aclass DBUnpickler(pickle.Unpickler):
27n/a
28n/a def __init__(self, file, connection):
29n/a super().__init__(file)
30n/a self.connection = connection
31n/a
32n/a def persistent_load(self, pid):
33n/a # This method is invoked whenever a persistent ID is encountered.
34n/a # Here, pid is the tuple returned by DBPickler.
35n/a cursor = self.connection.cursor()
36n/a type_tag, key_id = pid
37n/a if type_tag == "MemoRecord":
38n/a # Fetch the referenced record from the database and return it.
39n/a cursor.execute("SELECT * FROM memos WHERE key=?", (str(key_id),))
40n/a key, task = cursor.fetchone()
41n/a return MemoRecord(key, task)
42n/a else:
43n/a # Always raises an error if you cannot return the correct object.
44n/a # Otherwise, the unpickler will think None is the object referenced
45n/a # by the persistent ID.
46n/a raise pickle.UnpicklingError("unsupported persistent object")
47n/a
48n/a
49n/adef main():
50n/a import io
51n/a import pprint
52n/a
53n/a # Initialize and populate our database.
54n/a conn = sqlite3.connect(":memory:")
55n/a cursor = conn.cursor()
56n/a cursor.execute("CREATE TABLE memos(key INTEGER PRIMARY KEY, task TEXT)")
57n/a tasks = (
58n/a 'give food to fish',
59n/a 'prepare group meeting',
60n/a 'fight with a zebra',
61n/a )
62n/a for task in tasks:
63n/a cursor.execute("INSERT INTO memos VALUES(NULL, ?)", (task,))
64n/a
65n/a # Fetch the records to be pickled.
66n/a cursor.execute("SELECT * FROM memos")
67n/a memos = [MemoRecord(key, task) for key, task in cursor]
68n/a # Save the records using our custom DBPickler.
69n/a file = io.BytesIO()
70n/a DBPickler(file).dump(memos)
71n/a
72n/a print("Pickled records:")
73n/a pprint.pprint(memos)
74n/a
75n/a # Update a record, just for good measure.
76n/a cursor.execute("UPDATE memos SET task='learn italian' WHERE key=1")
77n/a
78n/a # Load the records from the pickle data stream.
79n/a file.seek(0)
80n/a memos = DBUnpickler(file, conn).load()
81n/a
82n/a print("Unpickled records:")
83n/a pprint.pprint(memos)
84n/a
85n/a
86n/aif __name__ == '__main__':
87n/a main()