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

1from typing import Optional 

2 

3import gws 

4import gws.lib.datetimex as dtx 

5import gws.lib.sqlitex 

6 

7gws.ext.new.storageProvider('sqlite') 

8 

9 

10class Config(gws.Config): 

11 """Configuration for sqlite storage.""" 

12 

13 path: Optional[str] 

14 """Storage path.""" 

15 

16 

17class Object(gws.StorageProvider): 

18 dbPath: str 

19 table = 'storage' 

20 

21 def configure(self): 

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

23 

24 def list_names(self, category): 

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

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

27 

28 def read(self, category, name): 

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

30 for rec in rs: 

31 rec = dict(rec) 

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

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

34 return gws.StorageRecord(**rec) 

35 

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

37 self._db().execute( 

38 f''' 

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

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

41 ON CONFLICT (category, name) DO UPDATE SET 

42 user_uid=excluded.user_uid, 

43 data=excluded.data, 

44 updated=excluded.updated 

45 ''', 

46 category=category, name=name, user_uid=user_uid, data=data, 

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

48 ) 

49 

50 def delete(self, category, name): 

51 self._db().execute( 

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

53 category=category, name=name 

54 ) 

55 

56 ## 

57 

58 _sqlitex: gws.lib.sqlitex.Object 

59 

60 def _db(self): 

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

62 ddl = f''' 

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

64 category TEXT NOT NULL, 

65 name TEXT NOT NULL, 

66 user_uid TEXT, 

67 data TEXT, 

68 created INTEGER, 

69 updated INTEGER, 

70 PRIMARY KEY (category, name) 

71 ) 

72 ''' 

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

74 return self._sqlitex