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 

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 return gws.StorageRecord(**rec) 

32 

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

34 rec = self.read(category, name) 

35 tmp = gws.u.random_string(64) 

36 

37 self._db().insert(self.table, dict( 

38 category=category, 

39 name=tmp if rec else name, 

40 user_uid=user_uid, 

41 data=data, 

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

43 updated=gws.u.stime() 

44 )) 

45 

46 if rec: 

47 self.delete(category, name) 

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

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