"""
This module implements mounting file systems via communication with udisksd.
"""
import logging
import time
import gi
gi.require_version('UDisks', '2.0')
from gi.repository import GLib, UDisks
[docs]
class UDisks2Device:
UNMOUNT_MAX_RETRIES = 5
UNMOUNT_BUSY_WAIT = 3 # s
[docs]
def __init__(self, devpath):
self._logger = logging.getLogger("Device: ")
self.devpath = devpath
self.fs = None
def _setup(self):
"""Try to find the devpath
Raises:
ValueError: no udisks2 device or no filesystem found on devpath
"""
client = UDisks.Client.new_sync(None)
manager = client.get_object_manager()
for obj in manager.get_objects():
block = obj.get_block()
if not block:
continue
device_path = block.get_cached_property("Device").get_bytestring().decode('utf-8')
if device_path == self.devpath:
self.fs = obj.get_filesystem()
if self.fs is None:
raise ValueError(f"no filesystem found on {self.devpath}")
return
raise ValueError(f"No udisks2 device found for {self.devpath}")
[docs]
def mount(self, readonly=False, retries=0):
opts = GLib.Variant('a{sv}', {'options': GLib.Variant('s', 'ro' if readonly else 'rw')})
try:
mountpoint = self.fs.call_mount_sync(opts, None)
except GLib.GError as err:
if not err.matches(UDisks.error_quark(), UDisks.Error.ALREADY_MOUNTED):
raise err
self._logger.warning('Unmounting lazily and remounting %s...', self.devpath)
self._unmount_lazy()
mountpoint = self.fs.call_mount_sync(opts, None)
return mountpoint
def _unmount_lazy(self):
opts = GLib.Variant('a{sv}', {'force': GLib.Variant('b', True)})
try:
self.fs.call_unmount_sync(opts, None)
except GLib.GError as err:
if not err.matches(UDisks.error_quark(), UDisks.Error.NOT_MOUNTED):
raise err
def _unmount(self):
opts = GLib.Variant('a{sv}', {'force': GLib.Variant('b', False)})
for _ in range(self.UNMOUNT_MAX_RETRIES):
try:
self.fs.call_unmount_sync(opts, None)
return
except GLib.GError as err:
if not err.matches(UDisks.error_quark(), UDisks.Error.DEVICE_BUSY):
raise err
self._logger.warning('waiting %s s for busy %s',
self.UNMOUNT_BUSY_WAIT, self.devpath)
time.sleep(self.UNMOUNT_BUSY_WAIT)
raise TimeoutError("Timeout waiting for device to become non-busy")
[docs]
def unmount(self, lazy=False):
if lazy:
self._unmount_lazy()
else:
self._unmount()
_devs = {}
def _get_udisks2_dev(devpath, retries):
"""Try to get the udisks2 device
Args:
devpath (str): Device name
retries (int): Number of retries to allow
Raises:
ValueError: Failed to obtain the device (e.g. does not exist)
"""
if devpath not in _devs:
dev = UDisks2Device(devpath=devpath)
while True:
try:
dev._setup()
break
except ValueError as exc:
if 'No udisks2 device' not in str(exc) or not retries:
raise
retries -= 1
dev._logger.warning('udisks2: Retrying %s...', devpath)
time.sleep(1)
# Success, so record the new device
_devs[devpath] = dev
return _devs[devpath]
[docs]
def handle_mount(devpath, retries=0):
dev = _get_udisks2_dev(devpath, retries)
return dev.mount()
[docs]
def handle_unmount(devpath, lazy=False):
dev = _get_udisks2_dev(devpath, 0)
return dev.unmount(lazy=lazy)
methods = {
'mount': handle_mount,
'unmount': handle_unmount,
}