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

1"""Loading and storing QGIS projects.""" 

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 """QGIS project error.""" 

19 

20 pass 

21 

22 

23class StoreType(gws.Enum): 

24 """Where a QGIS project is stored.""" 

25 

26 file = 'file' 

27 """Project file (``.qgs`` or ``.qgz``).""" 

28 postgres = 'postgres' 

29 """Postgres table ``qgis_projects``.""" 

30 

31 

32class Store(gws.Data): 

33 """Location of a QGIS project.""" 

34 

35 type: StoreType 

36 """Store type.""" 

37 path: gws.FilePath 

38 """Project file path, for file stores.""" 

39 dbUid: str 

40 """Database provider UID, for Postgres stores.""" 

41 schema: str 

42 """Schema of the projects table, for Postgres stores. Defaults to ``public``.""" 

43 projectName: str 

44 """Project name.""" 

45 

46 

47_PRJ_EXT = '.qgs' 

48_ZIP_EXT = '.qgz' 

49_PRJ_TABLE = 'qgis_projects' 

50 

51 

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

53 """Load a QGIS project from a store. 

54 

55 Args: 

56 root: Root object, used to find the database provider. 

57 store: Project location. 

58 

59 Returns: 

60 The project. 

61 

62 Raises: 

63 ``Error``: If the project cannot be found or loaded. 

64 """ 

65 if store.type == StoreType.file: 

66 return from_path(store.path) 

67 if store.type == StoreType.postgres: 

68 return _from_db(root, store) 

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

70 

71 

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

73 """Load a QGIS project from a ``.qgs`` or ``.qgz`` file. 

74 

75 Args: 

76 path: Project file path. 

77 

78 Returns: 

79 The project. 

80 

81 Raises: 

82 ``Error``: If a ``.qgz`` archive contains no project or the QGIS version is not supported. 

83 """ 

84 if path.endswith(_ZIP_EXT): 

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

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

87 

88 

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

90 """Create a QGIS project from XML text. 

91 

92 Args: 

93 text: Project XML. 

94 

95 Returns: 

96 The project. 

97 

98 Raises: 

99 ``Error``: If the QGIS version is not supported. 

100 """ 

101 return Object(text) 

102 

103 

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

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

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

107 if k.endswith(_PRJ_EXT): 

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

109 raise Error(f'no qgis project') 

110 

111 

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

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

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

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

116 

117 with db.connect() as conn: 

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

119 return _from_zipped_bytes(row[0]) 

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

121 

122 

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

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

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

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

127 

128 metadata = { 

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

130 'last_modified_user': 'GWS', 

131 } 

132 

133 with db.connect() as conn: 

134 conn.execute(tab.delete().where(tab.c.name == store.projectName + '.bak')) 

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

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

137 name=store.projectName, 

138 metadata=metadata, 

139 content=content, 

140 )) 

141 conn.commit() 

142 

143 

144class Object: 

145 """A QGIS project, held as XML text. 

146 

147 Only QGIS 3 projects are supported. The parsed XML tree is created on 

148 demand and is not pickled. 

149 """ 

150 

151 version: str 

152 """QGIS version that wrote the project, e.g. ``3.34.0``.""" 

153 sourceHash: str 

154 """SHA-256 hash of the project XML text.""" 

155 

156 def __init__(self, text: str): 

157 """Create a project from XML text. 

158 

159 Args: 

160 text: Project XML. 

161 

162 Raises: 

163 ``Error``: If the QGIS version is not supported. 

164 """ 

165 self.text = text 

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

167 

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

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

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

171 self.version = ver 

172 

173 def __getstate__(self): 

174 """Return the pickle state without the parsed XML tree.""" 

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

176 

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

178 """Return the root element of the project XML. 

179 

180 The XML is parsed on the first call. Changes to the tree are kept 

181 and written by ``to_xml``, ``to_path`` and ``to_store``. 

182 

183 Returns: 

184 Root element. 

185 """ 

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

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

188 return getattr(self, '_xml_root') 

189 

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

191 """Save the project to a store. 

192 

193 In a Postgres store, an existing project with the same name is 

194 renamed to ``<name>.bak``, replacing an older backup. 

195 

196 Args: 

197 root: Root object, used to find the database provider. 

198 store: Project location. 

199 

200 Raises: 

201 ``Error``: If the store type is not supported. 

202 """ 

203 if store.type == StoreType.file: 

204 return self.to_path(store.path) 

205 if store.type == StoreType.postgres: 

206 src = self.to_xml() 

207 name = store.projectName + _PRJ_EXT 

208 content = gws.lib.zipx.zip_to_bytes([{name: src}]) 

209 return _to_db(root, store, content) 

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

211 

212 def to_path(self, path: str): 

213 """Save the project to a file. 

214 

215 If the path ends with ``.qgz``, the project is saved as a zip archive. 

216 

217 Args: 

218 path: File path. 

219 """ 

220 src = self.to_xml() 

221 if path.endswith(_ZIP_EXT): 

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

223 content = gws.lib.zipx.zip_to_bytes([{name: src}]) 

224 gws.u.write_file_b(path, content) 

225 else: 

226 gws.u.write_file(path, src) 

227 

228 def to_xml(self): 

229 """Serialize the project XML. 

230 

231 Returns: 

232 Project XML text. 

233 """ 

234 return self.xml_root().to_string() 

235 

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

237 """Parse the project capabilities. 

238 

239 Returns: 

240 Project capabilities. 

241 

242 Raises: 

243 ``gws.Error``: If the project CRS is invalid. 

244 """ 

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