Skip to content
Snippets Groups Projects

Добавлен базовый класс для сервиса расширений и модуль для установки зависимостей расширения

Merged Podosochnyy Maxim requested to merge feature/add-ext-service into master
All threads resolved!
5 files
+ 667
1
Compare changes
  • Side-by-side
  • Inline
Files
5
+ 126
0
import grpc
from perxis.extensions import extension_pb2, extension_pb2_grpc
from perxis.roles import roles_pb2_grpc, roles_pb2
from perxis.clients import clients_pb2_grpc, clients_pb2
from perxis.collections import collections_pb2_grpc, collections_pb2
from perxis.environments import environments_pb2_grpc
from perxis.extensions.extension_setup import ExtensionSetup
class ExtensionService(extension_pb2_grpc.ExtensionServicer):
extension_id: str
collections: list[collections_pb2.Collection] = []
roles: list[roles_pb2.Role] = []
clients: list[clients_pb2.Client] = []
def __init__(self,
collections_service: collections_pb2_grpc.CollectionsStub,
environments_service: environments_pb2_grpc.EnvironmentsStub,
roles_service: roles_pb2_grpc.RolesStub,
clients_service: clients_pb2_grpc.ClientsStub,
):
self.collections_service = collections_service
self.environments_service = environments_service
self.roles_service = roles_service
self.clients_service = clients_service
self.extension_setup = ExtensionSetup(
self.collections_service, self.environments_service,
self.roles_service, self.clients_service
)
for collection in self.collections or []:
self.extension_setup.add_collection(collection)
for role in self.roles or []:
self.extension_setup.add_role(role)
for client in self.clients or []:
self.extension_setup.add_client(client)
def Install(self, request: extension_pb2.InstallRequest, context):
errors_list = self.extension_setup.install(
request.space_id, request.env_id, request.force
)
if errors_list:
context.set_code(grpc.StatusCode.UNKNOWN)
context.set_details("; ".join(errors_list))
response_state = extension_pb2.ExtensionRequestResult.State.OK \
if not errors_list \
else extension_pb2.ExtensionRequestResult.State.ERROR
return extension_pb2.InstallResponse(
results=[extension_pb2.ExtensionRequestResult(
extension=self.extension_id,
state=response_state,
error="; ".join(errors_list) if errors_list else None,
msg="Ok" if not errors_list else None
)]
)
def Update(self, request: extension_pb2.UpdateRequest, context):
errors_list = self.extension_setup.update(
request.space_id, request.env_id, request.force
)
if errors_list:
context.set_code(grpc.StatusCode.UNKNOWN)
context.set_details("; ".join(errors_list))
response_state = extension_pb2.ExtensionRequestResult.State.OK \
if not errors_list \
else extension_pb2.ExtensionRequestResult.State.ERROR
return extension_pb2.UpdateResponse(
results=[extension_pb2.ExtensionRequestResult(
extension=self.extension_id,
state=response_state,
error="; ".join(errors_list) if errors_list else None,
msg="Ok" if not errors_list else None
)]
)
def Uninstall(self, request: extension_pb2.UninstallRequest, context):
errors_list: list[str] = self.extension_setup.uninstall(request.space_id, request.env_id, request.remove)
response_state = extension_pb2.ExtensionRequestResult.State.OK \
if not errors_list \
else extension_pb2.ExtensionRequestResult.State.ERROR
return extension_pb2.UninstallResponse(
results=[extension_pb2.ExtensionRequestResult(
extension=self.extension_id,
state=response_state,
error="; ".join(errors_list) if errors_list else None,
msg="Ok" if not errors_list else None
)]
)
def Check(self, request: extension_pb2.CheckRequest, context):
errors_list = self.extension_setup.check(request.space_id, request.env_id)
if errors_list:
context.set_code(grpc.StatusCode.UNKNOWN)
context.set_details("; ".join(errors_list))
response_state = extension_pb2.ExtensionRequestResult.State.OK \
if not errors_list \
else extension_pb2.ExtensionRequestResult.State.ERROR
return extension_pb2.CheckResponse(
results=[extension_pb2.ExtensionRequestResult(
extension=self.extension_id,
state=response_state,
error="; ".join(errors_list) if errors_list else None,
msg="Ok" if not errors_list else None
)]
)
def Action(self, request: extension_pb2.ActionRequest, context):
context.set_code(grpc.StatusCode.UNKNOWN)
context.set_details("Unknown action")
return None
Loading