Coverage for gws-app/gws/plugin/storage_provider/sqlite/__init__.py: 100%

1"""SQLite storage provider. 

2 

3The ``sqlite`` storage provider keeps saved user entries in an SQLite 

4database file, by default ``storage8.sqlite`` in the GWS misc directory. 

5The ``storage`` table is created on first use. Each entry is a row keyed 

6by category and name, with the serialized data, the UID of the user who 

7wrote it, and the creation and update timestamps. 

8 

9Example:: 

10 

11 storage { 

12 providers+ { 

13 type "sqlite" 

14 path "/data/storage.sqlite" 

15 } 

16 } 

17""" 

18 

19from typing import Optional 

20 

21import gws 

22import gws.lib.datetimex as dtx 

23import gws.lib.sqlitex 

24 

25 

26@gws.ext.config.storageProvider('sqlite') 

27class Config(gws.Config): 

28 """Storage provider that keeps saved user data in an SQLite file.""" 

29 

30 path: Optional[str] 

31 """Database file.""" 

32 

33 

34@gws.ext.object.storageProvider('sqlite') 

35class Object(gws.StorageProvider): 

36 """Storage provider backed by an SQLite database file.""" 

37 

38 dbPath: str 

39 """Path to the database file.""" 

40 table = 'storage' 

41 """Name of the table that holds the entries.""" 

42 

43 def configure(self): 

44 self.dbPath = self.cfg('path', default=f'{gws.c.MISC_DIR}/storage8.sqlite') 

45 

46 def list_names(self, category): 

47 rs = self._db().select(f'SELECT name FROM {self.table} WHERE category=:category', category=category) 

48 return sorted(rec['name'] for rec in rs) 

49 

50 def read(self, category, name): 

51 rs = self._db().select(f'SELECT * FROM {self.table} WHERE category=:category AND name=:name', category=category, name=name) 

52 for rec in rs: 

53 rec = dict(rec) 

54 rec['created'] = dtx.from_timestamp(rec['created']) 

55 rec['updated'] = dtx.from_timestamp(rec['updated']) 

56 return gws.StorageRecord(**rec) 

57 

58 def write(self, category, name, data, user_uid): 

59 self._db().execute( 

60 f''' 

61 INSERT INTO {self.table} (category, name, user_uid, data, created, updated) 

62 VALUES (:category, :name, :user_uid, :data, :created, :updated) 

63 ON CONFLICT (category, name) DO UPDATE SET 

64 user_uid=excluded.user_uid, 

65 data=excluded.data, 

66 updated=excluded.updated 

67 ''', 

68 category=category, name=name, user_uid=user_uid, data=data, 

69 created=gws.u.stime(), updated=gws.u.stime() 

70 ) 

71 

72 def delete(self, category, name): 

73 self._db().execute( 

74 f'DELETE FROM {self.table} WHERE category=:category AND name=:name', 

75 category=category, name=name 

76 ) 

77 

78 ## 

79 

80 _sqlitex: gws.lib.sqlitex.Object 

81 

82 def _db(self): 

83 """Return the database object, creating the table if needed.""" 

84 if getattr(self, '_sqlitex', None) is None: 

85 ddl = f''' 

86 CREATE TABLE IF NOT EXISTS {self.table} ( 

87 category TEXT NOT NULL, 

88 name TEXT NOT NULL, 

89 user_uid TEXT, 

90 data TEXT, 

91 created INTEGER, 

92 updated INTEGER, 

93 PRIMARY KEY (category, name) 

94 ) 

95 ''' 

96 self._sqlitex = gws.lib.sqlitex.Object(self.dbPath, ddl) 

97 return self._sqlitex