diff --git a/examples/example_fastapi/main.py b/examples/example_fastapi/main.py index 0f9195f7e..516d667a8 100644 --- a/examples/example_fastapi/main.py +++ b/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) diff --git a/runtimes/aleph-debian-11-python/init1.py b/runtimes/aleph-debian-11-python/init1.py index beb7060ea..f41128a8b 100644 --- a/runtimes/aleph-debian-11-python/init1.py +++ b/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: diff --git a/tests/supervisor/test_jwk.py b/tests/supervisor/test_jwk.py index f2fbd2efe..cc3b0ab09 100644 --- a/tests/supervisor/test_jwk.py +++ b/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,7 +23,7 @@ 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) @@ -31,7 +31,7 @@ async def test_valid_signature(valid_jwk_headers: Dict[str, Any], mocker): @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() diff --git a/vm_connector/main.py b/vm_connector/main.py index 02662d923..22a566f6b 100644 --- a/vm_connector/main.py +++ b/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.