Coverage for python / lsst / dax / apdb / legacy_config.py: 77%

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 

22"""Module containing old pex_config-based configuration class. 

23 

24The main purpose of this class is to support conversion of the old 

25configuration files to the new pydantic-based format. It will be removed 

26after deprecation release 29. 

27""" 

28 

29from __future__ import annotations 

30 

31import os 

32from typing import TYPE_CHECKING 

33 

34from lsst.pex.config import Config, Field 

35 

36if TYPE_CHECKING: 

37 from .config import ApdbConfig as NewApdbConfig 

38 

39 

40def _data_file_name(basename: str) -> str: 

41 """Return path name of a data file in sdm_schemas package.""" 

42 return os.path.join("${SDM_SCHEMAS_DIR}", "yml", basename) 

43 

44 

45class ApdbConfig(Config): 

46 """Part of Apdb configuration common to all implementations.""" 

47 

48 read_sources_months = Field[int](doc="Number of months of history to read from DiaSource", default=12) 

49 read_forced_sources_months = Field[int]( 

50 doc="Number of months of history to read from DiaForcedSource", default=12 

51 ) 

52 schema_file = Field[str]( 

53 doc="Location of (YAML) configuration file with standard schema", default=_data_file_name("apdb.yaml") 

54 ) 

55 schema_name = Field[str](doc="Name of the schema in YAML configuration file.", default="ApdbSchema") 

56 extra_schema_file = Field[str]( 

57 doc=( 

58 "Location of (YAML) configuration file with extra schema, " 

59 "definitions in this file are merged with the definitions in " 

60 "'schema_file', extending or replacing parts of the schema." 

61 ), 

62 default=None, 

63 optional=True, 

64 deprecated="This field is deprecated, its value is not used.", 

65 ) 

66 use_insert_id = Field[bool]( 

67 doc=( 

68 "If True, make and fill additional tables used for replication. " 

69 "Databases created with earlier versions of APDB may not have these tables, " 

70 "and corresponding methods will not work for them." 

71 ), 

72 default=False, 

73 ) 

74 replica_chunk_seconds = Field[int]( 

75 default=600, 

76 doc="Time extent for replica chunks, new chunks are created every specified number of seconds.", 

77 ) 

78 

79 def to_model(self) -> NewApdbConfig: 

80 """Convert pex_config configuration to a new pydantic model.""" 

81 raise NotImplementedError() 

82 

83 @classmethod 

84 def legacy_load(cls, config: bytes) -> ApdbConfig: 

85 """Load legacy configuration from pex_config configuration file. 

86 

87 Parameters 

88 ---------- 

89 config : `bytes` 

90 Configuration data. 

91 

92 Returns 

93 ------- 

94 config : `ApdbConfig` 

95 Legacy configuration instance. 

96 """ 

97 # As old configuration classes have been moved, we need to rewrite 

98 # their locations in the config. 

99 config = config.replace(b"lsst.dax.apdb.sql.apdbSql", b"lsst.dax.apdb.sql.legacy_config") 

100 config = config.replace(b"lsst.dax.apdb.apdbSql", b"lsst.dax.apdb.sql.legacy_config") 

101 config = config.replace( 

102 b"lsst.dax.apdb.cassandra.apdbCassandra", b"lsst.dax.apdb.cassandra.legacy_config" 

103 ) 

104 config = config.replace(b"lsst.dax.apdb.apdbCassandra", b"lsst.dax.apdb.cassandra.legacy_config") 

105 return Config._fromPython(config.decode())