Coverage for amqtt/utils.py: 80%

1from __future__ import annotations 

2 

3from importlib import import_module 

4import logging 

5from pathlib import Path 

6import secrets 

7import string 

8import sys 

9import typing 

10from typing import Any 

11 

12import yaml 

13 

14if typing.TYPE_CHECKING: 14 ↛ 15line 14 didn't jump to line 15 because the condition on line 14 was never true

15 from amqtt.session import Session 

16 

17logger = logging.getLogger(__name__) 

18 

19 

20def format_client_message( 

21 session: Session | None = None, 

22 address: str | None = None, 

23 port: int | None = None, 

24) -> str: 

25 """Format a client message for logging.""" 

26 if session: 

27 return f"(client id={session.client_id})" 

28 if address is not None and port is not None: 

29 return f"(client @={address}:{port})" 

30 return "(unknown client)" 

31 

32 

33def gen_client_id() -> str: 

34 """Generate a random client ID.""" 

35 gen_id = "amqtt/" 

36 

37 # Use secrets to generate a secure random client ID 

38 # Defining a valid set of characters for client ID generation 

39 valid_chars = string.ascii_letters + string.digits 

40 gen_id += "".join(secrets.choice(valid_chars) for _ in range(16)) 

41 return gen_id 

42 

43 

44def read_yaml_config(config_file: str | Path) -> dict[str, Any] | None: 

45 """Read a YAML configuration file.""" 

46 try: 

47 with Path(str(config_file)).open(encoding="utf-8") as stream: 

48 yaml_result: dict[str, Any] = yaml.full_load(stream) 

49 return yaml_result 

50 except yaml.YAMLError: 

51 logger.exception(f"Invalid config_file {config_file}") 

52 return None 

53 

54 

55def cached_import(module_path: str, class_name: str | None = None) -> Any: 

56 """Return cached import of a class from a module path (or retrieve, cache and then return).""" 

57 # Check whether module is loaded and fully initialized. 

58 if not ((module := sys.modules.get(module_path)) 

59 and (spec := getattr(module, "__spec__", None)) 

60 and getattr(spec, "_initializing", False) is False): 

61 module = import_module(module_path) 

62 if class_name: 62 ↛ 64line 62 didn't jump to line 64 because the condition on line 62 was always true

63 return getattr(module, class_name) 

64 return module 

65 

66 

67def import_string(dotted_path: str) -> Any: 

68 """Import a dotted module path. 

69 

70 Returns: 

71 attribute/class designated by the last name in the path 

72 

73 Raises: 

74 ImportError (if the import failed) 

75 

76 """ 

77 try: 

78 module_path, class_name = dotted_path.rsplit(".", 1) 

79 except ValueError as err: 

80 msg = f"{dotted_path} doesn't look like a module path" 

81 raise ImportError(msg) from err 

82 

83 try: 

84 return cached_import(module_path, class_name) 

85 except AttributeError as err: 

86 msg = f'Module "{module_path}" does not define a "{class_name}" attribute/class' 

87 

88 raise ImportError(msg) from err