Fix: Small code quality fixes by ruff check --fix

src/aleph/vm/controllers/interface.py

@@ -2,7 +2,8 @@
 import logging
 from abc import ABC
 from asyncio.subprocess import Process
-from typing import Any, Coroutine, Optional
+from collections.abc import Coroutine
+from typing import Any, Optional
 
 from aleph_message.models import ItemHash
 from aleph_message.models.execution.environment import MachineResources

src/aleph/vm/controllers/qemu/instance.py

@@ -6,7 +6,7 @@
 from asyncio import Task
 from asyncio.subprocess import Process
 from pathlib import Path
-from typing import Callable, Dict, Generic, Optional, Tuple, TypedDict, TypeVar, Union
+from typing import Callable, Generic, Optional, TypedDict, TypeVar, Union
 
 import psutil
 from aleph_message.models import ItemHash
@@ -37,11 +37,10 @@
 
 
 class AlephQemuResources(AlephFirecrackerResources):
- async def download_all(self):
+ async def download_all(self) -> None:
  volume = self.message_content.rootfs
  parent_image_path = await get_rootfs_base_path(volume.parent.ref)
  self.rootfs_path = await self.make_writable_volume(parent_image_path, volume)
- return
 
  async def make_writable_volume(self, parent_image_path, volume: Union[PersistentVolume, RootfsVolume]):
  """Create a new qcow2 image file based on the passed one, that we give to the VM to write onto"""
@@ -90,7 +89,7 @@ class EntryDict(TypedDict):
  MESSAGE: str
 
 
-def make_logs_queue(stdout_identifier, stderr_identifier, skip_past=True) -> Tuple[asyncio.Queue, Callable[[], None]]:
+def make_logs_queue(stdout_identifier, stderr_identifier, skip_past=True) -> tuple[asyncio.Queue, Callable[[], None]]:
  """Create a queue which streams the logs for the process.
 
  @param stdout_identifier: journald identifier for process stdout
@@ -149,7 +148,7 @@ class AlephQemuInstance(Generic[ConfigurationType], CloudInitMixin, AlephVmContr
  support_snapshot = False
  qmp_socket_path = None
  persistent = True
- _queue_cancellers: Dict[asyncio.Queue, Callable] = {}
+ _queue_cancellers: dict[asyncio.Queue, Callable] = {}
  controller_configuration: Configuration
 
  def __repr__(self):

src/aleph/vm/hypervisors/qemu/qemuvm.py

@@ -11,7 +11,7 @@
 from aleph.vm.controllers.qemu.instance import logger
 
 
-class QemuVM(object):
+class QemuVM:
  qemu_bin_path: str
  cloud_init_drive_path: Optional[str]
  image_path: str
@@ -26,7 +26,7 @@
  if self.qemu_process:
  return f"<QemuVM: {self.qemu_process.pid}>"
  else:
- return f"<QemuVM: not running>"
+ return "<QemuVM: not running>"
 
  def __init__(self, config: QemuVMConfiguration):
  self.qemu_bin_path = config.qemu_bin_path

src/aleph/vm/models.py

@@ -2,9 +2,10 @@
 import logging
 import uuid
 from asyncio import Task
+from collections.abc import Coroutine
 from dataclasses import dataclass
 from datetime import datetime, timezone
-from typing import TYPE_CHECKING, Callable, Coroutine, Optional, Union
+from typing import TYPE_CHECKING, Callable, Optional, Union
 
 from aleph_message.models import (
  ExecutableContent,

src/aleph/vm/orchestrator/payment.py

@@ -1,7 +1,8 @@
 import asyncio
 import logging
+from collections.abc import Iterable
 from decimal import Decimal
-from typing import Iterable, Optional
+from typing import Optional
 
 import aiohttp
 from aleph_message.models import ItemHash, PaymentType

src/aleph/vm/orchestrator/views/authentication.py

@@ -1,8 +1,9 @@
 import functools
 import json
 import logging
+from collections.abc import Awaitable, Coroutine
 from datetime import datetime, timedelta, timezone
-from typing import Any, Awaitable, Callable, Coroutine, Literal, Union
+from typing import Any, Callable, Literal, Union
 
 import pydantic
 from aiohttp import web

src/aleph/vm/orchestrator/views/host_status.py

@@ -1,6 +1,7 @@
 import logging
 import socket
-from typing import Any, Awaitable, Callable, Optional, Tuple
+from collections.abc import Awaitable
+from typing import Any, Callable, Optional
 
 import aiohttp
 
@@ -45,7 +46,7 @@ async def check_host_egress_ipv6() -> bool:
  return await check_ip_connectivity(settings.CONNECTIVITY_IPV6_URL)
 
 
-async def resolve_dns(hostname: str) -> Tuple[Optional[str], Optional[str]]:
+async def resolve_dns(hostname: str) -> tuple[Optional[str], Optional[str]]:
  """Resolve a hostname to an IPv4 and IPv6 address."""
  ipv4: Optional[str] = None
  ipv6: Optional[str] = None

src/aleph/vm/pool.py

@@ -5,7 +5,7 @@
 import logging
 from collections.abc import Iterable
 from datetime import datetime, timezone
-from typing import Dict, Optional
+from typing import Optional
 
 from aleph_message.models import (
  Chain,
@@ -284,9 +284,9 @@
  )
  return executions or []
 
- def get_executions_by_sender(self, payment_type: PaymentType) -> Dict[str, Dict[str, list[VmExecution]]]:
+ def get_executions_by_sender(self, payment_type: PaymentType) -> dict[str, dict[str, list[VmExecution]]]:
  """Return all executions of the given type, grouped by sender and by chain."""
- executions_by_sender: Dict[str, Dict[str, list[VmExecution]]] = {}
+ executions_by_sender: dict[str, dict[str, list[VmExecution]]] = {}
  for vm_hash, execution in self.executions.items():
  if execution.vm_hash in (settings.CHECK_FASTAPI_VM_ID, settings.LEGACY_CHECK_FASTAPI_VM_ID):
  # Ignore Diagnostic VM execution

src/aleph/vm/utils.py

@@ -10,7 +10,7 @@
 from dataclasses import is_dataclass
 from pathlib import Path
 from shutil import disk_usage
-from typing import Any, Callable, Dict, Optional
+from typing import Any, Callable, Optional
 
 import aiodns
 import msgpack
@@ -22,10 +22,10 @@
 logger = logging.getLogger(__name__)
 
 
-def get_message_executable_content(message_dict: Dict) -> ExecutableContent:
+def get_message_executable_content(message_dict: dict) -> ExecutableContent:
  try:
  return ProgramContent.parse_obj(message_dict)
- except ValueError as error:
+ except ValueError:
  return InstanceContent.parse_obj(message_dict)
 
 
@@ -190,11 +190,11 @@
  try:
  hex_address = hexstr_if_str(to_hex, value).lower()
  except AttributeError:
- raise TypeError("Value must be any string, instead got type {}".format(type(value)))
+ raise TypeError(f"Value must be any string, instead got type {type(value)}")
  if is_address(hex_address):
  return HexAddress(HexStr(hex_address))
  else:
- raise ValueError("Unknown format {}, attempted to normalize to {}".format(value, hex_address))
+ raise ValueError(f"Unknown format {value}, attempted to normalize to {hex_address}")
 
 
 def md5sum(file_path: Path) -> str:
@@ -205,7 +205,7 @@
 def file_hashes_differ(source: Path, destination: Path, checksum: Callable[[Path], str] = md5sum) -> bool:
  """Check if the MD5 hash of two files differ."""
  if not source.exists():
- raise FileNotFoundError("Source file does not exist: {}".format(source))
+ raise FileNotFoundError(f"Source file does not exist: {source}")
 
  if not destination.exists():
  return True