Coverage for python / lsst / daf / butler / registry / connectionString.py: 26%

1# This file is part of daf_butler. 

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 software is dual licensed under the GNU General Public License and also 

10# under a 3-clause BSD license. Recipients may choose which of these licenses 

11# to use; please see the files gpl-3.0.txt and/or bsd_license.txt, 

12# respectively. If you choose the GPL option then the following text applies 

13# (but note that there is still no warranty even if you opt for BSD instead): 

14# 

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

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

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

18# (at your option) any later version. 

19# 

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

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

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

23# GNU General Public License for more details. 

24# 

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

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

27 

28from __future__ import annotations 

29 

30__all__ = ("ConnectionStringFactory",) 

31 

32from typing import TYPE_CHECKING 

33 

34from sqlalchemy.engine import url 

35 

36from lsst.utils.db_auth import DbAuth, DbAuthNotFoundError 

37 

38if TYPE_CHECKING: 

39 from ._config import RegistryConfig 

40 

41 

42class ConnectionStringFactory: 

43 """Factory for `sqlalchemy.engine.url.URL` instances. 

44 

45 The factory constructs a connection string URL object by parsing the 

46 connection string, the 'db' key in the registry configuration. 

47 Username, password, host, port or database can be specified as keys in the 

48 config explicitly. If username or password are missing a matching DB is 

49 found in the credentials file pointed to by `DB_AUTH_ENVVAR` or 

50 `DB_AUTH_PATH` values. 

51 """ 

52 

53 keys = ("username", "password", "host", "port", "database") 

54 

55 @classmethod 

56 def fromConfig(cls, registryConfig: RegistryConfig, *, db_auth_path: str | None = None) -> url.URL: 

57 """Parse the `db`, and, if they exist, username, password, host, port 

58 and database keys from the given config. 

59 

60 If no username and password are found in the connection string, or in 

61 the config, they are retrieved from a file at `DB_AUTH_PATH` or 

62 `DB_AUTH_ENVVAR`. Sqlite dialect does not require a password. 

63 

64 The `db` key value of the given config specifies the default connection 

65 string, such that if no additional connection string parameters are 

66 provided or retrieved, the `db` key is returned unmodified. 

67 

68 Parameters 

69 ---------- 

70 registryConfig : `RegistryConfig` 

71 Registry configuration. 

72 db_auth_path : `str`, optional 

73 An alternative location of DbAuth file, mostly useful for tests. 

74 

75 Returns 

76 ------- 

77 connectionString : `sqlalchemy.engine.url.URL` 

78 URL object representing the connection string. 

79 

80 Raises 

81 ------ 

82 DbAuthPermissionsError 

83 If the credentials file has incorrect permissions. 

84 DbAuthError 

85 A problem occurred when retrieving DB authentication. 

86 

87 Notes 

88 ----- 

89 Matching requires dialect, host and database keys. If dialect is not 

90 specified in the db string and host and database keys are not found in 

91 the db string, or as explicit keys in the config, the matcher returns 

92 an unchanged connection string. 

93 Insufficiently specified connection strings are interpreted as an 

94 indication that a 3rd party authentication mechanism, such as Oracle 

95 Wallet, is being used and therefore are left unmodified. 

96 """ 

97 # this import can not live on the top because of circular import issue 

98 from ._config import RegistryConfig 

99 

100 regConf = RegistryConfig(registryConfig) 

101 conStr = url.make_url(regConf["db"]) 

102 

103 for key in cls.keys: 

104 if getattr(conStr, key) is None: 

105 # check for SQLAlchemy >= 1.4 

106 if hasattr(conStr, "set"): 

107 conStr = conStr.set(**{key: regConf.get(key)}) 

108 else: 

109 # SQLAlchemy 1.3 or earlier, mutate in place 

110 setattr(conStr, key, regConf.get(key)) 

111 

112 # Allow 3rd party authentication mechanisms by assuming connection 

113 # string is correct when we can not recognize (dialect, host, database) 

114 # matching keys. 

115 if any((conStr.drivername is None, conStr.host is None, conStr.database is None)): 

116 return conStr 

117 

118 # Ignore when credentials are not set up, or when no matches are found 

119 try: 

120 dbAuth = DbAuth(db_auth_path) 

121 auth = dbAuth.getAuth( 

122 conStr.drivername, conStr.username, conStr.host, conStr.port, conStr.database 

123 ) 

124 except DbAuthNotFoundError: 

125 # credentials file doesn't exist or no matches were found 

126 pass 

127 else: 

128 # only assign auth when *no* errors were raised 

129 conStr = conStr.set(username=auth[0], password=auth[1]) 

130 

131 return conStr