refactor(services): extract get_drive to the base Service class

selfprivacy_api/services/bitwarden/__init__.py

@@ -121,14 +121,6 @@ class Bitwarden(Service):
     def get_folders() -> typing.List[str]:
         return ["/var/lib/bitwarden", "/var/lib/bitwarden_rs"]
 
-    @staticmethod
-    def get_drive() -> str:
-        with ReadUserData() as user_data:
-            if user_data.get("useBinds", False):
-                return user_data.get("bitwarden", {}).get("location", "sda1")
-            else:
-                return "sda1"
-
     @staticmethod
     def get_dns_records() -> typing.List[ServiceDnsRecord]:
         """Return list of DNS records for Bitwarden service."""

selfprivacy_api/services/gitea/__init__.py

@@ -116,14 +116,6 @@ class Gitea(Service):
     def get_folders() -> typing.List[str]:
         return ["/var/lib/gitea"]
 
-    @staticmethod
-    def get_drive() -> str:
-        with ReadUserData() as user_data:
-            if user_data.get("useBinds", False):
-                return user_data.get("gitea", {}).get("location", "sda1")
-            else:
-                return "sda1"
-
     @staticmethod
     def get_dns_records() -> typing.List[ServiceDnsRecord]:
         return [

selfprivacy_api/services/jitsi/__init__.py

@@ -122,10 +122,6 @@ class Jitsi(Service):
     def get_folders() -> typing.List[str]:
         return ["/var/lib/jitsi-meet"]
 
-    @staticmethod
-    def get_drive() -> str:
-        return BlockDevices().get_root_block_device().name
-
     @staticmethod
     def get_dns_records() -> typing.List[ServiceDnsRecord]:
         ip4 = network_utils.get_ip4()

selfprivacy_api/services/mailserver/__init__.py

@@ -105,14 +105,6 @@ class MailServer(Service):
     def get_folders() -> typing.List[str]:
         return ["/var/vmail", "/var/sieve"]
 
-    @staticmethod
-    def get_drive() -> str:
-        with utils.ReadUserData() as user_data:
-            if user_data.get("useBinds", False):
-                return user_data.get("email", {}).get("location", "sda1")
-            else:
-                return "sda1"
-
     @staticmethod
     def get_dns_records() -> typing.List[ServiceDnsRecord]:
         domain = utils.get_domain()

selfprivacy_api/services/nextcloud/__init__.py

@@ -120,15 +120,6 @@ class Nextcloud(Service):
     def get_folders() -> typing.List[str]:
         return ["/var/lib/nextcloud"]
 
-    @staticmethod
-    def get_drive() -> str:
-        """Get the name of disk where Nextcloud is installed."""
-        with ReadUserData() as user_data:
-            if user_data.get("useBinds", False):
-                return user_data.get("nextcloud", {}).get("location", "sda1")
-            else:
-                return "sda1"
-
     @staticmethod
     def get_dns_records() -> typing.List[ServiceDnsRecord]:
         return [

selfprivacy_api/services/ocserv/__init__.py

@@ -98,10 +98,6 @@ class Ocserv(Service):
     def get_logs():
         return ""
 
-    @staticmethod
-    def get_drive() -> str:
-        return BlockDevices().get_root_block_device().name
-
     @staticmethod
     def get_dns_records() -> typing.List[ServiceDnsRecord]:
         return [

selfprivacy_api/services/pleroma/__init__.py

@@ -119,14 +119,6 @@ class Pleroma(Service):
             ),
         ]
 
-    @staticmethod
-    def get_drive() -> str:
-        with ReadUserData() as user_data:
-            if user_data.get("useBinds", False):
-                return user_data.get("pleroma", {}).get("location", "sda1")
-            else:
-                return "sda1"
-
     @staticmethod
     def get_dns_records() -> typing.List[ServiceDnsRecord]:
         return [

selfprivacy_api/services/service.py

@@ -6,10 +6,11 @@ import typing
 from pydantic import BaseModel
 from selfprivacy_api.jobs import Job
 
-from selfprivacy_api.utils.block_devices import BlockDevice
+from selfprivacy_api.utils.block_devices import BlockDevice, BlockDevices
 
 from selfprivacy_api.services.generic_size_counter import get_storage_usage
 from selfprivacy_api.services.owned_path import OwnedPath
+from selfprivacy_api import utils
 from selfprivacy_api.utils.waitloop import wait_until_true
 
 DEFAULT_START_STOP_TIMEOUT = 10 * 60
@@ -197,10 +198,23 @@ class Service(ABC):
     def get_dns_records() -> typing.List[ServiceDnsRecord]:
         pass
 
-    @staticmethod
-    @abstractmethod
-    def get_drive() -> str:
-        pass
+    @classmethod
+    def get_drive(cls) -> str:
+        """
+        Get the name of the drive/volume where the service is located.
+        Example values are `sda1`, `vda`, `sdb`.
+        """
+        root_device: str = BlockDevices().get_root_block_device().name
+        if not cls.is_movable():
+            return root_device
+        with utils.ReadUserData() as userdata:
+            if userdata.get("useBinds", False):
+                return userdata.get(cls.get_id(), {}).get(
+                    "location",
+                    root_device,
+                )
+            else:
+                return root_device
 
     @classmethod
     def get_folders(cls) -> typing.List[str]: