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

1from typing import Optional 

2 

3import gws 

4import gws.base.storage 

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 

17_DEFAULT_DB_PATH = gws.c.MISC_DIR + '/' + 'storage8.sqlite' 

18_TABLE = 'storage' 

19 

20_INIT_DDL = f''' 

21 CREATE TABLE IF NOT EXISTS {_TABLE} ( 

22 category TEXT NOT NULL, 

23 name TEXT NOT NULL, 

24 user_uid TEXT, 

25 data TEXT, 

26 created INTEGER, 

27 updated INTEGER, 

28 PRIMARY KEY (category, name) 

29 ) 

30''' 

31 

32 

33class Object(gws.StorageProvider): 

34 dbPath: str 

35 

36 def configure(self): 

37 self.dbPath = self.cfg('path', default=_DEFAULT_DB_PATH) 

38 

39 def list_names(self, category): 

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

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

42 

43 def read(self, category, name): 

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

45 for rec in rs: 

46 return gws.StorageRecord(**rec) 

47 

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

49 rec = self.read(category, name) 

50 tmp = gws.u.random_string(64) 

51 

52 self._db().insert(_TABLE, dict( 

53 category=category, 

54 name=tmp if rec else name, 

55 user_uid=user_uid, 

56 data=data, 

57 created=rec.created if rec else gws.u.stime(), 

58 updated=gws.u.stime() 

59 )) 

60 

61 if rec: 

62 self.delete(category, name) 

63 self._db().execute(f'UPDATE {_TABLE} SET name=:name WHERE name=:tmp', name=name, tmp=tmp) 

64 

65 def delete(self, category, name): 

66 self._db().execute( 

67 f'DELETE FROM {_TABLE} WHERE category=:category AND name=:name', 

68 category=category, name=name 

69 ) 

70 

71 ## 

72 

73 _sqlitex: gws.lib.sqlitex.Object 

74 

75 def _db(self): 

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

77 self._sqlitex = gws.lib.sqlitex.Object(self.dbPath, _INIT_DDL) 

78 return self._sqlitex