feat: 更多的 HTTP / ASGI 组件

.pre-commit-config.yaml

@@ -13,7 +13,7 @@ repos:
       - id: end-of-file-fixer
       - id: trailing-whitespace
   - repo: https://github.com/psf/black
-    rev: "25.1.0"
+    rev: "26.3.1"
     hooks:
       - id: black
   - repo: https://github.com/pycqa/isort

exam_aio.py

@@ -0,0 +1,48 @@
+import asyncio
+
+from creart import it
+from launart import Launart, Service
+from launart.status import Phase
+
+from graia.amnesia.builtins.http import request
+from graia.amnesia.builtins.http.aiohttp import AiohttpClientService
+
+# from graia.amnesia.builtins.http.httpx import HttpxClientService
+# from graia.amnesia.builtins.http.niquests import NiquestsClientService
+# from graia.amnesia.builtins.http.pyreqwest import PyReqwestClientService
+from graia.amnesia.builtins.http.httptypes import HTTPStatusError, Request
+
+manager = it(Launart)
+manager.add_component(AiohttpClientService())
+
+
+class MainService(Service):
+    id = "main"
+
+    @property
+    def required(self):
+        return set()
+
+    @property
+    def stages(self) -> set[Phase]:
+        return {"blocking"}
+
+    async def launch(self, manager: Launart) -> None:
+        async with self.stage("blocking"):
+            await asyncio.sleep(2)
+            resp = await request(Request("GET", "https://httpbin.org/get"))
+            print(resp.status_code)
+            print(await resp.aread())
+            print(await resp.ajson())
+            resp1 = await request(Request("GET", "https://httpbin.org/image"), stream=True)
+            async for chunk in resp1.aiter_bytes(1024):
+                print(f"chunk: {len(chunk)} bytes")
+            try:
+                async with await request(Request("GET", "https://httpbin.org/status/404")) as resp:
+                    resp.raise_for_status()
+            except HTTPStatusError as e:
+                print(e)
+
+
+manager.add_component(MainService())
+manager.launch_blocking()

exam_granian.py

@@ -0,0 +1,38 @@
+from creart import it
+from launart import Launart
+
+from graia.amnesia.builtins.asgi import GranianASGIService
+from graia.amnesia.builtins.asgi.granian import GranianOptions
+
+
+async def app(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": 200,
+            "headers": [[b"content-type", b"text/plain"]],
+        }
+    )
+    await send(
+        {
+            "type": "http.response.body",
+            "body": b"Hello, world!",
+        }
+    )
+
+
+manager = it(Launart)
+manager.add_component(
+    GranianASGIService("127.0.0.1", 5333, {"/": app}, options=GranianOptions(log_access=False), patch_logger=True)
+)
+manager.launch_blocking()

exam_hypercorn.py

@@ -2,7 +2,37 @@
 from launart import Launart
 
 from graia.amnesia.builtins.asgi import HypercornASGIService
+from graia.amnesia.builtins.asgi.hypercorn import HypercornOptions
+
+
+async def app(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": 200,
+            "headers": [[b"content-type", b"text/plain"]],
+        }
+    )
+    await send(
+        {
+            "type": "http.response.body",
+            "body": b"Hello, world!",
+        }
+    )
+
 
 manager = it(Launart)
-manager.add_component(HypercornASGIService("127.0.0.1", 5333, patch_logger=True))
+manager.add_component(
+    HypercornASGIService("127.0.0.1", 5333, {"/": app}, HypercornOptions(accesslog="-"), patch_logger=True)
+)
 manager.launch_blocking()

exam_uvicorn.py

@@ -0,0 +1,38 @@
+from creart import it
+from launart import Launart
+
+from graia.amnesia.builtins.asgi import UvicornASGIService
+from graia.amnesia.builtins.asgi.uvicorn import UvicornOptions
+
+
+async def app(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": 200,
+            "headers": [[b"content-type", b"text/plain"]],
+        }
+    )
+    await send(
+        {
+            "type": "http.response.body",
+            "body": b"Hello, world!",
+        }
+    )
+
+
+manager = it(Launart)
+manager.add_component(
+    UvicornASGIService("127.0.0.1", 5333, {"/": app}, options=UvicornOptions(access_log=False), patch_logger=True)
+)
+manager.launch_blocking()