Coverage for python / lsst / dax / apdb / sql / apdbMetadataSql.py: 20%

1# This file is part of dax_apdb. 

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 

22from __future__ import annotations 

23 

24__all__ = ["ApdbMetadataSql"] 

25 

26from collections.abc import Generator 

27 

28import sqlalchemy 

29 

30from ..apdbMetadata import ApdbMetadata 

31 

32 

33class ApdbMetadataSql(ApdbMetadata): 

34 """Implementation of `ApdbMetadata` for SQL backend. 

35 

36 Parameters 

37 ---------- 

38 engine : `sqlalchemy.engine.Engine` 

39 Database access engine. 

40 table : `sqlalchemy.schema.Table` 

41 Database table holding metadata. 

42 """ 

43 

44 def __init__(self, engine: sqlalchemy.engine.Engine, table: sqlalchemy.schema.Table): 

45 self._engine = engine 

46 self._table = table 

47 

48 def get(self, key: str, default: str | None = None) -> str | None: 

49 # Docstring is inherited. 

50 sql = sqlalchemy.sql.select(self._table.columns.value).where(self._table.columns.name == key) 

51 with self._engine.begin() as conn: 

52 result = conn.execute(sql) 

53 value = result.scalar() 

54 if value is not None: 

55 return value 

56 return default 

57 

58 def set(self, key: str, value: str, *, force: bool = False) -> None: 

59 # Docstring is inherited. 

60 if not key or not value: 

61 raise ValueError("name and value cannot be empty") 

62 try: 

63 insert = sqlalchemy.sql.insert(self._table).values(name=key, value=value) 

64 with self._engine.begin() as conn: 

65 conn.execute(insert) 

66 except sqlalchemy.exc.IntegrityError as exc: 

67 # Try to update if it exists. 

68 if not force: 

69 raise KeyError(f"Metadata key {key!r} already exists") from exc 

70 update = ( 

71 sqlalchemy.sql.update(self._table).where(self._table.columns.name == key).values(value=value) 

72 ) 

73 with self._engine.begin() as conn: 

74 result = conn.execute(update) 

75 if result.rowcount != 1: 

76 raise RuntimeError(f"Metadata update failed unexpectedly, count={result.rowcount}") 

77 

78 def delete(self, key: str) -> bool: 

79 # Docstring is inherited. 

80 stmt = sqlalchemy.sql.delete(self._table).where(self._table.columns.name == key) 

81 with self._engine.begin() as conn: 

82 result = conn.execute(stmt) 

83 return result.rowcount > 0 

84 

85 def items(self) -> Generator[tuple[str, str], None, None]: 

86 # Docstring is inherited. 

87 stmt = sqlalchemy.sql.select(self._table.columns.name, self._table.columns.value) 

88 with self._engine.begin() as conn: 

89 result = conn.execute(stmt) 

90 for row in result: 

91 yield row._tuple() 

92 

93 def empty(self) -> bool: 

94 # Docstring is inherited. 

95 stmt = sqlalchemy.sql.select(sqlalchemy.sql.func.count()).select_from(self._table) 

96 with self._engine.begin() as conn: 

97 result = conn.execute(stmt) 

98 count = result.scalar() 

99 return count == 0