Coverage for src/lexigram/web/server/runner.py: 22%

1"""ASGI server execution. Runs the configured web application.""" 

2 

3from __future__ import annotations 

4 

5from typing import Any 

6 

7from lexigram.logging import get_logger 

8 

9logger = get_logger(__name__) 

10 

11 

12def _to_import_string(app: Any) -> str | None: 

13 """Derive an import string (``module:attr``) from an ASGI app instance. 

14 

15 Granian requires a string import path for multiprocessing workers. 

16 """ 

17 try: 

18 module = app.__class__.__module__ 

19 if module == "builtins" or module.startswith("starlette."): 

20 return None 

21 for candidate in ("app", "application", "asgi"): 

22 if hasattr(app.__class__, candidate): 

23 return f"{module}:{candidate}" 

24 return f"{module}:{app.__class__.__name__}" 

25 except (AttributeError, TypeError): 

26 return None 

27 

28 

29def run_server( 

30 app: Any, 

31 host: str = "127.0.0.1", 

32 port: int = 8000, 

33 **kwargs: Any, 

34) -> None: 

35 """Run the web application. 

36 

37 Uses Granian when ``app`` is a string import path (multiprocess-safe). 

38 Falls back to Uvicorn when ``app`` is an instance (Granian cannot accept 

39 instances directly). 

40 

41 Args: 

42 app: ASGI application instance or import string (``"module:attr"``). 

43 host: Bind address (default: 127.0.0.1). 

44 port: Bind port (default: 8000). 

45 **kwargs: Additional arguments (passed to Uvicorn config). 

46 

47 Raises: 

48 ImportError: If neither Granian nor Uvicorn is installed. 

49 """ 

50 if isinstance(app, str): 

51 _run_granian(app, host, port, **kwargs) 

52 return 

53 

54 _run_uvicorn(app, host, port, **kwargs) 

55 

56 

57def _run_granian( 

58 app: str, 

59 host: str, 

60 port: int, 

61 **kwargs: Any, 

62) -> None: 

63 """Run via Granian (multiprocess).""" 

64 try: 

65 from granian import Granian 

66 from granian.constants import Interfaces 

67 except ImportError as e: 

68 raise ImportError( 

69 "Granian is not installed. Install 'granian' to use run_server() with string apps.", 

70 ) from e 

71 

72 logger.info("starting_granian_server", host=host, port=port, kwargs=kwargs) 

73 server = Granian( 

74 app, 

75 address=host, 

76 port=port, 

77 interface=Interfaces.ASGI, 

78 **kwargs, 

79 ) 

80 server.serve() 

81 

82 

83async def run_server_async( 

84 app: Any, 

85 host: str = "127.0.0.1", 

86 port: int = 8000, 

87 **kwargs: Any, 

88) -> None: 

89 """Run the web application asynchronously. 

90 

91 Uses Uvicorn (supports both string paths and app instances). Prefer 

92 :func:`run_server` for production deployments with Granian. 

93 

94 Args: 

95 app: ASGI application instance or import string. 

96 host: Bind address. 

97 port: Bind port. 

98 **kwargs: Additional arguments passed to Uvicorn config. 

99 """ 

100 import uvicorn 

101 

102 logger.info("starting_uvicorn_server", host=host, port=port, kwargs=kwargs) 

103 config = uvicorn.Config(app, host=host, port=port, **kwargs) 

104 server = uvicorn.Server(config) 

105 await server.serve() 

106 

107 

108def _run_uvicorn( 

109 app: Any, 

110 host: str, 

111 port: int, 

112 **kwargs: Any, 

113) -> None: 

114 """Run via Uvicorn (supports app instances).""" 

115 import asyncio 

116 

117 import uvicorn 

118 

119 logger.info("starting_uvicorn_server", host=host, port=port, kwargs=kwargs) 

120 config = uvicorn.Config(app, host=host, port=port, **kwargs) 

121 server = uvicorn.Server(config) 

122 loop = asyncio.new_event_loop() 

123 asyncio.set_event_loop(loop) 

124 loop.run_until_complete(server.serve()) 

125 

126 

127__all__ = ["run_server", "run_server_async"]