Fix: Missing type annotation

Adding type annotation for a better code clarity and safety.
This has been done on the migration of pydantic branch, but adding it first because it should not
break the code and facilitate the merge of the pydantic PR

Author: Antonyjin

src/aleph/vm/conf.py

@@ -9,13 +9,12 @@
 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
+from typing import Any, List, Literal, NewType, Optional
 
 from aleph_message.models import Chain
 from aleph_message.models.execution.environment import HypervisorType
 from pydantic import BaseSettings, Field, HttpUrl
 from pydantic.env_settings import DotenvType, env_file_sentinel
-from pydantic.typing import StrPath
 
 from aleph.vm.orchestrator.chain import STREAM_CHAINS
 from aleph.vm.utils import (
@@ -111,7 +110,7 @@ def obtain_dns_ips(dns_resolver: DnsResolver, network_interface: str) -> list[st
 
 
 class Settings(BaseSettings):
-    SUPERVISOR_HOST = "127.0.0.1"
+    SUPERVISOR_HOST: str = "127.0.0.1"
     SUPERVISOR_PORT: int = 4020
 
     # Public domain name
@@ -123,32 +122,32 @@ class Settings(BaseSettings):
     START_ID_INDEX: int = 4
     PREALLOC_VM_COUNT: int = 0
     REUSE_TIMEOUT: float = 60 * 60.0
-    WATCH_FOR_MESSAGES = True
-    WATCH_FOR_UPDATES = True
+    WATCH_FOR_MESSAGES: bool = True
+    WATCH_FOR_UPDATES: bool = True
 
-    API_SERVER = "https://official.aleph.cloud"
+    API_SERVER: str = "https://official.aleph.cloud"
     # Connect to the Quad9 VPN provider using their IPv4 and IPv6 addresses.
-    CONNECTIVITY_IPV4_URL = "https://9.9.9.9/"
-    CONNECTIVITY_IPV6_URL = "https://[2620:fe::fe]/"
-    CONNECTIVITY_DNS_HOSTNAME = "example.org"
+    CONNECTIVITY_IPV4_URL: str = "https://9.9.9.9/"
+    CONNECTIVITY_IPV6_URL: str = "https://[2620:fe::fe]/"
+    CONNECTIVITY_DNS_HOSTNAME: str = "example.org"
 
-    USE_JAILER = True
+    USE_JAILER: bool = True
     # Changelog: PRINT_SYSTEM_LOGS use to print the MicroVM logs with the supervisor output.
     # They are now in separate journald entries, disabling the settings disable the logs output of Firecracker VM (only)
     # via the serial console. This break the logs endpoint for program, as such disabling it in prod is not recommended.
-    PRINT_SYSTEM_LOGS = True
-    IGNORE_TRACEBACK_FROM_DIAGNOSTICS = True
-    LOG_LEVEL = "INFO"
-    DEBUG_ASYNCIO = False
+    PRINT_SYSTEM_LOGS: bool = True
+    IGNORE_TRACEBACK_FROM_DIAGNOSTICS: bool = True
+    LOG_LEVEL: str = "INFO"
+    DEBUG_ASYNCIO: bool = False
 
     # Networking does not work inside Docker/Podman
-    ALLOW_VM_NETWORKING = True
+    ALLOW_VM_NETWORKING: bool = True
     NETWORK_INTERFACE: str | None = None
-    IPV4_ADDRESS_POOL = Field(
+    IPV4_ADDRESS_POOL: str = Field(
         default="172.16.0.0/12",
         description="IPv4 address range used to provide networks to VMs.",
     )
-    IPV4_NETWORK_PREFIX_LENGTH = Field(
+    IPV4_NETWORK_PREFIX_LENGTH: int = Field(
         default=24,
         description="Individual VM network prefix length in bits",
     )
@@ -180,30 +179,30 @@ class Settings(BaseSettings):
     DNS_NAMESERVERS_IPV4: list[str] | None
     DNS_NAMESERVERS_IPV6: list[str] | None
 
-    FIRECRACKER_PATH = Path("/opt/firecracker/firecracker")
-    JAILER_PATH = Path("/opt/firecracker/jailer")
-    SEV_CTL_PATH = Path("/opt/sevctl")
-    LINUX_PATH = Path("/opt/firecracker/vmlinux.bin")
+    FIRECRACKER_PATH: Path = Path("/opt/firecracker/firecracker")
+    JAILER_PATH: Path = Path("/opt/firecracker/jailer")
+    SEV_CTL_PATH: Path = Path("/opt/sevctl")
+    LINUX_PATH: Path = Path("/opt/firecracker/vmlinux.bin")
     INIT_TIMEOUT: float = 20.0
 
     CONNECTOR_URL = Url("http://localhost:4021")
 
-    CACHE_ROOT = Path("/var/cache/aleph/vm")
-    MESSAGE_CACHE: Path = Field(
+    CACHE_ROOT: Path = Path("/var/cache/aleph/vm")
+    MESSAGE_CACHE: Optional[Path] = Field(
         None,
         description="Default to CACHE_ROOT/message",
     )
-    CODE_CACHE: Path = Field(None, description="Default to CACHE_ROOT/code")
-    RUNTIME_CACHE: Path = Field(None, description="Default to CACHE_ROOT/runtime")
-    DATA_CACHE: Path = Field(None, description="Default to CACHE_ROOT/data")
+    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")
 
-    EXECUTION_ROOT = Path("/var/lib/aleph/vm")
-    JAILER_BASE_DIRECTORY: Path = Field(None, description="Default to EXECUTION_ROOT/jailer")
-    EXECUTION_DATABASE: Path = Field(
+    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(
         None, description="Location of database file. Default to EXECUTION_ROOT/executions.sqlite3"
     )
-    EXECUTION_LOG_ENABLED = False
-    EXECUTION_LOG_DIRECTORY: Path = Field(
+    EXECUTION_LOG_ENABLED: bool = False
+    EXECUTION_LOG_DIRECTORY: Optional[Path] = Field(
         None, description="Location of executions log. Default to EXECUTION_ROOT/executions/"
     )
 
@@ -212,8 +211,8 @@ class Settings(BaseSettings):
     )
     JAILER_BASE_DIR: Path = Field(None)
 
-    MAX_PROGRAM_ARCHIVE_SIZE = 10_000_000  # 10 MB
-    MAX_DATA_ARCHIVE_SIZE = 10_000_000  # 10 MB
+    MAX_PROGRAM_ARCHIVE_SIZE: int = 10_000_000  # 10 MB
+    MAX_DATA_ARCHIVE_SIZE: int = 10_000_000  # 10 MB
 
     PAYMENT_MONITOR_INTERVAL: float = Field(
         default=60.0,
@@ -254,7 +253,7 @@ class Settings(BaseSettings):
     )
 
     # hashlib.sha256(b"secret-token").hexdigest()
-    ALLOCATION_TOKEN_HASH = "151ba92f2eb90bce67e912af2f7a5c17d8654b3d29895b042107ea312a7eebda"
+    ALLOCATION_TOKEN_HASH: str = "151ba92f2eb90bce67e912af2f7a5c17d8654b3d29895b042107ea312a7eebda"
 
     ENABLE_QEMU_SUPPORT: bool = Field(default=True)
     INSTANCE_DEFAULT_HYPERVISOR: HypervisorType | None = Field(
@@ -268,12 +267,12 @@ class Settings(BaseSettings):
         "with SEV and SEV-ES",
     )
 
-    CONFIDENTIAL_DIRECTORY: Path = Field(
+    CONFIDENTIAL_DIRECTORY: Optional[Path] = Field(
         None,
         description="Confidential Computing default directory. Default to EXECUTION_ROOT/confidential",
     )
 
-    CONFIDENTIAL_SESSION_DIRECTORY: Path = Field(None, description="Default to EXECUTION_ROOT/sessions")
+    CONFIDENTIAL_SESSION_DIRECTORY: Optional[Path] = Field(None, description="Default to EXECUTION_ROOT/sessions")
 
     ENABLE_GPU_SUPPORT: bool = Field(
         default=False,
@@ -285,11 +284,13 @@ class Settings(BaseSettings):
 
     # Tests on programs
     FAKE_DATA_PROGRAM: Path | None = None
-    BENCHMARK_FAKE_DATA_PROGRAM = Path(abspath(join(__file__, "../../../../examples/example_fastapi")))
+    BENCHMARK_FAKE_DATA_PROGRAM: Path = Path(abspath(join(__file__, "../../../../examples/example_fastapi")))
 
-    FAKE_DATA_MESSAGE = Path(abspath(join(__file__, "../../../../examples/program_message_from_aleph.json")))
+    FAKE_DATA_MESSAGE: Path = Path(abspath(join(__file__, "../../../../examples/program_message_from_aleph.json")))
     FAKE_DATA_DATA: Path | None = Path(abspath(join(__file__, "../../../../examples/data/")))
-    FAKE_DATA_RUNTIME = Path(abspath(join(__file__, "../../../../runtimes/aleph-debian-12-python/rootfs.squashfs")))
+    FAKE_DATA_RUNTIME: Path = Path(
+        abspath(join(__file__, "../../../../runtimes/aleph-debian-12-python/rootfs.squashfs"))
+    )
     FAKE_DATA_VOLUME: Path | None = Path(abspath(join(__file__, "../../../../examples/volumes/volume-venv.squashfs")))
 
     # Tests on instances
@@ -299,9 +300,9 @@ class Settings(BaseSettings):
         description="Identifier of the instance message used when testing the launch of an instance from the network",
     )
 
-    USE_FAKE_INSTANCE_BASE = False
-    FAKE_INSTANCE_BASE = Path(abspath(join(__file__, "../../../../runtimes/instance-rootfs/debian-12.btrfs")))
-    FAKE_QEMU_INSTANCE_BASE = Path(abspath(join(__file__, "../../../../runtimes/instance-rootfs/rootfs.img")))
+    USE_FAKE_INSTANCE_BASE: bool = False
+    FAKE_INSTANCE_BASE: Path = Path(abspath(join(__file__, "../../../../runtimes/instance-rootfs/debian-12.btrfs")))
+    FAKE_QEMU_INSTANCE_BASE: Path = Path(abspath(join(__file__, "../../../../runtimes/instance-rootfs/rootfs.img")))
     FAKE_INSTANCE_ID: str = Field(
         default="decadecadecadecadecadecadecadecadecadecadecadecadecadecadecadeca",
         description="Identifier used for the 'fake instance' message defined in "
@@ -310,8 +311,8 @@ class Settings(BaseSettings):
     FAKE_INSTANCE_MESSAGE = Path(abspath(join(__file__, "../../../../examples/instance_message_from_aleph.json")))
     FAKE_INSTANCE_QEMU_MESSAGE = Path(abspath(join(__file__, "../../../../examples/qemu_message_from_aleph.json")))
 
-    CHECK_FASTAPI_VM_ID = "63faf8b5db1cf8d965e6a464a0cb8062af8e7df131729e48738342d956f29ace"
-    LEGACY_CHECK_FASTAPI_VM_ID = "67705389842a0a1b95eaa408b009741027964edc805997475e95c505d642edd8"
+    CHECK_FASTAPI_VM_ID: str = "63faf8b5db1cf8d965e6a464a0cb8062af8e7df131729e48738342d956f29ace"
+    LEGACY_CHECK_FASTAPI_VM_ID: str = "67705389842a0a1b95eaa408b009741027964edc805997475e95c505d642edd8"
 
     # Developer options
 
@@ -408,10 +409,14 @@ def setup(self):
         STREAM_CHAINS[Chain.AVAX].rpc = str(self.RPC_AVAX)
         STREAM_CHAINS[Chain.BASE].rpc = str(self.RPC_BASE)
 
-        os.makedirs(self.MESSAGE_CACHE, exist_ok=True)
-        os.makedirs(self.CODE_CACHE, exist_ok=True)
-        os.makedirs(self.RUNTIME_CACHE, exist_ok=True)
-        os.makedirs(self.DATA_CACHE, exist_ok=True)
+        if self.MESSAGE_CACHE:
+            os.makedirs(self.MESSAGE_CACHE, exist_ok=True)
+        if self.CODE_CACHE:
+            os.makedirs(self.CODE_CACHE, exist_ok=True)
+        if self.RUNTIME_CACHE:
+            os.makedirs(self.RUNTIME_CACHE, exist_ok=True)
+        if self.DATA_CACHE:
+            os.makedirs(self.DATA_CACHE, exist_ok=True)
 
         os.makedirs(self.EXECUTION_ROOT, exist_ok=True)
 
@@ -427,10 +432,14 @@ def setup(self):
 
             self.LINUX_PATH = linux_path_on_device
 
-        os.makedirs(self.EXECUTION_LOG_DIRECTORY, exist_ok=True)
-        os.makedirs(self.PERSISTENT_VOLUMES_DIR, exist_ok=True)
-        os.makedirs(self.CONFIDENTIAL_DIRECTORY, exist_ok=True)
-        os.makedirs(self.CONFIDENTIAL_SESSION_DIRECTORY, exist_ok=True)
+        if self.EXECUTION_LOG_DIRECTORY:
+            os.makedirs(self.EXECUTION_LOG_DIRECTORY, exist_ok=True)
+        if self.PERSISTENT_VOLUMES_DIR:
+            os.makedirs(self.PERSISTENT_VOLUMES_DIR, exist_ok=True)
+        if self.CONFIDENTIAL_DIRECTORY:
+            os.makedirs(self.CONFIDENTIAL_DIRECTORY, exist_ok=True)
+        if self.CONFIDENTIAL_SESSION_DIRECTORY:
+            os.makedirs(self.CONFIDENTIAL_SESSION_DIRECTORY, exist_ok=True)
 
         self.API_SERVER = self.API_SERVER.rstrip("/")
 
@@ -479,7 +488,7 @@ def __init__(
         _env_file: DotenvType | None = env_file_sentinel,
         _env_file_encoding: str | None = None,
         _env_nested_delimiter: str | None = None,
-        _secrets_dir: StrPath | None = None,
+        _secrets_dir: Path | None = None,
         **values: Any,
     ) -> None:
         super().__init__(_env_file, _env_file_encoding, _env_nested_delimiter, _secrets_dir, **values)