def invoke()

in awscli/customizations/sessionmanager.py [0:0]
45 lines of code
14 McCabe index (conditional complexity)

    def invoke(self, service_name, operation_name, parameters,
               parsed_globals):
        client = self._session.create_client(
            service_name, region_name=parsed_globals.region,
            endpoint_url=parsed_globals.endpoint_url,
            verify=parsed_globals.verify_ssl)
        response = client.start_session(**parameters)
        session_id = response['SessionId']
        region_name = client.meta.region_name
        # Profile_name is used to passed on to session manager plugin
        # to fetch same profile credentials to make an api call in the plugin.
        # If --profile flag is configured, pass it to Session Manager plugin.
        # If not, set empty string.
        profile_name = parsed_globals.profile \
            if parsed_globals.profile is not None else ''
        endpoint_url = client.meta.endpoint_url
        ssm_env_name = self.DEFAULT_SSM_ENV_NAME

        try:
            session_parameters = {
                "SessionId": response["SessionId"],
                "TokenValue": response["TokenValue"],
                "StreamUrl": response["StreamUrl"],
            }
            start_session_response = json.dumps(session_parameters)

            plugin_version = check_output(
                ["session-manager-plugin", "--version"], text=True
            )
            env = os.environ.copy()

            # Check if this plugin supports passing the start session response
            # as an environment variable name. If it does, it will set the
            # value to the response from the start_session operation to the env
            # variable defined in DEFAULT_SSM_ENV_NAME. If the session plugin
            # version is invalid or older than the version defined in
            # LAST_PLUGIN_VERSION_WITHOUT_ENV_VAR, it will fall back to
            # passing the start_session response directly.
            version_requirement = VersionRequirement(
                min_version=self.LAST_PLUGIN_VERSION_WITHOUT_ENV_VAR
            )
            if version_requirement.meets_requirement(plugin_version):
                env[ssm_env_name] = start_session_response
                start_session_response = ssm_env_name
            # ignore_user_entered_signals ignores these signals
            # because if signals which kills the process are not
            # captured would kill the foreground process but not the
            # background one. Capturing these would prevents process
            # from getting killed and these signals are input to plugin
            # and handling in there
            with ignore_user_entered_signals():
                # call executable with necessary input
                check_call(["session-manager-plugin",
                            start_session_response,
                            region_name,
                            "StartSession",
                            profile_name,
                            json.dumps(parameters),
                            endpoint_url], env=env)

            return 0
        except OSError as ex:
            if ex.errno == errno.ENOENT:
                logger.debug('SessionManagerPlugin is not present',
                             exc_info=True)
                # start-session api call returns response and starts the
                # session on ssm-agent and response is forwarded to
                # session-manager-plugin. If plugin is not present, terminate
                # is called so that service and ssm-agent terminates the
                # session to avoid zombie session active on ssm-agent for
                # default self terminate time
                client.terminate_session(SessionId=session_id)
                raise ValueError(''.join(ERROR_MESSAGE))