feat(asgi): hypercorn

exam_hypercorn.py

@@ -1,9 +1,8 @@
 from creart import it
 from launart import Launart
 
-from graia.amnesia.builtins.asgi import HypercornASGIService
+from graia.amnesia.builtins.asgi import UvicornASGIService
 
 manager = it(Launart)
-manager.add_component(serv := HypercornASGIService("127.0.0.1", 5333))
-serv.patch_logger()
+manager.add_component(serv := UvicornASGIService("127.0.0.1", 5333, patch_logger=True))
 manager.launch_blocking()

pyproject.toml

@@ -26,12 +26,12 @@ aiohttp = [
 sqla = [
     "sqlalchemy>=2.0.25",
 ]
-uvi = [
-    "uvicorn>=0.35.0",
-]
-hyper = [
+hypercorn = [
     "hypercorn>=0.17.3",
 ]
+uvicorn = [
+    "uvicorn>=0.35.0",
+]
 
 [build-system]
 requires = ["pdm-backend"]

src/graia/amnesia/builtins/asgi/__init__.py

@@ -1,24 +1,21 @@
 try:
-    from .uvicorn import UvicornASGIService as UvicornASGIService
+    from .uvicorn import UvicornASGIService as _UvicornASGIService
 except ImportError:
-    UvicornASGIService = None
+    _UvicornASGIService = None
 
 try:
-    from .hypercorn import HypercornASGIService as HypercornASGIService
+    from .hypercorn import HypercornASGIService as _HypercornASGIService
 except ImportError:
-    HypercornASGIService = None
+    _HypercornASGIService = None
 
 
 def __getattr__(name):
     if name == "UvicornASGIService":
-        if UvicornASGIService is None:
-            raise ImportError("Please install `uvicorn` first. Install with `pip install graia-amnesia[uvi]`")
-        return UvicornASGIService
+        if _UvicornASGIService is None:
+            raise ImportError("Please install `uvicorn` first. Install with `pip install graia-amnesia[uvicorn]`")
+        return _UvicornASGIService
     if name == "HypercornASGIService":
-        if HypercornASGIService is None:
-            raise ImportError("Please install `hypercorn` first. Install with `pip install graia-amnesia[hyper]`")
-        return HypercornASGIService
+        if _HypercornASGIService is None:
+            raise ImportError("Please install `hypercorn` first. Install with `pip install graia-amnesia[hypercorn]`")
+        return _HypercornASGIService
     raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
-
-
-__all__ = ["UvicornASGIService", "HypercornASGIService"]

src/graia/amnesia/builtins/asgi/__init__.pyi

@@ -0,0 +1,2 @@
+from .hypercorn import HypercornASGIService as HypercornASGIService
+from .uvicorn import UvicornASGIService as UvicornASGIService

src/graia/amnesia/builtins/asgi/common.py

@@ -0,0 +1,19 @@
+async def empty_asgi_handler(scope, receive, send):
+    if scope["type"] == "lifespan":
+        while True:
+            message = await receive()
+            if message["type"] == "lifespan.startup":
+                await send({"type": "lifespan.startup.complete"})
+                return
+            elif message["type"] == "lifespan.shutdown":
+                await send({"type": "lifespan.shutdown.complete"})
+                return
+
+    await send(
+        {
+            "type": "http.response.start",
+            "status": 404,
+            "headers": [(b"content-length", b"0")],
+        }
+    )
+    await send({"type": "http.response.body"})

src/graia/amnesia/builtins/asgi/hypercorn.py

@@ -6,46 +6,20 @@
 from ssl import VerifyFlags, VerifyMode
 from typing import Any, List, Optional, TypedDict, Union
 
+from hypercorn.asyncio import serve
+from hypercorn.config import Config
+from hypercorn.logging import Logger
+from hypercorn.typing import ResponseSummary, WWWScope
 from launart import Launart, Service
 from launart.status import Phase
 from launart.utilles import any_completed
 from loguru import logger
 
-try:
-    from hypercorn.asyncio import serve
-    from hypercorn.config import Config
-    from hypercorn.logging import Logger
-    from hypercorn.typing import ResponseSummary, WWWScope
-except ImportError:
-    raise ImportError(
-        "dependency 'hypercorn' is required for asgi-service:hypercorn\nplease install it or install 'graia-amnesia[hyper]'"
-    )
-
 from . import asgitypes
+from .common import empty_asgi_handler
 from .middleware import DispatcherMiddleware
 
 
-async def _empty_asgi_handler(scope, receive, send):
-    if scope["type"] == "lifespan":
-        while True:
-            message = await receive()
-            if message["type"] == "lifespan.startup":
-                await send({"type": "lifespan.startup.complete"})
-                return
-            elif message["type"] == "lifespan.shutdown":
-                await send({"type": "lifespan.shutdown.complete"})
-                return
-
-    await send(
-        {
-            "type": "http.response.start",
-            "status": 404,
-            "headers": [(b"content-length", b"0")],
-        }
-    )
-    await send({"type": "http.response.body"})
-
-
 class HypercornOptions(TypedDict, total=False):
     insecure_bind: Union[List[str], str]
     """default: []"""
@@ -195,11 +169,14 @@ def __init__(
         port: int,
         mounts: dict[str, asgitypes.ASGI3Application] | None = None,
         options: HypercornOptions | None = None,
+        patch_logger: bool = True,
     ):
         self.host = host
         self.port = port
-        self.middleware = DispatcherMiddleware(mounts or {"\0\0\0": _empty_asgi_handler})
+        self.middleware = DispatcherMiddleware(mounts or {"\0\0\0": empty_asgi_handler})
         self.options = options or {}
+        if patch_logger:
+            self.options["logger_class"] = LoguruLogger  # type: ignore
         super().__init__()
 
     @property
@@ -231,6 +208,3 @@ async def launch(self, manager: Launart) -> None:
                 await asyncio.wait_for(serve_task, timeout=5.0)
             except asyncio.TimeoutError:
                 logger.warning("timeout, force exit hypercorn server...")
-
-    def patch_logger(self) -> None:
-        self.options["logger_class"] = LoguruLogger  # type: ignore

src/graia/amnesia/builtins/asgi/uvicorn.py

@@ -9,41 +9,15 @@
 from launart.status import Phase
 from launart.utilles import any_completed
 from loguru import logger
+from uvicorn import Config, Server
 from uvicorn.config import LOG_LEVELS, HTTPProtocolType, LifespanType, LoopSetupType, WSProtocolType
 
-try:
-    from uvicorn import Config, Server
-except ImportError:
-    raise ImportError(
-        "dependency 'uvicorn' is required for asgi-service:uvicorn\nplease install it or install 'graia-amnesia[uvi]'"
-    )
-
 from ..utils import LoguruHandler
 from . import asgitypes
+from .common import empty_asgi_handler
 from .middleware import DispatcherMiddleware
 
 
-async def _empty_asgi_handler(scope, receive, send):
-    if scope["type"] == "lifespan":
-        while True:
-            message = await receive()
-            if message["type"] == "lifespan.startup":
-                await send({"type": "lifespan.startup.complete"})
-                return
-            elif message["type"] == "lifespan.shutdown":
-                await send({"type": "lifespan.shutdown.complete"})
-                return
-
-    await send(
-        {
-            "type": "http.response.start",
-            "status": 404,
-            "headers": [(b"content-length", b"0")],
-        }
-    )
-    await send({"type": "http.response.body"})
-
-
 class WithoutSigHandlerServer(Server):
     def install_signal_handlers(self) -> None:
         pass
@@ -153,10 +127,12 @@ def __init__(
         port: int,
         mounts: dict[str, asgitypes.ASGI3Application] | None = None,
         options: UvicornOptions | None = None,
+        patch_logger: bool = True,
     ):
         self.host = host
         self.port = port
-        self.middleware = DispatcherMiddleware(mounts or {"\0\0\0": _empty_asgi_handler})
+        self.patch_logger = patch_logger
+        self.middleware = DispatcherMiddleware(mounts or {"\0\0\0": empty_asgi_handler})
         self.options: UvicornOptions = options or {}
         super().__init__()
 
@@ -173,6 +149,8 @@ async def launch(self, manager: Launart) -> None:
             self.server = WithoutSigHandlerServer(
                 Config(self.middleware, host=self.host, port=self.port, factory=False, **self.options)
             )
+            if self.patch_logger:
+                self._patch_logger()
             serve_task = asyncio.create_task(self.server.serve())
 
         async with self.stage("blocking"):
@@ -185,7 +163,7 @@ async def launch(self, manager: Launart) -> None:
             if not serve_task.done():
                 logger.warning("timeout, force exit uvicorn server...")
 
-    def patch_logger(self) -> None:
+    def _patch_logger(self) -> None:
         log_level = 20
         if "log_level" in self.options and (_log_level := self.options["log_level"]) is not None:
             if isinstance(_log_level, str):

src/graia/amnesia/builtins/utils.py

@@ -8,8 +8,6 @@ class LoguruHandler(logging.Handler):
     def emit(self, record: logging.LogRecord):
         try:
             level = logger.level(record.levelname).name
-            if record.levelno <= logging.INFO:
-                level = {"DEBUG": "TRACE", "INFO": "DEBUG"}.get(level, level)
         except ValueError:
             level = record.levelno
 
@@ -18,7 +16,9 @@ def emit(self, record: logging.LogRecord):
             frame = frame.f_back
             depth += 1
 
-        logger.opt(depth=depth, exception=record.exc_info).log(level, record.getMessage())
+        logger.opt(depth=depth, exception=record.exc_info).patch(lambda rec: rec.update(name=record.name)).log(
+            level, record.getMessage()
+        )
 
 
 def get_subclasses(cls):