Coverage for fastblocks / main.py: 0%

1import typing as t 

2from contextlib import suppress 

3from pathlib import Path 

4 

5from acb import ensure_registration, register_pkg 

6from acb.adapters import register_adapters, root_path 

7from acb.depends import depends 

8 

9_app_instance = None 

10_logger_instance = None 

11 

12 

13async def get_app() -> t.Any: 

14 global _app_instance, _logger_instance 

15 if _app_instance is None: 

16 current_dir = Path.cwd() 

17 fastblocks_pkg_dir = Path(__file__).parent 

18 fastblocks_root_dir = fastblocks_pkg_dir.parent 

19 is_dev_mode = current_dir == fastblocks_pkg_dir or ( 

20 current_dir == fastblocks_root_dir 

21 and (current_dir / "pyproject.toml").exists() 

22 ) 

23 if is_dev_mode: 

24 msg = ( 

25 "FastBlocks cannot be run from its own package directory. " 

26 "Please run from your application directory." 

27 ) 

28 raise RuntimeError( 

29 msg, 

30 ) 

31 try: 

32 register_pkg() 

33 except Exception as e: 

34 msg = f"Failed to register FastBlocks adapters: {e}" 

35 raise RuntimeError(msg) from e 

36 try: 

37 await ensure_registration() 

38 except Exception as e: 

39 msg = f"Failed to register packages: {e}" 

40 raise RuntimeError(msg) from e 

41 with suppress(Exception): 

42 await register_adapters(root_path) 

43 try: 

44 _app_instance = await depends.get("app") 

45 except Exception as e: 

46 msg = f"Failed to get app adapter: {e}. Make sure adapters are properly registered and configured." 

47 raise RuntimeError( 

48 msg, 

49 ) from e 

50 try: 

51 _logger_instance = await depends.get("logger") 

52 except Exception as e: 

53 import logging 

54 

55 _logger_instance = logging.getLogger("fastblocks") 

56 _logger_instance.warning( 

57 f"Failed to get logger adapter, using fallback: {e}", 

58 ) 

59 return _app_instance 

60 

61 

62def get_logger() -> t.Any: 

63 return _logger_instance 

64 

65 

66class LazyApp: 

67 def __getattr__(self, name: str) -> t.Any: 

68 if _app_instance is None: 

69 msg = "FastBlocks app has not finished initializing" 

70 raise AttributeError(msg) 

71 return getattr(_app_instance, name) 

72 

73 async def __call__(self, scope: t.Any, receive: t.Any, send: t.Any) -> None: 

74 app = await get_app() 

75 await app(scope, receive, send) 

76 

77 

78class LazyLogger: 

79 def __getattr__(self, name: str) -> t.Any: 

80 if _logger_instance is None: 

81 msg = "FastBlocks logger has not finished initializing" 

82 raise AttributeError(msg) 

83 return getattr(_logger_instance, name) 

84 

85 def __call__(self, *args: t.Any, **kwargs: t.Any) -> t.Any: 

86 logger = _logger_instance 

87 if callable(logger): 

88 return logger(*args, **kwargs) 

89 return logger 

90 

91 

92app = LazyApp() 

93logger = LazyLogger()