SelfPrivacy
/
selfprivacy-rest-api
7
2
Fork 1
Code Issues 15 Pull Requests 4 Projects Releases 1 Wiki Activity

Add more fields to GraphQL storage query

graphql
Inex Code 2022-07-30 17:48:33 +03:00
parent 1f64a76723
commit 67c8486c9b
6 changed files with 131 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
selfprivacy_api/graphql/mutations/storage_mutation.py
View File

@ -22,3 +22,41 @@ class StorageMutations:
return GenericMutationReturn(
success=True, code=200, message="Volume resize started"
)
@strawberry.mutation(permission_classes=[IsAuthenticated])
def mount_volume(self, name: str) -> GenericMutationReturn:
"""Mount volume"""
volume = BlockDevices().get_block_device(name)
if volume is None:
return GenericMutationReturn(
success=False, code=404, message="Volume not found"
)
is_success = volume.mount()
if is_success:
return GenericMutationReturn(
success=True,
code=200,
message="Volume mounted, rebuild the system to apply changes",
)
return GenericMutationReturn(
success=False, code=409, message="Volume not mounted (already mounted?)"
)
@strawberry.mutation(permission_classes=[IsAuthenticated])
def unmount_volume(self, name: str) -> GenericMutationReturn:
"""Unmount volume"""
volume = BlockDevices().get_block_device(name)
if volume is None:
return GenericMutationReturn(
success=False, code=404, message="Volume not found"
)
is_success = volume.unmount()
if is_success:
return GenericMutationReturn(
success=True,
code=200,
message="Volume unmounted, rebuild the system to apply changes",
)
return GenericMutationReturn(
success=False, code=409, message="Volume not unmounted (already unmounted?)"
)

14
selfprivacy_api/graphql/queries/storage.py
View File

@ -7,25 +7,37 @@ from selfprivacy_api.utils.block_devices import BlockDevices
@strawberry.type
class StorageVolume:
"""Stats and basic info about a volume or a system disk."""
total_space: str
free_space: str
used_space: str
root: bool
name: str
model: str
serial: str
type: str
@strawberry.type
class Storage:
"""GraphQL queries to get storage information."""
@strawberry.field
def volumes(self) -> typing.List[StorageVolume]:
"""Get list of volumes"""
return [
StorageVolume(
total_space=str(volume.fssize) if volume.fssize is not None else str(volume.size),
total_space=str(volume.fssize)
if volume.fssize is not None
else str(volume.size),
free_space=str(volume.fsavail),
used_space=str(volume.fsused),
root=volume.name == "sda1",
name=volume.name,
model=volume.model,
serial=volume.serial,
type=volume.type,
)
for volume in BlockDevices().get_block_devices()
]

11
selfprivacy_api/migrations/__init__.py
View File

@ -11,10 +11,17 @@ Adding DISABLE_ALL to that array disables the migrations module entirely.
from selfprivacy_api.utils import ReadUserData
from selfprivacy_api.migrations.fix_nixos_config_branch import FixNixosConfigBranch
from selfprivacy_api.migrations.create_tokens_json import CreateTokensJson
from selfprivacy_api.migrations.migrate_to_selfprivacy_channel import MigrateToSelfprivacyChannel
from selfprivacy_api.migrations.migrate_to_selfprivacy_channel import (
MigrateToSelfprivacyChannel,
)
from selfprivacy_api.migrations.mount_volume import MountVolume
migrations = [FixNixosConfigBranch(), CreateTokensJson(), MigrateToSelfprivacyChannel(), MountVolume()]
migrations = [
FixNixosConfigBranch(),
CreateTokensJson(),
MigrateToSelfprivacyChannel(),
MountVolume(),
]
def run_migrations():

13
selfprivacy_api/migrations/mount_volume.py
View File

@ -5,6 +5,7 @@ from selfprivacy_api.migrations.migration import Migration
from selfprivacy_api.utils import ReadUserData, WriteUserData
from selfprivacy_api.utils.block_devices import BlockDevices
class MountVolume(Migration):
"""Mount volume."""
@ -37,11 +38,13 @@ class MountVolume(Migration):
with WriteUserData() as userdata:
userdata["volumes"] = []
if is_there_a_volume:
userdata["volumes"].append({
"device": "/dev/sdb",
"mountPoint": "/volumes/sdb",
"fsType": "ext4",
})
userdata["volumes"].append(
{
"device": "/dev/sdb",
"mountPoint": "/volumes/sdb",
"fsType": "ext4",
}
)
print("Done")
except Exception as e:
print(e)

13
selfprivacy_api/services/service.py
View File

@ -1,6 +1,7 @@
"""Abstract class for a service running on a server"""
from abc import ABC, abstractmethod
from enum import Enum
import typing
class ServiceStatus(Enum):
@ -13,6 +14,14 @@ class ServiceStatus(Enum):
OFF = "OFF"
class ServiceDnsRecord:
type: str
name: str
content: str
ttl: int
priority: typing.Optional[int]
class Service(ABC):
"""
Service here is some software that is hosted on the server and
@ -78,3 +87,7 @@ class Service(ABC):
@abstractmethod
def get_storage_usage(self):
pass
@abstractmethod
def get_dns_records(self) -> typing.List[ServiceDnsRecord]:
pass

52
selfprivacy_api/utils/block_devices.py
View File

@ -3,6 +3,8 @@ import subprocess
import json
import typing
from selfprivacy_api.utils import WriteUserData
def get_block_device(device_name):
"""
@ -14,7 +16,7 @@ def get_block_device(device_name):
"-J",
"-b",
"-o",
"NAME,PATH,FSAVAIL,FSSIZE,FSTYPE,FSUSED,MOUNTPOINT,LABEL,UUID,SIZE",
"NAME,PATH,FSAVAIL,FSSIZE,FSTYPE,FSUSED,MOUNTPOINT,LABEL,UUID,SIZE, MODEL,SERIAL,TYPE",
device_name,
]
)
@ -49,6 +51,9 @@ class BlockDevice:
self.label = block_device["label"]
self.uuid = block_device["uuid"]
self.size = block_device["size"]
self.model = block_device["model"]
self.serial = block_device["serial"]
self.type = block_device["type"]
self.locked = False
def __str__(self):
@ -76,6 +81,9 @@ class BlockDevice:
self.label = device["label"]
self.uuid = device["uuid"]
self.size = device["size"]
self.model = device["model"]
self.serial = device["serial"]
self.type = device["type"]
return {
"name": self.name,
@ -88,6 +96,9 @@ class BlockDevice:
"label": self.label,
"uuid": self.uuid,
"size": self.size,
"model": self.model,
"serial": self.serial,
"type": self.type,
}
def resize(self):
@ -99,6 +110,40 @@ class BlockDevice:
resize_block_device(self.path)
self.locked = False
def mount(self) -> bool:
"""
Mount the block device.
"""
with WriteUserData() as user_data:
if "volumes" not in user_data:
user_data["volumes"] = []
# Check if the volume is already mounted
for volume in user_data["volumes"]:
if volume["device"] == self.path:
return False
user_data["volumes"].append(
{
"device": self.path,
"mountPoint": f"/volumes/{self.name}",
"fsType": self.fstype,
}
)
return True
def unmount(self) -> bool:
"""
Unmount the block device.
"""
with WriteUserData() as user_data:
if "volumes" not in user_data:
user_data["volumes"] = []
# Check if the volume is already mounted
for volume in user_data["volumes"]:
if volume["device"] == self.path:
user_data["volumes"].remove(volume)
return True
return False
class BlockDevices:
"""Singleton holding all Block devices"""
@ -125,12 +170,15 @@ class BlockDevices:
"-J",
"-b",
"-o",
"NAME,PATH,FSAVAIL,FSSIZE,FSTYPE,FSUSED,MOUNTPOINT,LABEL,UUID,SIZE",
"NAME,PATH,FSAVAIL,FSSIZE,FSTYPE,FSUSED,MOUNTPOINT,LABEL,UUID,SIZE,MODEL,SERIAL,TYPE",
]
)
lsblk_output = lsblk_output.decode("utf-8")
lsblk_output = json.loads(lsblk_output)
for device in lsblk_output["blockdevices"]:
# Ignore devices with type "rom"
if device["type"] == "rom":
continue
if device["fstype"] is None:
if "children" in device:
for child in device["children"]: