Coverage for python/lsst/summit/utils/consdbClient.py: 19%

138 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2025-10-14 03:12 -0700

1# This file is part of summit_utils. 

2# 

3# Developed for the LSST Data Management System. 

4# This product includes software developed by the LSST Project 

5# (http://www.lsst.org). 

6# See the COPYRIGHT file at the top-level directory of this distribution 

7# for details of code ownership. 

8# 

9# This program is free software: you can redistribute it and/or modify 

10# it under the terms of the GNU General Public License as published by 

11# the Free Software Foundation, either version 3 of the License, or 

12# (at your option) any later version. 

13# 

14# This program is distributed in the hope that it will be useful, 

15# but WITHOUT ANY WARRANTY; without even the implied warranty of 

16# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

17# GNU General Public License for more details. 

18# 

19# You should have received a copy of the GNU General Public License 

20# along with this program. If not, see <http://www.gnu.org/licenses/>. 

21 

22import logging 

23import os 

24from dataclasses import dataclass 

25from typing import Any 

26from urllib.parse import quote, urlparse 

27 

28import requests 

29from astropy.table import Table 

30 

31__all__ = ["ConsDbClient", "FlexibleMetadataInfo"] 

32 

33 

34logger = logging.getLogger(__name__) 

35 

36 

37def _urljoin(*args: str) -> str: 

38 """Join parts of a URL with slashes. 

39 

40 Does not do any quoting. Mostly to remove a level of list-making. 

41 

42 Parameters 

43 ---------- 

44 *args : `str` 

45 Each parameter is a URL part. 

46 

47 Returns 

48 ------- 

49 url : `str` 

50 The joined URL. 

51 """ 

52 return "/".join(args) 

53 

54 

55def _check_status(r: requests.Response) -> None: 

56 """Check the status of an HTTP response and raise if an error. 

57 

58 Adds additional response information to the raise_for_status exception. 

59 

60 Parameters 

61 ---------- 

62 r : `requests.Response` 

63 The response to check. 

64 

65 Raises 

66 ------ 

67 requests.HTTPError 

68 Raised if a non-successful status is returned. 

69 """ 

70 try: 

71 r.raise_for_status() 

72 except requests.HTTPError as e: 

73 try: 

74 json_data = e.response.json() 

75 e.add_note(str(json_data)) 

76 if "message" in json_data: 

77 e.add_note(f"\n\n{json_data['message']}") 

78 except requests.JSONDecodeError: 

79 pass 

80 raise e 

81 

82 

83def clean_url(resp: requests.Response, *args, **kwargs) -> requests.Response: 

84 """Parse url from response and remove netloc portion. 

85 

86 Set new url in response and return response 

87 

88 Parameters 

89 ---------- 

90 resp : `requests.Response` 

91 The response that could contain a URL with tokens 

92 """ 

93 url = urlparse(resp.url) 

94 short_user = f"{url.username[:2]}***" if url.username is not None else "" 

95 short_pass = f":{url.password[:2]}***" if url.password is not None else "" 

96 netloc = f"{short_user}{short_pass}@{url.hostname}" 

97 resp.url = url._replace(netloc=netloc).geturl() 

98 return resp 

99 

100 

101@dataclass 

102class FlexibleMetadataInfo: 

103 """Description of a flexible metadata value. 

104 

105 Parameters 

106 ---------- 

107 dtype : `str` 

108 Data type of the flexible metadata value. 

109 One of ``bool``, ``int``, ``float``, or ``str``. 

110 doc : `str` 

111 Documentation string. 

112 unit : `str`, optional 

113 Unit of the value. 

114 ucd : `str`, optional 

115 IVOA Unified Content Descriptor 

116 (https://www.ivoa.net/documents/UCD1+/). 

117 """ 

118 

119 dtype: str 

120 doc: str 

121 unit: str | None = None 

122 ucd: str | None = None 

123 

124 

125class ConsDbClient: 

126 """A client library for accessing the Consolidated Database. 

127 

128 This library provides a basic interface for using flexible metadata 

129 (key/value pairs associated with observation ids from an observation 

130 type table), determining the schema of ConsDB tables, querying the 

131 ConsDB using a general SQL SELECT statement, and inserting into 

132 ConsDB tables. 

133 

134 Parameters 

135 ---------- 

136 url : `str`, optional 

137 Base URL of the Web service, defaults to the value of environment 

138 variable ``LSST_CONSDB_PQ_URL`` (the location of the publish/query 

139 service). 

140 token : `str`, optional 

141 Authentication token for the RSP. The token must begin with "gt-". 

142 

143 Notes 

144 ----- 

145 This client is a thin layer over the publish/query Web service, which 

146 avoids having a dependency on database drivers. 

147 

148 It enforces the return of query results as Astropy Tables. 

149 """ 

150 

151 def __init__(self, url: str | None = None, token: str | None = None): 

152 self.session = requests.Session() 

153 self.session.hooks["response"].append(clean_url) 

154 

155 if token is not None: 

156 if not token.startswith("gt-"): 

157 raise ValueError("token must start with `gt-`.") 

158 

159 self.session.headers.update({"Authorization": f"Bearer {token}"}) 

160 

161 if url is None: 

162 self.url = os.environ["LSST_CONSDB_PQ_URL"] 

163 else: 

164 self.url = url 

165 self.url = self.url.rstrip("/") 

166 

167 def _handle_get(self, url: str, query: dict[str, str | list[str]] | None = None) -> Any: 

168 """Submit GET requests to the server. 

169 

170 Parameters 

171 ---------- 

172 url : `str` 

173 URL to GET. 

174 query : `dict` [`str`, `str` | `list` [`str`]], optional 

175 Query parameters to attach to the URL. 

176 

177 Raises 

178 ------ 

179 requests.RequestException 

180 Raised if any kind of connection error occurs. 

181 requests.HTTPError 

182 Raised if a non-successful status is returned. 

183 requests.JSONDecodeError 

184 Raised if the result does not decode as JSON. 

185 

186 Returns 

187 ------- 

188 result : `Any` 

189 Result of decoding the Web service result content as JSON. 

190 """ 

191 logger.debug(f"GET {url}") 

192 response = self.session.get(url, params=query) 

193 _check_status(response) 

194 return response.json() 

195 

196 def _handle_post(self, url: str, data: dict[str, Any]) -> requests.Response: 

197 """Submit POST requests to the server. 

198 

199 Parameters 

200 ---------- 

201 url : `str` 

202 URL to POST. 

203 data : `dict` [`str`, `Any`] 

204 Key/value pairs of data to POST. 

205 

206 Raises 

207 ------ 

208 requests.RequestException 

209 Raised if any kind of connection error occurs. 

210 requests.HTTPError 

211 Raised if a non-successful status is returned. 

212 

213 Returns 

214 ------- 

215 result : `requests.Response` 

216 The raw Web service result object. 

217 """ 

218 logger.debug(f"POST {url}: {data}") 

219 response = self.session.post(url, json=data) 

220 _check_status(response) 

221 return response 

222 

223 @staticmethod 

224 def compute_flexible_metadata_table_name(instrument: str, obs_type: str) -> str: 

225 """Compute the name of a flexible metadata table. 

226 

227 Each instrument and observation type made with that instrument can 

228 have a flexible metadata table. This function is useful when 

229 issuing SQL queries, and it avoids a round-trip to the server. 

230 

231 Parameters 

232 ---------- 

233 instrument : `str` 

234 Name of the instrument (e.g. ``LATISS``). 

235 obs_type : `str` 

236 Name of the observation type (e.g. ``Exposure``). 

237 

238 Returns 

239 ------- 

240 table_name : `str` 

241 Name of the appropriate flexible metadata table. 

242 """ 

243 return f"cdb_{instrument}.{obs_type}_flexdata" 

244 

245 @staticmethod 

246 def compute_fixed_metadata_namespace(instrument: str) -> str: 

247 """Compute the namespace for a fixed metadata table. 

248 

249 Each instrument has its own namespace in the ConsDB. 

250 This function is useful when issuing SQL queries, and it avoids a 

251 round-trip to the server. 

252 

253 Parameters 

254 ---------- 

255 instrument : `str` 

256 Name of the instrument (e.g. ``LATISS``). 

257 

258 Returns 

259 ------- 

260 namespace_name : `str` 

261 Name of the appropriate namespace 

262 """ 

263 return f"cdb_{instrument}" 

264 

265 def add_flexible_metadata_key( 

266 self, 

267 instrument: str, 

268 obs_type: str, 

269 key: str, 

270 dtype: str, 

271 doc: str, 

272 unit: str | None = None, 

273 ucd: str | None = None, 

274 ) -> requests.Response: 

275 """Add a key to a flexible metadata table. 

276 

277 Parameters 

278 ---------- 

279 instrument : `str` 

280 Name of the instrument (e.g. ``LATISS``). 

281 obs_type : `str` 

282 Name of the observation type (e.g. ``Exposure``). 

283 key : `str` 

284 Name of the key to be added (must not already exist). 

285 dtype : `str` 

286 One of ``bool``, ``int``, ``float``, or ``str``. 

287 doc : `str` 

288 Documentation string for the key. 

289 unit : `str`, optional 

290 Unit for the value. Should be from the IVOA 

291 (https://www.ivoa.net/documents/VOUnits/) or astropy. 

292 ucd : `str`, optional 

293 IVOA Unified Content Descriptor 

294 (https://www.ivoa.net/documents/UCD1+/). 

295 

296 Returns 

297 ------- 

298 response : `requests.Response` 

299 HTTP response from the server, with 200 status for success. 

300 

301 Raises 

302 ------ 

303 requests.RequestException 

304 Raised if any kind of connection error occurs. 

305 requests.HTTPError 

306 Raised if a non-successful status is returned. 

307 """ 

308 data = {"key": key, "dtype": dtype, "doc": doc} 

309 if unit is not None: 

310 data["unit"] = unit 

311 if ucd is not None: 

312 data["ucd"] = ucd 

313 url = _urljoin(self.url, "flex", quote(instrument), quote(obs_type), "addkey") 

314 return self._handle_post(url, data) 

315 

316 def get_flexible_metadata_keys(self, instrument: str, obs_type: str) -> dict[str, FlexibleMetadataInfo]: 

317 """Retrieve the valid keys for a flexible metadata table. 

318 

319 Parameters 

320 ---------- 

321 instrument : `str` 

322 Name of the instrument (e.g. ``LATISS``). 

323 obs_type : `str` 

324 Name of the observation type (e.g. ``Exposure``). 

325 

326 Returns 

327 ------- 

328 key_info : `dict` [ `str`, `FlexibleMetadataInfo` ] 

329 Dict of keys and information values. 

330 

331 Raises 

332 ------ 

333 requests.RequestException 

334 Raised if any kind of connection error occurs. 

335 requests.HTTPError 

336 Raised if a non-successful status is returned. 

337 """ 

338 url = _urljoin(self.url, "flex", quote(instrument), quote(obs_type), "schema") 

339 result = self._handle_get(url) 

340 return {key: FlexibleMetadataInfo(*value) for key, value in result.items()} 

341 

342 def get_flexible_metadata( 

343 self, instrument: str, obs_type: str, obs_id: int, keys: list[str] | None = None 

344 ) -> dict[str, Any]: 

345 """Get the flexible metadata for an observation. 

346 

347 Parameters 

348 ---------- 

349 instrument : `str` 

350 Name of the instrument (e.g. ``LATISS``). 

351 obs_type : `str` 

352 Name of the observation type (e.g. ``Exposure``). 

353 obs_id : `int` 

354 Unique observation id. 

355 keys : `list` [ `str` ], optional 

356 List of keys to be retrieved; all if not specified. 

357 

358 Returns 

359 ------- 

360 result_dict : `dict` [ `str`, `Any` ] 

361 Dictionary of key/value pairs for the observation. 

362 

363 Raises 

364 ------ 

365 requests.RequestException 

366 Raised if any kind of connection error occurs. 

367 requests.HTTPError 

368 Raised if a non-successful status is returned. 

369 """ 

370 url = _urljoin( 

371 self.url, 

372 "flex", 

373 quote(instrument), 

374 quote(obs_type), 

375 "obs", 

376 quote(str(obs_id)), 

377 ) 

378 return self._handle_get(url, {"k": keys} if keys else None) 

379 

380 def get_all_metadata( 

381 self, instrument: str, obs_type: str, obs_id: int, flex: bool = False 

382 ) -> dict[str, Any]: 

383 """Get all metadata for an observation. 

384 

385 Parameters 

386 ---------- 

387 instrument : `str` 

388 Name of the instrument (e.g. ``LATISS``). 

389 obs_type : `str` 

390 Name of the observation type (e.g. ``Exposure``). 

391 obs_id : `int` 

392 Unique observation id. 

393 flex : `bool` 

394 Include flexible metadata. 

395 

396 Returns 

397 ------- 

398 result_dict : `dict` [ `str`, `Any` ] 

399 Dictionary of key/value pairs for the observation. 

400 

401 Raises 

402 ------ 

403 requests.RequestException 

404 Raised if any kind of connection error occurs. 

405 requests.HTTPError 

406 Raised if a non-successful status is returned. 

407 """ 

408 url = _urljoin( 

409 self.url, 

410 "query", 

411 quote(instrument), 

412 quote(obs_type), 

413 "obs", 

414 quote(str(obs_id)), 

415 ) 

416 return self._handle_get(url, {"flex": "1"} if flex else None) 

417 

418 def insert_flexible_metadata( 

419 self, 

420 instrument: str, 

421 obs_type: str, 

422 obs_id: int, 

423 values: dict[str, Any] | None = None, 

424 *, 

425 allow_update: bool = False, 

426 **kwargs, 

427 ) -> requests.Response: 

428 """Set flexible metadata values for an observation. 

429 

430 Parameters 

431 ---------- 

432 instrument : `str` 

433 Name of the instrument (e.g. ``LATISS``). 

434 obs_type : `str` 

435 Name of the observation type (e.g. ``Exposure``). 

436 obs_id : `int` 

437 Unique observation id. 

438 values : `dict` [ `str`, `Any` ], optional 

439 Dictionary of key/value pairs to add for the observation. 

440 allow_update : `bool`, optional 

441 If ``True``, allow replacement of values of existing keys. 

442 **kwargs : `dict` 

443 Additional key/value pairs, overriding ``values``. 

444 

445 Returns 

446 ------- 

447 response : `requests.Response` 

448 HTTP response from the server, with 200 status for success. 

449 

450 Raises 

451 ------ 

452 ValueError 

453 Raised if no values are provided in ``values`` or kwargs. 

454 requests.RequestException 

455 Raised if any kind of connection error occurs. 

456 requests.HTTPError 

457 Raised if a non-successful status is returned. 

458 """ 

459 if values: 

460 values.update(kwargs) 

461 else: 

462 values = kwargs 

463 if not values: 

464 raise ValueError(f"No values to set for {instrument} {obs_type} {obs_id}") 

465 data = {"values": values} 

466 url = _urljoin( 

467 self.url, 

468 "flex", 

469 quote(instrument), 

470 quote(obs_type), 

471 "obs", 

472 quote(str(obs_id)), 

473 ) 

474 if allow_update: 

475 url += "?u=1" 

476 return self._handle_post(url, data) 

477 

478 def insert( 

479 self, 

480 instrument: str, 

481 table: str, 

482 obs_id: tuple[int, int] | int, 

483 values: dict[str, Any], 

484 *, 

485 allow_update=False, 

486 **kwargs, 

487 ) -> requests.Response: 

488 """Insert values into a single ConsDB fixed metadata table. 

489 

490 Parameters 

491 ---------- 

492 instrument : `str` 

493 Name of the instrument (e.g. ``LATISS``). 

494 table : `str` 

495 Name of the table to insert into. 

496 obs_id : `tuple` [ `int`, `int`] or `int` 

497 Unique observation id or day_obs and seq_num. 

498 values : `dict` [ `str`, `Any` ], optional 

499 Dictionary of column/value pairs to add for the observation. 

500 allow_update : `bool`, optional 

501 If ``True``, allow replacement of values of existing columns. 

502 **kwargs : `dict` 

503 Additional column/value pairs, overriding ``values``. 

504 

505 Returns 

506 ------- 

507 response : `requests.Response` 

508 HTTP response from the server, with 200 status for success. 

509 

510 Raises 

511 ------ 

512 ValueError 

513 Raised if no values are provided in ``values`` or kwargs. 

514 requests.RequestException 

515 Raised if any kind of connection error occurs. 

516 requests.HTTPError 

517 Raised if a non-successful status is returned. 

518 """ 

519 if values: 

520 values.update(kwargs) 

521 else: 

522 values = kwargs 

523 if not values: 

524 raise ValueError(f"No values to insert for {instrument} {table} {obs_id}") 

525 

526 data: dict[str, Any] 

527 if isinstance(obs_id, tuple): 

528 data = {"table": table, "values": values} 

529 url = _urljoin( 

530 self.url, 

531 "insert", 

532 quote(instrument), 

533 quote(table), 

534 "by_seq_num", 

535 quote(str(obs_id[0])), 

536 quote(str(obs_id[1])), 

537 ) 

538 else: 

539 data = {"table": table, "obs_id": obs_id, "values": values} 

540 url = _urljoin( 

541 self.url, 

542 "insert", 

543 quote(instrument), 

544 quote(table), 

545 "obs", 

546 quote(str(obs_id)), 

547 ) 

548 if allow_update: 

549 url += "?u=1" 

550 return self._handle_post(url, data) 

551 

552 def insert_multiple( 

553 self, 

554 instrument: str, 

555 table: str, 

556 obs_dict: dict[int, dict[str, Any]], 

557 *, 

558 allow_update=False, 

559 ) -> requests.Response: 

560 """Insert values into a single ConsDB fixed metadata table. 

561 

562 Parameters 

563 ---------- 

564 instrument : `str` 

565 Name of the instrument (e.g. ``LATISS``). 

566 table : `str` 

567 Name of the table to insert into. 

568 obs_dict : `dict` [ `int`, `dict` [ `str`, `Any` ] ] 

569 Dictionary of observation ids, each with a dictionary of 

570 column/value pairs to add for each observation. 

571 allow_update : `bool`, optional 

572 If ``True``, allow replacement of values of existing columns. 

573 

574 Returns 

575 ------- 

576 response : `requests.Response` 

577 HTTP response from the server, with 200 status for success. 

578 

579 Raises 

580 ------ 

581 ValueError 

582 Raised if no values are provided in ``obs_dict``. 

583 requests.RequestException 

584 Raised if any kind of connection error occurs. 

585 requests.HTTPError 

586 Raised if a non-successful status is returned. 

587 """ 

588 if not obs_dict: 

589 raise ValueError(f"No values to insert for {instrument} {table}") 

590 data = {"table": table, "obs_dict": obs_dict} 

591 url = _urljoin( 

592 self.url, 

593 "insert", 

594 quote(instrument), 

595 quote(table), 

596 ) 

597 if allow_update: 

598 url += "?u=1" 

599 return self._handle_post(url, data) 

600 

601 def query(self, query: str) -> Table: 

602 """Query the ConsDB database. 

603 

604 Parameters 

605 ---------- 

606 query : `str` 

607 A SQL query (currently) to the database. 

608 

609 Returns 

610 ------- 

611 result : `Table` 

612 An ``astropy.Table`` containing the query results. 

613 

614 Raises 

615 ------ 

616 requests.RequestException 

617 Raised if any kind of connection error occurs. 

618 requests.HTTPError 

619 Raised if a non-successful status is returned. 

620 

621 Notes 

622 ----- 

623 This is a very general query interface because it is expected that 

624 a wide variety of types of queries will be needed. If some types prove 

625 to be common, syntactic sugar could be added to make them simpler. 

626 """ 

627 url = _urljoin(self.url, "query") 

628 data = {"query": query} 

629 result = self._handle_post(url, data).json() 

630 

631 columns = result.get("columns", []) 

632 if not columns: 

633 # No result columns 

634 return Table(rows=[]) 

635 

636 rows = result.get("data", []) 

637 if not rows: 

638 # No result rows 

639 return Table(names=columns) 

640 

641 return Table(rows=rows, names=columns) 

642 

643 def schema( 

644 self, instrument: str | None = None, table: str | None = None 

645 ) -> dict[str, tuple[str, str]] | list[str]: 

646 """Retrieve information about ConsDB. 

647 

648 If ``instrument`` and ``table`` are given, return the schema of a 

649 fixed metadata table in ConsDB. 

650 

651 If only ``instrument`` is given, return the names of all tables 

652 for that instrument. 

653 

654 If no arguments are given, return the names of all instruments. 

655 

656 Parameters 

657 ---------- 

658 instrument : `str`, optional 

659 Name of the instrument (e.g. ``LATISS``). 

660 table : `str`, optional 

661 Name of the table to insert into. 

662 

663 Returns 

664 ------- 

665 info : `list` [ `str` ] or `dict` [ `str`, `tuple` [ `str`, `str` ] ] 

666 A list of instrument strings or table names, or else a dict of 

667 columns with values that are tuples containing a data type string 

668 and a documentation string. 

669 

670 Raises 

671 ------ 

672 ValueError 

673 Raised if only ``table`` is given. 

674 requests.RequestException 

675 Raised if any kind of connection error occurs. 

676 requests.HTTPError 

677 Raised if a non-successful status is returned. 

678 

679 Notes 

680 ----- 

681 Fixed metadata data types may use the full database vocabulary, 

682 unlike flexible metadata data types. 

683 """ 

684 if instrument is None: 

685 if table is not None: 

686 raise ValueError("Must specify instrument if table is given") 

687 url = _urljoin(self.url, "schema") 

688 elif table is None: 

689 url = _urljoin(self.url, "schema", quote(instrument)) 

690 else: 

691 url = _urljoin(self.url, "schema", quote(instrument), quote(table)) 

692 result = self._handle_get(url) 

693 if instrument is not None and table is not None: 

694 return {key: (str(value[0]), str(value[1])) for key, value in result.items()} 

695 else: 

696 return [str(value) for value in result]