mirror of
https://git.selfprivacy.org/SelfPrivacy/selfprivacy-rest-api.git
synced 2024-08-19 15:02:58 +00:00
Co-authored-by: inexcode <inex.code@selfprivacy.org> Reviewed-on: https://git.selfprivacy.org/SelfPrivacy/selfprivacy-rest-api/pulls/13
170 lines
5.3 KiB
Python
170 lines
5.3 KiB
Python
"""Services mutations"""
|
|
# pylint: disable=too-few-public-methods
|
|
import typing
|
|
import strawberry
|
|
from selfprivacy_api.graphql import IsAuthenticated
|
|
from selfprivacy_api.graphql.common_types.jobs import job_to_api_job
|
|
|
|
from selfprivacy_api.graphql.common_types.service import (
|
|
Service,
|
|
service_to_graphql_service,
|
|
)
|
|
from selfprivacy_api.graphql.mutations.mutation_interface import (
|
|
GenericJobButationReturn,
|
|
GenericMutationReturn,
|
|
)
|
|
|
|
from selfprivacy_api.services import get_service_by_id
|
|
from selfprivacy_api.utils.block_devices import BlockDevices
|
|
|
|
|
|
@strawberry.type
|
|
class ServiceMutationReturn(GenericMutationReturn):
|
|
"""Service mutation return type."""
|
|
|
|
service: typing.Optional[Service] = None
|
|
|
|
|
|
@strawberry.input
|
|
class MoveServiceInput:
|
|
"""Move service input type."""
|
|
|
|
service_id: str
|
|
location: str
|
|
|
|
|
|
@strawberry.type
|
|
class ServiceJobMutationReturn(GenericJobButationReturn):
|
|
"""Service job mutation return type."""
|
|
|
|
service: typing.Optional[Service] = None
|
|
|
|
|
|
@strawberry.type
|
|
class ServicesMutations:
|
|
"""Services mutations."""
|
|
|
|
@strawberry.mutation(permission_classes=[IsAuthenticated])
|
|
def enable_service(self, service_id: str) -> ServiceMutationReturn:
|
|
"""Enable service."""
|
|
service = get_service_by_id(service_id)
|
|
if service is None:
|
|
return ServiceMutationReturn(
|
|
success=False,
|
|
message="Service not found.",
|
|
code=404,
|
|
)
|
|
service.enable()
|
|
return ServiceMutationReturn(
|
|
success=True,
|
|
message="Service enabled.",
|
|
code=200,
|
|
service=service_to_graphql_service(service),
|
|
)
|
|
|
|
@strawberry.mutation(permission_classes=[IsAuthenticated])
|
|
def disable_service(self, service_id: str) -> ServiceMutationReturn:
|
|
"""Disable service."""
|
|
service = get_service_by_id(service_id)
|
|
if service is None:
|
|
return ServiceMutationReturn(
|
|
success=False,
|
|
message="Service not found.",
|
|
code=404,
|
|
)
|
|
service.disable()
|
|
return ServiceMutationReturn(
|
|
success=True,
|
|
message="Service disabled.",
|
|
code=200,
|
|
service=service_to_graphql_service(service),
|
|
)
|
|
|
|
@strawberry.mutation(permission_classes=[IsAuthenticated])
|
|
def stop_service(self, service_id: str) -> ServiceMutationReturn:
|
|
"""Stop service."""
|
|
service = get_service_by_id(service_id)
|
|
if service is None:
|
|
return ServiceMutationReturn(
|
|
success=False,
|
|
message="Service not found.",
|
|
code=404,
|
|
)
|
|
service.stop()
|
|
return ServiceMutationReturn(
|
|
success=True,
|
|
message="Service stopped.",
|
|
code=200,
|
|
service=service_to_graphql_service(service),
|
|
)
|
|
|
|
@strawberry.mutation(permission_classes=[IsAuthenticated])
|
|
def start_service(self, service_id: str) -> ServiceMutationReturn:
|
|
"""Start service."""
|
|
service = get_service_by_id(service_id)
|
|
if service is None:
|
|
return ServiceMutationReturn(
|
|
success=False,
|
|
message="Service not found.",
|
|
code=404,
|
|
)
|
|
service.start()
|
|
return ServiceMutationReturn(
|
|
success=True,
|
|
message="Service started.",
|
|
code=200,
|
|
service=service_to_graphql_service(service),
|
|
)
|
|
|
|
@strawberry.mutation(permission_classes=[IsAuthenticated])
|
|
def restart_service(self, service_id: str) -> ServiceMutationReturn:
|
|
"""Restart service."""
|
|
service = get_service_by_id(service_id)
|
|
if service is None:
|
|
return ServiceMutationReturn(
|
|
success=False,
|
|
message="Service not found.",
|
|
code=404,
|
|
)
|
|
service.restart()
|
|
return ServiceMutationReturn(
|
|
success=True,
|
|
message="Service restarted.",
|
|
code=200,
|
|
service=service_to_graphql_service(service),
|
|
)
|
|
|
|
@strawberry.mutation(permission_classes=[IsAuthenticated])
|
|
def move_service(self, input: MoveServiceInput) -> ServiceJobMutationReturn:
|
|
"""Move service."""
|
|
service = get_service_by_id(input.service_id)
|
|
if service is None:
|
|
return ServiceJobMutationReturn(
|
|
success=False,
|
|
message="Service not found.",
|
|
code=404,
|
|
)
|
|
if not service.is_movable():
|
|
return ServiceJobMutationReturn(
|
|
success=False,
|
|
message="Service is not movable.",
|
|
code=400,
|
|
service=service_to_graphql_service(service),
|
|
)
|
|
volume = BlockDevices().get_block_device(input.location)
|
|
if volume is None:
|
|
return ServiceJobMutationReturn(
|
|
success=False,
|
|
message="Volume not found.",
|
|
code=404,
|
|
service=service_to_graphql_service(service),
|
|
)
|
|
job = service.move_to_volume(volume)
|
|
return ServiceJobMutationReturn(
|
|
success=True,
|
|
message="Service moved.",
|
|
code=200,
|
|
service=service_to_graphql_service(service),
|
|
job=job_to_api_job(job),
|
|
)
|