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

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.begin() 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.begin() 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 

142 

143class Object: 

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

145 

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

147 demand and is not pickled. 

148 """ 

149 

150 version: str 

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

152 sourceHash: str 

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

154 

155 def __init__(self, text: str): 

156 """Create a project from XML text. 

157 

158 Args: 

159 text: Project XML. 

160 

161 Raises: 

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

163 """ 

164 self.text = text 

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

166 

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

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

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

170 self.version = ver 

171 

172 def __getstate__(self): 

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

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

175 

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

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

178 

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

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

181 

182 Returns: 

183 Root element. 

184 """ 

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

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

187 return getattr(self, '_xml_root') 

188 

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

190 """Save the project to a store. 

191 

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

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

194 

195 Args: 

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

197 store: Project location. 

198 

199 Raises: 

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

201 """ 

202 if store.type == StoreType.file: 

203 return self.to_path(store.path) 

204 if store.type == StoreType.postgres: 

205 src = self.to_xml() 

206 name = store.projectName + _PRJ_EXT 

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

208 return _to_db(root, store, content) 

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

210 

211 def to_path(self, path: str): 

212 """Save the project to a file. 

213 

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

215 

216 Args: 

217 path: File path. 

218 """ 

219 src = self.to_xml() 

220 if path.endswith(_ZIP_EXT): 

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

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

223 gws.u.write_file_b(path, content) 

224 else: 

225 gws.u.write_file(path, src) 

226 

227 def to_xml(self): 

228 """Serialize the project XML. 

229 

230 Returns: 

231 Project XML text. 

232 """ 

233 return self.xml_root().to_string() 

234 

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

236 """Parse the project capabilities. 

237 

238 Returns: 

239 Project capabilities. 

240 

241 Raises: 

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

243 """ 

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