hatch fmt on examples and tests/

Author: olethanh

examples/example_fastapi/main.py

@@ -8,7 +8,7 @@
 from datetime import datetime, timezone
 from os import listdir
 from pathlib import Path
-from typing import Any, Optional
+from typing import Any
 
 import aiohttp
 from aleph_message.models import (
@@ -128,8 +128,8 @@ async def read_aleph_messages() -> dict[str, MessagesResponse]:
 async def resolve_dns_hostname():
     """Check if DNS resolution is working."""
     hostname = "example.org"
-    ipv4: Optional[str] = None
-    ipv6: Optional[str] = None
+    ipv4: str | None = None
+    ipv6: str | None = None
 
     info = socket.getaddrinfo(hostname, 80, proto=socket.IPPROTO_TCP)
     if not info:
@@ -170,7 +170,7 @@ async def connect_ipv4():
         sock.settimeout(5)
         sock.connect((ipv4_host, 53))
         return {"result": True}
-    except socket.timeout:
+    except TimeoutError:
         logger.warning(f"Socket connection for host {ipv4_host} failed")
         return {"result": False}
 

examples/example_fastapi_1.py

@@ -1,5 +1,3 @@
-from typing import Optional
-
 from fastapi import FastAPI
 
 app = FastAPI()
@@ -11,10 +9,10 @@ def read_root():
 
 
 @app.get("/run/{item_id}")
-def read_item(item_id: str, q: Optional[str] = None):
+def read_item(item_id: str, q: str | None = None):
     return {"item_id": item_id, "q": q}
 
 
 @app.post("/run/{item_id}")
-def read_item_post(item_id: str, q: Optional[str] = None):
+def read_item_post(item_id: str, q: str | None = None):
     return {"item_id_post": item_id, "q": q}

runtimes/aleph-debian-12-python/init1.py

@@ -19,13 +19,11 @@
 import sys
 import traceback
 from collections.abc import AsyncIterable
-from contextlib import redirect_stdout
 from dataclasses import dataclass, field
 from enum import Enum
-from io import StringIO
 from os import system
 from shutil import make_archive
-from typing import Any, Literal, NewType, Optional, Union, cast
+from typing import Any, Literal, NewType, cast
 
 import aiohttp
 import msgpack
@@ -66,14 +64,14 @@ class ConfigurationPayload:
     code: bytes
     encoding: Encoding
     entrypoint: str
-    ip: Optional[str] = None
-    ipv6: Optional[str] = None
-    route: Optional[str] = None
-    ipv6_gateway: Optional[str] = None
+    ip: str | None = None
+    ipv6: str | None = None
+    route: str | None = None
+    ipv6_gateway: str | None = None
     dns_servers: list[str] = field(default_factory=list)
     volumes: list[Volume] = field(default_factory=list)
-    variables: Optional[dict[str, str]] = None
-    authorized_keys: Optional[list[str]] = None
+    variables: dict[str, str] | None = None
+    authorized_keys: list[str] | None = None
 
 
 @dataclass
@@ -107,19 +105,19 @@ def setup_hostname(hostname: str):
     system(f"hostname {hostname}")
 
 
-def setup_variables(variables: Optional[dict[str, str]]):
+def setup_variables(variables: dict[str, str] | None):
     if variables is None:
         return
     for key, value in variables.items():
         os.environ[key] = value
 
 
 def setup_network(
-    ipv4: Optional[str],
-    ipv6: Optional[str],
-    ipv4_gateway: Optional[str],
-    ipv6_gateway: Optional[str],
-    dns_servers: Optional[list[str]] = None,
+    ipv4: str | None,
+    ipv6: str | None,
+    ipv4_gateway: str | None,
+    ipv6_gateway: str | None,
+    dns_servers: list[str] | None = None,
 ):
     """Setup the system with info from the host."""
     dns_servers = dns_servers or []
@@ -188,9 +186,7 @@ def setup_volumes(volumes: list[Volume]):
     system("mount")
 
 
-async def wait_for_lifespan_event_completion(
-    application: ASGIApplication, event: Union[Literal["startup", "shutdown"]]
-):
+async def wait_for_lifespan_event_completion(application: ASGIApplication, event: Literal["startup", "shutdown"]):
     """
     Send the startup lifespan signal to the ASGI app.
     Specification: https://asgi.readthedocs.io/en/latest/specs/lifespan.html
@@ -295,7 +291,7 @@ async def setup_code(
     encoding: Encoding,
     entrypoint: str,
     interface: Interface,
-) -> Union[ASGIApplication, subprocess.Popen]:
+) -> ASGIApplication | subprocess.Popen:
     if interface == Interface.asgi:
         return await setup_code_asgi(code=code, encoding=encoding, entrypoint=entrypoint)
     elif interface == Interface.executable:
@@ -304,7 +300,7 @@ async def setup_code(
         raise ValueError("Invalid interface. This should never happen.")
 
 
-async def run_python_code_http(application: ASGIApplication, scope: dict) -> tuple[dict, dict, str, Optional[bytes]]:
+async def run_python_code_http(application: ASGIApplication, scope: dict) -> tuple[dict, dict, str, bytes | None]:
     logger.debug("Running code")
     # Execute in the same process, saves ~20ms than a subprocess
 
@@ -386,7 +382,7 @@ def show_loading():
     return headers, body
 
 
-async def run_executable_http(scope: dict) -> tuple[dict, dict, str, Optional[bytes]]:
+async def run_executable_http(scope: dict) -> tuple[dict, dict, str, bytes | None]:
     logger.debug("Calling localhost")
 
     tries = 0
@@ -413,7 +409,7 @@ async def run_executable_http(scope: dict) -> tuple[dict, dict, str, Optional[by
 async def process_instruction(
     instruction: bytes,
     interface: Interface,
-    application: Union[ASGIApplication, subprocess.Popen],
+    application: ASGIApplication | subprocess.Popen,
 ) -> AsyncIterable[bytes]:
     if instruction == b"halt":
         logger.info("Received halt command")
@@ -443,11 +439,11 @@ async def process_instruction(
         logger.debug("msgpack.loads )")
         payload = RunCodePayload(**msg_)
 
-        output: Optional[str] = None
+        output: str | None = None
         try:
             headers: dict
             body: dict
-            output_data: Optional[bytes]
+            output_data: bytes | None
 
             if interface == Interface.asgi:
                 application = cast(ASGIApplication, application)
@@ -540,7 +536,7 @@ async def main() -> None:
     setup_system(config)
 
     try:
-        app: Union[ASGIApplication, subprocess.Popen] = await setup_code(
+        app: ASGIApplication | subprocess.Popen = await setup_code(
             config.code, config.encoding, config.entrypoint, config.interface
         )
         client.send(msgpack.dumps({"success": True}))

tests/supervisor/test_execution.py

@@ -1,5 +1,4 @@
 import asyncio
-import json
 import logging
 from typing import Any
 

vm_connector/main.py

@@ -1,6 +1,5 @@
 import json
 import logging
-from typing import Optional
 
 import aiohttp
 from aleph_client.asynchronous import create_post
@@ -24,7 +23,7 @@ def read_root():
     return {"Server": "Aleph.im VM Connector"}
 
 
-async def get_latest_message_amend(ref: str, sender: str) -> Optional[dict]:
+async def get_latest_message_amend(ref: str, sender: str) -> dict | None:
     async with aiohttp.ClientSession() as session:
         url = f"{settings.API_SERVER}/api/v0/messages.json?msgType=STORE&sort_order=-1&refs={ref}&addresses={sender}"
         resp = await session.get(url)
@@ -36,7 +35,7 @@ async def get_latest_message_amend(ref: str, sender: str) -> Optional[dict]:
             return None
 
 
-async def get_message(hash_: str) -> Optional[dict]:
+async def get_message(hash_: str) -> dict | None:
     async with aiohttp.ClientSession() as session:
         url = f"{settings.API_SERVER}/api/v0/messages.json?hashes={hash_}"
         resp = await session.get(url)