mirror of
https://git.selfprivacy.org/SelfPrivacy/selfprivacy-rest-api.git
synced 2024-11-29 23:41:28 +00:00
118 lines
3.4 KiB
Python
118 lines
3.4 KiB
Python
"""Services status"""
|
|
# pylint: disable=too-few-public-methods
|
|
from enum import Enum
|
|
import typing
|
|
import strawberry
|
|
import datetime
|
|
|
|
from selfprivacy_api.graphql.common_types.dns import DnsRecord
|
|
from selfprivacy_api.graphql.common_types.storage_usage import (
|
|
StorageUsageInterface,
|
|
StorageVolume,
|
|
)
|
|
from selfprivacy_api.services import get_all_services, get_service_by_id
|
|
from selfprivacy_api.services import Service as ServiceInterface
|
|
from selfprivacy_api.utils.block_devices import BlockDevices
|
|
|
|
|
|
@strawberry.enum
|
|
class ServiceStatusEnum(Enum):
|
|
RUNNING = "RUNNING"
|
|
DEGRADED = "DEGRADED"
|
|
ERROR = "ERROR"
|
|
STOPPED = "STOPPED"
|
|
OFF = "OFF"
|
|
|
|
|
|
@strawberry.type
|
|
class ServiceStorageUsage(StorageUsageInterface):
|
|
"""Storage usage for a service"""
|
|
|
|
service: typing.Optional["Service"]
|
|
|
|
|
|
def get_storage_usage(root: "Service") -> ServiceStorageUsage:
|
|
"""Get storage usage for a service"""
|
|
service = get_service_by_id(root.id)
|
|
if service is None:
|
|
return ServiceStorageUsage(
|
|
service=service,
|
|
title="Not found",
|
|
used_space="0",
|
|
volume=get_volume_by_id("sda1"),
|
|
)
|
|
return ServiceStorageUsage(
|
|
service=service_to_graphql_service(service),
|
|
title=service.get_display_name(),
|
|
used_space=str(service.get_storage_usage()),
|
|
volume=get_volume_by_id(service.get_location()),
|
|
)
|
|
|
|
|
|
@strawberry.type
|
|
class Service:
|
|
storage_usage: ServiceStorageUsage = strawberry.field(resolver=get_storage_usage)
|
|
id: str
|
|
display_name: str
|
|
description: str
|
|
svg_icon: str
|
|
is_movable: bool
|
|
is_required: bool
|
|
is_enabled: bool
|
|
status: ServiceStatusEnum
|
|
url: typing.Optional[str]
|
|
dns_records: typing.Optional[typing.List[DnsRecord]]
|
|
|
|
|
|
def service_to_graphql_service(service: ServiceInterface) -> Service:
|
|
"""Convert service to graphql service"""
|
|
return Service(
|
|
id=service.get_id(),
|
|
display_name=service.get_display_name(),
|
|
description=service.get_description(),
|
|
svg_icon=service.get_svg_icon(),
|
|
is_movable=service.is_movable(),
|
|
is_required=service.is_required(),
|
|
is_enabled=service.is_enabled(),
|
|
status=ServiceStatusEnum(service.get_status().value),
|
|
url=service.get_url(),
|
|
dns_records=[
|
|
DnsRecord(
|
|
record_type=record.type,
|
|
name=record.name,
|
|
content=record.content,
|
|
ttl=record.ttl,
|
|
priority=record.priority,
|
|
)
|
|
for record in service.get_dns_records()
|
|
],
|
|
)
|
|
|
|
|
|
def get_volume_by_id(volume_id: str) -> typing.Optional[StorageVolume]:
|
|
"""Get volume by id"""
|
|
volume = BlockDevices().get_block_device(volume_id)
|
|
if volume is None:
|
|
return None
|
|
return StorageVolume(
|
|
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,
|
|
usages=[],
|
|
)
|
|
|
|
|
|
@strawberry.type
|
|
class Services:
|
|
@strawberry.field
|
|
def all_services(self, info) -> typing.List[Service]:
|
|
services = get_all_services()
|
|
return [service_to_graphql_service(service) for service in services]
|