Coverage for faststream / _internal / cli / utils / logs.py: 46%

1import json (empty)

2import logging (empty)

3import logging.config (empty)

4from collections import defaultdict (empty)

5from enum import Enum (empty)

6from pathlib import Path (empty)

7from typing import TYPE_CHECKING, Any (empty)

8 

9import typer (empty)

10 

11from faststream.exceptions import INSTALL_TOML, INSTALL_YAML (empty)

12 

13if TYPE_CHECKING: (empty)

14 from faststream._internal.application import Application 

15 

16 

17class LogLevels(str, Enum): (empty)

18 """A class to represent log levels. 

19 

20 Attributes: 

21 critical : critical log level 

22 error : error log level 

23 warning : warning log level 

24 info : info log level 

25 debug : debug log level 

26 """ 

27 

28 critical = "critical" (empty)

29 fatal = "fatal" (empty)

30 error = "error" (empty)

31 warning = "warning" (empty)

32 warn = "warn" (empty)

33 info = "info" (empty)

34 debug = "debug" (empty)

35 notset = "notset" (empty)

36 

37 

38LOG_LEVELS: defaultdict[str, int] = defaultdict( (empty)

39 lambda: logging.INFO, 

40 critical=logging.CRITICAL, 

41 fatal=logging.FATAL, 

42 error=logging.ERROR, 

43 warning=logging.WARNING, 

44 warn=logging.WARNING, 

45 info=logging.INFO, 

46 debug=logging.DEBUG, 

47 notset=logging.NOTSET, 

48) 

49 

50 

51class LogFiles(str, Enum): (empty)

52 """The class to represent supported log configuration files.""" 

53 

54 json = ".json" (empty)

55 yaml = ".yaml" (empty)

56 yml = ".yml" (empty)

57 toml = ".toml" (empty)

58 

59 

60def get_log_level(level: LogLevels | str | int) -> int: (empty)

61 """Get the log level. 

62 

63 Args: 

64 level: The log level to get. Can be an integer, a LogLevels enum value, or a string. 

65 

66 Returns: 

67 The log level as an integer. 

68 

69 """ 

70 if isinstance(level, int): 70 ↛ 71line 70 didn't jump to line 71 because the condition on line 70 was never true(empty)

71 return level 

72 

73 if isinstance(level, LogLevels): (empty)

74 return LOG_LEVELS[level.value] (empty)

75 

76 if isinstance(level, str): # pragma: no branch (empty)

77 return LOG_LEVELS[level.lower()] (empty)

78 

79 return None 

80 

81 

82def set_log_level(level: int, app: "Application") -> None: (empty)

83 """Sets the log level for an application.""" 

84 if app.logger and getattr(app.logger, "setLevel", None): (empty)

85 app.logger.setLevel(level) # type: ignore[attr-defined] (empty)

86 

87 for broker in app.brokers: (empty)

88 broker.config.logger.set_level(level) (empty)

89 

90 

91def _get_json_config(file: Path) -> dict[str, Any] | Any: (empty)

92 """Parse json config file to dict.""" 

93 with file.open("r", encoding="utf-8") as config_file: 

94 return json.load(config_file) 

95 

96 

97def _get_yaml_config(file: Path) -> dict[str, Any] | Any: (empty)

98 """Parse yaml config file to dict.""" 

99 try: 

100 import yaml 

101 except ImportError as e: 

102 typer.echo(INSTALL_YAML, err=True) 

103 raise typer.Exit(1) from e 

104 

105 with file.open("r", encoding="utf-8") as config_file: 

106 return yaml.safe_load(config_file) 

107 

108 

109def _get_toml_config(file: Path) -> dict[str, Any] | Any: (empty)

110 """Parse toml config file to dict.""" 

111 try: 

112 import tomllib 

113 except ImportError: 

114 try: 

115 import tomli as tomllib 

116 except ImportError as e: 

117 typer.echo(INSTALL_TOML, err=True) 

118 raise typer.Exit(1) from e 

119 

120 with file.open("rb") as config_file: 

121 return tomllib.load(config_file) 

122 

123 

124def _get_log_config(file: Path) -> dict[str, Any] | Any: (empty)

125 """Read dict config from file.""" 

126 if not file.exists(): 

127 msg = f"File {file} specified to --log-config not found" 

128 raise ValueError(msg) 

129 

130 file_format = file.suffix 

131 

132 if file_format == LogFiles.json: 

133 logging_config = _get_json_config(file) 

134 elif file_format in {LogFiles.yaml, LogFiles.yml}: 

135 logging_config = _get_yaml_config(file) 

136 elif file_format == LogFiles.toml: 

137 logging_config = _get_toml_config(file) 

138 else: 

139 msg = f"Format {file_format} specified to --log-config file is not supported" 

140 raise ValueError(msg) 

141 

142 return logging_config 

143 

144 

145def set_log_config(file: Path) -> None: (empty)

146 """Set the logging config from file.""" 

147 configuration = _get_log_config(file) 

148 logging.config.dictConfig(configuration)