Fix typing: Replace Dict->dict, List->list

Author: hoh

examples/example_fastapi/main.py

@@ -7,7 +7,7 @@
 from datetime import datetime
 from os import listdir
 from pathlib import Path
-from typing import Dict, Optional
+from typing import Optional
 
 import aiohttp
 from fastapi import FastAPI
@@ -84,7 +84,7 @@ async def check_lifespan():
 
 
 @app.get("/environ")
-async def environ() -> Dict[str, str]:
+async def environ() -> dict[str, str]:
     """List environment variables"""
     return dict(os.environ)
 

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

@@ -18,24 +18,14 @@
 import subprocess
 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,
-    AsyncIterable,
-    Dict,
-    List,
-    Literal,
-    NewType,
-    Optional,
-    Tuple,
-    Union,
-    cast,
-)
+from typing import Any, Literal, NewType, Optional, Union, cast
 
 import aiohttp
 import msgpack
@@ -80,15 +70,15 @@ class ConfigurationPayload:
     ipv6: Optional[str] = None
     route: Optional[str] = None
     ipv6_gateway: Optional[str] = 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
+    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
 
 
 @dataclass
 class RunCodePayload:
-    scope: Dict
+    scope: dict
 
 
 # Open a socket to receive instructions from the host
@@ -117,7 +107,7 @@ def setup_hostname(hostname: str):
     system(f"hostname {hostname}")
 
 
-def setup_variables(variables: Optional[Dict[str, str]]):
+def setup_variables(variables: Optional[dict[str, str]]):
     if variables is None:
         return
     for key, value in variables.items():
@@ -129,7 +119,7 @@ def setup_network(
     ipv6: Optional[str],
     ipv4_gateway: Optional[str],
     ipv6_gateway: Optional[str],
-    dns_servers: Optional[List[str]] = None,
+    dns_servers: Optional[list[str]] = None,
 ):
     """Setup the system with info from the host."""
     dns_servers = dns_servers or []
@@ -180,13 +170,13 @@ def setup_input_data(input_data: bytes):
             os.system("unzip -q /opt/input.zip -d /data")
 
 
-def setup_authorized_keys(authorized_keys: List[str]) -> None:
+def setup_authorized_keys(authorized_keys: list[str]) -> None:
     path = Path("/root/.ssh/authorized_keys")
     path.parent.mkdir(exist_ok=True)
     path.write_text("\n".join(key for key in authorized_keys))
 
 
-def setup_volumes(volumes: List[Volume]):
+def setup_volumes(volumes: list[Volume]):
     for volume in volumes:
         logger.debug(f"Mounting /dev/{volume.device} on {volume.mount}")
         os.makedirs(volume.mount, exist_ok=True)
@@ -213,7 +203,7 @@ async def receive():
             "type": f"lifespan.{event}",
         }
 
-    async def send(response: Dict):
+    async def send(response: dict):
         response_type = response.get("type")
         if response_type == f"lifespan.{event}.complete":
             lifespan_completion.set()
@@ -260,7 +250,7 @@ async def setup_code_asgi(code: bytes, encoding: Encoding, entrypoint: str) -> A
         app = getattr(module, app_name)
     elif encoding == Encoding.plain:
         # Execute the code and extract the entrypoint
-        locals: Dict[str, Any] = {}
+        locals: dict[str, Any] = {}
         exec(code, globals(), locals)
         app = locals[entrypoint]
     else:
@@ -313,7 +303,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, Optional[bytes]]:
     logger.debug("Running code")
     with StringIO() as buf, redirect_stdout(buf):
         # Execute in the same process, saves ~20ms than a subprocess
@@ -335,14 +325,14 @@ async def send(dico):
         await application(scope, receive, send)
 
         logger.debug("Waiting for headers")
-        headers: Dict
+        headers: dict
         if scope["type"] == "http":
             headers = await send_queue.get()
         else:
             headers = {}
 
         logger.debug("Waiting for body")
-        response_body: Dict = await send_queue.get()
+        response_body: dict = await send_queue.get()
 
         logger.debug("Waiting for buffer")
         output = buf.getvalue()
@@ -394,7 +384,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, Optional[bytes]]:
     logger.debug("Calling localhost")
 
     tries = 0
@@ -453,8 +443,8 @@ async def process_instruction(
 
         output: Optional[str] = None
         try:
-            headers: Dict
-            body: Dict
+            headers: dict
+            body: dict
             output_data: Optional[bytes]
 
             if interface == Interface.asgi:
@@ -532,7 +522,7 @@ def setup_system(config: ConfigurationPayload):
     logger.debug("Setup finished")
 
 
-def umount_volumes(volumes: List[Volume]):
+def umount_volumes(volumes: list[Volume]):
     "Umount user related filesystems"
     system("sync")
     for volume in volumes:

tests/supervisor/test_jwk.py

@@ -7,7 +7,7 @@
 # Avoid failures linked to settings when initializing the global VmPool object
 os.environ["ALEPH_VM_ALLOW_VM_NETWORKING"] = "False"
 
-from typing import Any, Dict
+from typing import Any
 
 import pytest
 
@@ -23,15 +23,15 @@ def valid_jwk_headers(mocker):
 
 @pytest.mark.skip(reason="TODO: Fix this test")
 @pytest.mark.asyncio
-async def test_valid_signature(valid_jwk_headers: Dict[str, Any], mocker):
+async def test_valid_signature(valid_jwk_headers: dict[str, Any], mocker):
     request = mocker.AsyncMock()
     request.headers = valid_jwk_headers
     await authenticate_jwk(request)
 
 
 @pytest.mark.skip(reason="TODO: Fix this test")
 @pytest.mark.asyncio
-async def test_invalid_signature(valid_jwk_headers: Dict[str, Any], mocker):
+async def test_invalid_signature(valid_jwk_headers: dict[str, Any], mocker):
     valid_jwk_headers["X-SignedOperation"] = (
         '{"time":"2023-07-14T22:14:14.132Z","signature":"96ffdbbd1704d5f6bfe4698235a0de0d2f58668deaa4371422bee26664f313f51fd483c78c34c6b317fc209779f9ddd9c45accf558e3bf881b49ad970ebf0ade"}'
     )
@@ -44,7 +44,7 @@ async def test_invalid_signature(valid_jwk_headers: Dict[str, Any], mocker):
 
 @pytest.mark.skip(reason="TODO: Fix this test")
 @pytest.mark.asyncio
-async def test_expired_token(valid_jwk_headers: Dict[str, Any], mocker):
+async def test_expired_token(valid_jwk_headers: dict[str, Any], mocker):
     mocker.patch("aleph.vm.orchestrator.views.authentication.is_token_still_valid", lambda timestamp: False)
     request = mocker.AsyncMock()
     request.headers = valid_jwk_headers
@@ -55,7 +55,7 @@ async def test_expired_token(valid_jwk_headers: Dict[str, Any], mocker):
 
 @pytest.mark.parametrize("missing_header", ["X-SignedPubKey", "X-SignedOperation"])
 @pytest.mark.asyncio
-async def test_missing_headers(valid_jwk_headers: Dict[str, Any], mocker, missing_header: str):
+async def test_missing_headers(valid_jwk_headers: dict[str, Any], mocker, missing_header: str):
     del valid_jwk_headers[missing_header]
 
     request = mocker.AsyncMock()

vm_connector/main.py

@@ -1,6 +1,6 @@
 import json
 import logging
-from typing import Dict, Optional, Union
+from typing import Optional
 
 import aiohttp
 from aleph_client.asynchronous import create_post
@@ -24,7 +24,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) -> Optional[dict]:
     async with aiohttp.ClientSession() as session:
         url = (
             f"{settings.API_SERVER}/api/v0/messages.json?msgType=STORE&sort_order=-1" f"&refs={ref}&addresses={sender}"
@@ -38,7 +38,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) -> Optional[dict]:
     async with aiohttp.ClientSession() as session:
         url = f"{settings.API_SERVER}/api/v0/messages.json?hashes={hash_}"
         resp = await session.get(url)
@@ -63,7 +63,7 @@ async def stream_url_chunks(url):
 
 
 @app.get("/download/message/{ref}")
-async def download_message(ref: str) -> Dict:
+async def download_message(ref: str) -> dict:
     """
     Fetch on Aleph and return a VM function message, after checking its validity.
     Used by the VM Supervisor run the code.