ms/py-executor/test/resources/sample-cba/1.0.0/Scripts/python/sample_script.py

   1 #!/usr/bin/python
   2 #
   3 #  Copyright © 2018-2019 AT&T Intellectual Property.
   4 #
   5 #  Licensed under the Apache License, Version 2.0 (the "License");
   6 #  you may not use this file except in compliance with the License.
   7 #  You may obtain a copy of the License at
   8 #
   9 #      http://www.apache.org/licenses/LICENSE-2.0
  10 #
  11 #  Unless required by applicable law or agreed to in writing, software
  12 #  distributed under the License is distributed on an "AS IS" BASIS,
  13 #  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14 #  See the License for the specific language governing permissions and
  15 #  limitations under the License.
  16
  17 import logging
  18 from blueprints_grpc import executor_utils
  19 from blueprints_grpc.blueprint_processing_server import AbstractScriptFunction
  20 from .module_utils import say_hi
  21 import json
  22
  23
  24 class SampleScript(AbstractScriptFunction):
  25     def __init__(self):
  26         self.logger = logging.getLogger(self.__class__.__name__)
  27
  28     def process(self, execution_request):
  29         self.logger.info("Request Received in Script : {}".format(execution_request))
  30         yield executor_utils.log_response(execution_request, "First message")
  31         yield executor_utils.log_response(execution_request, "Second message")
  32
  33         # TO check , If this class could call other python files.
  34         say_hi()
  35
  36         # Check Yield should be called from other methods.
  37         yield from self.send_notification(execution_request)
  38
  39         response_data = """{
  40         "property" : "value"
  41          }
  42         """
  43         response_payload_json = json.loads(response_data)
  44
  45         yield executor_utils.success_response(execution_request, response_payload_json, 200)
  46
  47     def recover(self, runtime_exception, execution_request):
  48         return None
  49
  50     def send_notification(self, execution_request):
  51         yield executor_utils.send_notification(execution_request, "I am notification")