refactor(service): break out ServiceStatus and ServiceDNSRecord
parent
962e8d5ca7
commit
f333e791e1
|
@ -0,0 +1,24 @@
|
||||||
|
from enum import Enum
|
||||||
|
from typing import Optional
|
||||||
|
from pydantic import BaseModel
|
||||||
|
|
||||||
|
|
||||||
|
class ServiceStatus(Enum):
|
||||||
|
"""Enum for service status"""
|
||||||
|
|
||||||
|
ACTIVE = "ACTIVE"
|
||||||
|
RELOADING = "RELOADING"
|
||||||
|
INACTIVE = "INACTIVE"
|
||||||
|
FAILED = "FAILED"
|
||||||
|
ACTIVATING = "ACTIVATING"
|
||||||
|
DEACTIVATING = "DEACTIVATING"
|
||||||
|
OFF = "OFF"
|
||||||
|
|
||||||
|
|
||||||
|
class ServiceDnsRecord(BaseModel):
|
||||||
|
type: str
|
||||||
|
name: str
|
||||||
|
content: str
|
||||||
|
ttl: int
|
||||||
|
display_name: str
|
||||||
|
priority: Optional[int] = None
|
|
@ -1,13 +1,15 @@
|
||||||
"""Abstract class for a service running on a server"""
|
"""Abstract class for a service running on a server"""
|
||||||
from abc import ABC, abstractmethod
|
from abc import ABC, abstractmethod
|
||||||
from enum import Enum
|
|
||||||
from typing import List, Optional
|
from typing import List, Optional
|
||||||
|
|
||||||
from pydantic import BaseModel
|
from selfprivacy_api import utils
|
||||||
from selfprivacy_api.jobs import Job, Jobs, JobStatus, report_progress
|
from selfprivacy_api.utils import ReadUserData, WriteUserData
|
||||||
|
from selfprivacy_api.utils.waitloop import wait_until_true
|
||||||
from selfprivacy_api.utils.block_devices import BlockDevice, BlockDevices
|
from selfprivacy_api.utils.block_devices import BlockDevice, BlockDevices
|
||||||
|
|
||||||
|
from selfprivacy_api.jobs import Job, Jobs, JobStatus, report_progress
|
||||||
|
|
||||||
|
from selfprivacy_api.models.services import ServiceStatus, ServiceDnsRecord
|
||||||
from selfprivacy_api.services.generic_size_counter import get_storage_usage
|
from selfprivacy_api.services.generic_size_counter import get_storage_usage
|
||||||
from selfprivacy_api.services.owned_path import OwnedPath, Bind
|
from selfprivacy_api.services.owned_path import OwnedPath, Bind
|
||||||
from selfprivacy_api.services.moving import (
|
from selfprivacy_api.services.moving import (
|
||||||
|
@ -20,34 +22,10 @@ from selfprivacy_api.services.moving import (
|
||||||
move_data_to_volume,
|
move_data_to_volume,
|
||||||
)
|
)
|
||||||
|
|
||||||
from selfprivacy_api import utils
|
|
||||||
from selfprivacy_api.utils.waitloop import wait_until_true
|
|
||||||
from selfprivacy_api.utils import ReadUserData, WriteUserData
|
|
||||||
|
|
||||||
DEFAULT_START_STOP_TIMEOUT = 5 * 60
|
DEFAULT_START_STOP_TIMEOUT = 5 * 60
|
||||||
|
|
||||||
|
|
||||||
class ServiceStatus(Enum):
|
|
||||||
"""Enum for service status"""
|
|
||||||
|
|
||||||
ACTIVE = "ACTIVE"
|
|
||||||
RELOADING = "RELOADING"
|
|
||||||
INACTIVE = "INACTIVE"
|
|
||||||
FAILED = "FAILED"
|
|
||||||
ACTIVATING = "ACTIVATING"
|
|
||||||
DEACTIVATING = "DEACTIVATING"
|
|
||||||
OFF = "OFF"
|
|
||||||
|
|
||||||
|
|
||||||
class ServiceDnsRecord(BaseModel):
|
|
||||||
type: str
|
|
||||||
name: str
|
|
||||||
content: str
|
|
||||||
ttl: int
|
|
||||||
display_name: str
|
|
||||||
priority: Optional[int] = None
|
|
||||||
|
|
||||||
|
|
||||||
class Service(ABC):
|
class Service(ABC):
|
||||||
"""
|
"""
|
||||||
Service here is some software that is hosted on the server and
|
Service here is some software that is hosted on the server and
|
||||||
|
|
|
@ -2,16 +2,16 @@
|
||||||
import subprocess
|
import subprocess
|
||||||
from typing import List
|
from typing import List
|
||||||
|
|
||||||
from selfprivacy_api.services.service import ServiceStatus
|
from selfprivacy_api.models.services import ServiceStatus
|
||||||
|
|
||||||
|
|
||||||
def get_service_status(service: str) -> ServiceStatus:
|
def get_service_status(unit: str) -> ServiceStatus:
|
||||||
"""
|
"""
|
||||||
Return service status from systemd.
|
Return service status from systemd.
|
||||||
Use systemctl show to get the status of a service.
|
Use systemctl show to get the status of a service.
|
||||||
Get ActiveState from the output.
|
Get ActiveState from the output.
|
||||||
"""
|
"""
|
||||||
service_status = subprocess.check_output(["systemctl", "show", service])
|
service_status = subprocess.check_output(["systemctl", "show", unit])
|
||||||
if b"LoadState=not-found" in service_status:
|
if b"LoadState=not-found" in service_status:
|
||||||
return ServiceStatus.OFF
|
return ServiceStatus.OFF
|
||||||
if b"ActiveState=active" in service_status:
|
if b"ActiveState=active" in service_status:
|
||||||
|
|
Loading…
Reference in New Issue