Coverage for gws-app/gws/plugin/qgis/project.py: 0%

1"""Qgis Project API.""" 

2 

3import os 

4 

5import gws 

6import gws.base.database 

7import gws.config.util 

8import gws.lib.jsonx 

9import gws.lib.datetimex 

10import gws.lib.xmlx 

11import gws.lib.zipx 

12import gws.lib.sa as sa 

13 

14from . import caps 

15 

16 

17class Error(gws.Error): 

18 pass 

19 

20 

21class StoreType(gws.Enum): 

22 file = 'file' 

23 postgres = 'postgres' 

24 

25 

26class Store(gws.Data): 

27 type: StoreType 

28 path: gws.FilePath 

29 dbUid: str 

30 schema: str 

31 projectName: str 

32 

33 

34_PRJ_EXT = '.qgs' 

35_ZIP_EXT = '.qgz' 

36_PRJ_TABLE = 'qgis_projects' 

37 

38 

39def from_store(root: gws.Root, store: Store) -> 'Object': 

40 if store.type == StoreType.file: 

41 return from_path(store.path) 

42 if store.type == StoreType.postgres: 

43 return _from_db(root, store) 

44 raise Error(f'qgis project cannot be loaded') 

45 

46 

47def from_path(path: str) -> 'Object': 

48 if path.endswith(_ZIP_EXT): 

49 return _from_zipped_bytes(gws.u.read_file_b(path)) 

50 return from_string(gws.u.read_file(path)) 

51 

52 

53def from_string(text: str) -> 'Object': 

54 return Object(text) 

55 

56 

57def _from_zipped_bytes(b: bytes) -> 'Object': 

58 d = gws.lib.zipx.unzip_bytes_to_dict(b) 

59 for k, v in d.items(): 

60 if k.endswith(_PRJ_EXT): 

61 return from_string(v.decode('utf8')) 

62 raise Error(f'no qgis project') 

63 

64 

65def _from_db(root: gws.Root, store: Store): 

66 db = root.app.databaseMgr.find_provider(ext_type='postgres', uid=store.dbUid) 

67 schema = store.get('schema') or 'public' 

68 tab = db.table(f'{schema}.{_PRJ_TABLE}') 

69 

70 with db.connect() as conn: 

71 for row in conn.execute(sa.select(tab.c.content).where(tab.c.name.__eq__(store.projectName))): 

72 return _from_zipped_bytes(row[0]) 

73 raise Error(f'{store.projectName!r} not found') 

74 

75 

76def _to_db(root: gws.Root, store: Store, content: bytes): 

77 db = root.app.databaseMgr.find_provider(ext_type='postgres', uid=store.dbUid) 

78 schema = store.get('schema') or 'public' 

79 tab = db.table(f'{schema}.{_PRJ_TABLE}') 

80 

81 metadata = { 

82 'last_modified_time': gws.lib.datetimex.to_iso_string(), 

83 'last_modified_user': 'GWS', 

84 } 

85 

86 with db.connect() as conn: 

87 conn.execute(tab.delete().where(tab.c.name.__eq__(store.projectName + '.bak'))) 

88 conn.execute(tab.update().values(name=store.projectName + '.bak').where(tab.c.name.__eq__(store.projectName))) 

89 conn.execute(tab.insert().values( 

90 name=store.projectName, 

91 metadata=metadata, 

92 content=content, 

93 )) 

94 conn.commit() 

95 

96 

97class Object: 

98 version: str 

99 sourceHash: str 

100 

101 def __init__(self, text: str): 

102 self.text = text 

103 self.sourceHash = gws.u.sha256(self.text) 

104 

105 ver = self.xml_root().get('version', '').split('-')[0] 

106 if not ver.startswith('3'): 

107 raise Error(f'unsupported qgis version {ver!r}') 

108 self.version = ver 

109 

110 def __getstate__(self): 

111 return gws.u.omit(vars(self), '_xml_root') 

112 

113 def xml_root(self) -> gws.XmlElement: 

114 if not hasattr(self, '_xml_root'): 

115 setattr(self, '_xml_root', gws.lib.xmlx.from_string(self.text)) 

116 return getattr(self, '_xml_root') 

117 

118 def to_store(self, root: gws.Root, store: Store): 

119 if store.type == StoreType.file: 

120 return self.to_path(store.path) 

121 if store.type == StoreType.postgres: 

122 src = self.to_xml() 

123 name = store.projectName + _PRJ_EXT 

124 content = gws.lib.zipx.zip_to_bytes({name: src}) 

125 return _to_db(root, store, content) 

126 raise Error(f'qgis project cannot be stored') 

127 

128 def to_path(self, path: str): 

129 src = self.to_xml() 

130 if path.endswith(_ZIP_EXT): 

131 name = os.path.basename(path).replace(_ZIP_EXT, _PRJ_EXT) 

132 content = gws.lib.zipx.zip_to_bytes({name: src}) 

133 gws.u.write_file_b(path, content) 

134 else: 

135 gws.u.write_file(path, src) 

136 

137 def to_xml(self): 

138 return self.xml_root().to_string() 

139 

140 def caps(self) -> caps.Caps: 

141 return caps.parse_element(self.xml_root())