Coverage for python/felis/db/dialects.py: 81%

1"""Get SQLAlchemy dialects and their type modules.""" 

2 

3# This file is part of felis. 

4# 

5# Developed for the LSST Data Management System. 

6# This product includes software developed by the LSST Project 

7# (https://www.lsst.org). 

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

9# for details of code ownership. 

10# 

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

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

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

14# (at your option) any later version. 

15# 

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

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

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

19# GNU General Public License for more details. 

20# 

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

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

23 

24from __future__ import annotations 

25 

26from collections.abc import Mapping 

27from types import MappingProxyType, ModuleType 

28 

29from sqlalchemy import dialects 

30from sqlalchemy.engine import Dialect 

31from sqlalchemy.engine.mock import create_mock_engine 

32 

33from .sqltypes import MYSQL, POSTGRES, SQLITE 

34 

35__all__ = ["get_dialect_module", "get_supported_dialects"] 

36 

37_DIALECT_NAMES = (MYSQL, POSTGRES, SQLITE) 

38"""List of supported dialect names. 

39 

40This list is used to create the dialect and module dictionaries. 

41""" 

42 

43 

44def _dialect(dialect_name: str) -> Dialect: 

45 """Create the SQLAlchemy dialect for the given name using a mock engine. 

46 

47 Parameters 

48 ---------- 

49 dialect_name 

50 The name of the dialect to create. 

51 

52 Returns 

53 ------- 

54 `~sqlalchemy.engine.Dialect` 

55 The SQLAlchemy dialect. 

56 """ 

57 return create_mock_engine(f"{dialect_name}://", executor=None).dialect 

58 

59 

60_DIALECTS = MappingProxyType({name: _dialect(name) for name in _DIALECT_NAMES}) 

61"""Dictionary of dialect names to SQLAlchemy dialects.""" 

62 

63 

64def get_supported_dialects() -> Mapping[str, Dialect]: 

65 """Get a dictionary of the supported SQLAlchemy dialects. 

66 

67 Returns 

68 ------- 

69 `dict` [ `str`, `~sqlalchemy.engine.Dialect`] 

70 A dictionary of the supported SQLAlchemy dialects. 

71 

72 Notes 

73 ----- 

74 The dictionary is keyed by the dialect name and the value is the SQLAlchemy 

75 dialect object. This function is intended as the primary interface for 

76 getting the supported dialects. 

77 """ 

78 return _DIALECTS 

79 

80 

81def _dialect_module(dialect_name: str) -> ModuleType: 

82 """Get the SQLAlchemy dialect module for the given name. 

83 

84 Parameters 

85 ---------- 

86 dialect_name 

87 The name of the dialect module to get from the SQLAlchemy package. 

88 """ 

89 return getattr(dialects, dialect_name) 

90 

91 

92_DIALECT_MODULES = MappingProxyType({name: _dialect_module(name) for name in _DIALECT_NAMES}) 

93"""Dictionary of dialect names to SQLAlchemy modules.""" 

94 

95 

96def get_dialect_module(dialect_name: str) -> ModuleType: 

97 """Get the SQLAlchemy dialect module for the given name. 

98 

99 Parameters 

100 ---------- 

101 dialect_name 

102 The name of the dialect module to get from the SQLAlchemy package. 

103 

104 Returns 

105 ------- 

106 `~types.ModuleType` 

107 The SQLAlchemy dialect module. 

108 

109 Raises 

110 ------ 

111 ValueError 

112 Raised if the dialect name is not supported. 

113 """ 

114 if dialect_name not in _DIALECT_MODULES: 

115 raise ValueError(f"Unsupported dialect: {dialect_name}") 

116 return _DIALECT_MODULES[dialect_name]