Clean optional and list with hatch fmt

Author: olethanh

src/aleph/vm/conf.py

@@ -9,7 +9,7 @@
 from os.path import abspath, exists, isdir, isfile, join
 from pathlib import Path
 from subprocess import CalledProcessError, check_output
-from typing import Any, List, Literal, NewType, Optional
+from typing import Any, Literal, NewType
 
 from aleph_message.models import Chain
 from aleph_message.models.execution.environment import HypervisorType
@@ -188,21 +188,21 @@ class Settings(BaseSettings):
     CONNECTOR_URL = Url("http://localhost:4021")
 
     CACHE_ROOT: Path = Path("/var/cache/aleph/vm")
-    MESSAGE_CACHE: Optional[Path] = Field(
+    MESSAGE_CACHE: Path | None = Field(
         None,
         description="Default to CACHE_ROOT/message",
     )
-    CODE_CACHE: Optional[Path] = Field(None, description="Default to CACHE_ROOT/code")
-    RUNTIME_CACHE: Optional[Path] = Field(None, description="Default to CACHE_ROOT/runtime")
-    DATA_CACHE: Optional[Path] = Field(None, description="Default to CACHE_ROOT/data")
+    CODE_CACHE: Path | None = Field(None, description="Default to CACHE_ROOT/code")
+    RUNTIME_CACHE: Path | None = Field(None, description="Default to CACHE_ROOT/runtime")
+    DATA_CACHE: Path | None = Field(None, description="Default to CACHE_ROOT/data")
 
     EXECUTION_ROOT: Path = Path("/var/lib/aleph/vm")
-    JAILER_BASE_DIRECTORY: Optional[Path] = Field(None, description="Default to EXECUTION_ROOT/jailer")
-    EXECUTION_DATABASE: Optional[Path] = Field(
+    JAILER_BASE_DIRECTORY: Path | None = Field(None, description="Default to EXECUTION_ROOT/jailer")
+    EXECUTION_DATABASE: Path | None = Field(
         None, description="Location of database file. Default to EXECUTION_ROOT/executions.sqlite3"
     )
     EXECUTION_LOG_ENABLED: bool = False
-    EXECUTION_LOG_DIRECTORY: Optional[Path] = Field(
+    EXECUTION_LOG_DIRECTORY: Path | None = Field(
         None, description="Location of executions log. Default to EXECUTION_ROOT/executions/"
     )
 
@@ -267,12 +267,12 @@ class Settings(BaseSettings):
         "with SEV and SEV-ES",
     )
 
-    CONFIDENTIAL_DIRECTORY: Optional[Path] = Field(
+    CONFIDENTIAL_DIRECTORY: Path | None = Field(
         None,
         description="Confidential Computing default directory. Default to EXECUTION_ROOT/confidential",
     )
 
-    CONFIDENTIAL_SESSION_DIRECTORY: Optional[Path] = Field(None, description="Default to EXECUTION_ROOT/sessions")
+    CONFIDENTIAL_SESSION_DIRECTORY: Path | None = Field(None, description="Default to EXECUTION_ROOT/sessions")
 
     ENABLE_GPU_SUPPORT: bool = Field(
         default=False,

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

@@ -5,7 +5,7 @@
 from asyncio import Task
 from asyncio.subprocess import Process
 from pathlib import Path
-from typing import Generic, List, TypeVar
+from typing import Generic, TypeVar
 
 import psutil
 from aleph_message.models import ItemHash
@@ -38,7 +38,7 @@
 
 
 class AlephQemuResources(AlephFirecrackerResources):
-    gpus: List[HostGPU] = []
+    gpus: list[HostGPU] = []
 
     async def download_runtime(self) -> None:
         volume = self.message_content.rootfs

src/aleph/vm/garbage_collector.py

@@ -98,6 +98,7 @@ def check_api(item_hash):
         f"systemctl status aleph-vm-controller@{item_hash}.service --no-pager",
         shell=True,
         capture_output=True,
+        check=False,
     )
     exit_code = proc_ret.returncode
     if exit_code == 0:

src/aleph/vm/models.py

@@ -6,7 +6,6 @@
 from collections.abc import Callable, Coroutine
 from dataclasses import dataclass
 from datetime import datetime, timezone
-from typing import List
 
 from aleph_message.models import (
     ExecutableContent,
@@ -76,7 +75,7 @@ class VmExecution:
         AlephProgramResources | AlephInstanceResources | AlephQemuResources | AlephQemuConfidentialInstance | None
     ) = None
     vm: AlephFirecrackerExecutable | AlephQemuInstance | AlephQemuConfidentialInstance | None = None
-    gpus: List[HostGPU] = []
+    gpus: list[HostGPU] = []
 
     times: VmExecutionTimes
 
@@ -223,7 +222,7 @@ async def prepare(self) -> None:
             self.times.prepared_at = datetime.now(tz=timezone.utc)
             self.resources = resources
 
-    def prepare_gpus(self, available_gpus: List[GpuDevice]) -> None:
+    def prepare_gpus(self, available_gpus: list[GpuDevice]) -> None:
         gpus = []
         if self.message.requirements and self.message.requirements.gpu:
             for gpu in self.message.requirements.gpu:

src/aleph/vm/orchestrator/resources.py

@@ -1,7 +1,6 @@
 import math
 from datetime import datetime, timezone
 from functools import lru_cache
-from typing import List, Optional
 
 import cpuinfo
 import psutil
@@ -77,8 +76,8 @@ class MachineProperties(BaseModel):
 
 
 class GpuProperties(BaseModel):
-    devices: Optional[List[GpuDevice]]
-    available_devices: Optional[List[GpuDevice]]
+    devices: list[GpuDevice] | None
+    available_devices: list[GpuDevice] | None
 
 
 class MachineUsage(BaseModel):

src/aleph/vm/pool.py

@@ -5,7 +5,6 @@
 import logging
 from collections.abc import Iterable
 from datetime import datetime, timezone
-from typing import List
 
 from aleph_message.models import (
     Chain,
@@ -45,7 +44,7 @@ class VmPool:
     snapshot_manager: SnapshotManager | None = None
     systemd_manager: SystemDManager
     creation_lock: asyncio.Lock
-    gpus: List[GpuDevice] = []
+    gpus: list[GpuDevice] = []
 
     def __init__(self, loop: asyncio.AbstractEventLoop):
         self.executions = {}
@@ -249,7 +248,7 @@ async def load_persistent_executions(self):
             if execution.is_running:
                 # TODO: Improve the way that we re-create running execution
                 # Load existing GPUs assigned to VMs
-                execution.gpus = parse_raw_as(List[HostGPU], saved_execution.gpus) if saved_execution.gpus else []
+                execution.gpus = parse_raw_as(list[HostGPU], saved_execution.gpus) if saved_execution.gpus else []
                 # Load and instantiate the rest of resources and already assigned GPUs
                 await execution.prepare()
                 if self.network:
@@ -303,7 +302,7 @@ def get_instance_executions(self) -> Iterable[VmExecution]:
         )
         return executions or []
 
-    def get_available_gpus(self) -> List[GpuDevice]:
+    def get_available_gpus(self) -> list[GpuDevice]:
         available_gpus = []
         for gpu in self.gpus:
             used = False

src/aleph/vm/resources.py

@@ -1,11 +1,9 @@
 import subprocess
 from enum import Enum
-from typing import List, Optional
 
 from aleph_message.models import HashableModel
 from pydantic import BaseModel, Extra, Field
 
-from aleph.vm.conf import settings
 from aleph.vm.orchestrator.utils import get_compatible_gpus
 
 
@@ -97,7 +95,7 @@ def is_kernel_enabled_gpu(pci_host: str) -> bool:
     return False
 
 
-def parse_gpu_device_info(line: str) -> Optional[GpuDevice]:
+def parse_gpu_device_info(line: str) -> GpuDevice | None:
     """Parse GPU device info from a line of lspci output."""
 
     pci_host, device = line.split(' "', maxsplit=1)
@@ -134,7 +132,7 @@ def parse_gpu_device_info(line: str) -> Optional[GpuDevice]:
     )
 
 
-def get_gpu_devices() -> Optional[List[GpuDevice]]:
+def get_gpu_devices() -> list[GpuDevice] | None:
     """Get GPU info using lspci command."""
 
     result = subprocess.run(["lspci", "-mmnnn"], capture_output=True, text=True, check=True)

src/aleph/vm/storage.py

@@ -10,7 +10,6 @@
 import logging
 import re
 import sys
-import uuid
 from datetime import datetime, timezone
 from pathlib import Path
 from shutil import copy2, make_archive
@@ -382,7 +381,7 @@ async def get_volume_path(volume: MachineVolume, namespace: str) -> Path:
             raise NotImplementedError(msg)
         if not re.match(r"^[\w\-_/]+$", volume_name):
             # Sanitize volume names
-            logger.debug(f"Invalid values for volume name: {repr(volume_name)} detected, sanitizing")
+            logger.debug(f"Invalid values for volume name: {volume_name!r} detected, sanitizing")
             volume_name = re.sub(r"[^\w\-_]", "_", volume_name)
         (Path(settings.PERSISTENT_VOLUMES_DIR) / namespace).mkdir(exist_ok=True)
         if volume.parent: