Source code for api4jenkins.input
# encoding: utf-8
import json
from .item import AsyncItem, Item
from .mix import AsyncRawJsonMixIn, RawJsonMixIn
[docs]
class PendingInputAction(RawJsonMixIn, Item):
''' this class implement functionality to process
`input step <https://www.jenkins.io/doc/pipeline/steps/pipeline-input-step/>`_
'''
def __init__(self, jenkins, raw):
super().__init__(
jenkins, f"{jenkins.url}{raw['abortUrl'].rstrip('abort').lstrip('/')}")
self.raw = raw
self.raw['_class'] = 'PendingInputAction'
[docs]
def abort(self):
'''submit `input step <https://www.jenkins.io/doc/pipeline/steps/pipeline-input-step/>`_'''
return self.handle_req('POST', 'abort')
[docs]
def submit(self, **params):
'''submit `input step <https://www.jenkins.io/doc/pipeline/steps/pipeline-input-step/>`_
- if input requires parameters:
- if submit without parameters, it will use default value of parameters
- if submit with wrong parameters, exception raised
- if input does not requires parameters, but submit with parameters, exception raised
'''
if params:
data = _make_input_params(self.raw, **params)
return self.handle_req('POST', 'submit', data=data)
return self.handle_req('POST', 'proceedEmpty')
def _make_input_params(api_json, **params):
input_args = [input['name'] for input in api_json['inputs']]
params_keys = list(params.keys())
if not input_args:
raise TypeError(f'input takes 0 argument, but got {params_keys}')
if any(k not in input_args for k in params_keys):
raise TypeError(
f'input takes arguments: {input_args}, but got {params_keys}')
params = [{'name': k, 'value': v} for k, v in params.items()]
return {'proceed': api_json['proceedText'],
'json': json.dumps({'parameter': params})}
[docs]
class AsyncPendingInputAction(AsyncRawJsonMixIn, AsyncItem):
def __init__(self, jenkins, raw):
super().__init__(
jenkins, f"{jenkins.url}{raw['abortUrl'].rstrip('abort').lstrip('/')}")
self.raw = raw
self.raw['_class'] = 'AsyncPendingInputAction'
[docs]
async def submit(self, **params):
if params:
data = _make_input_params(self.raw, **params)
return await self.handle_req('POST', 'submit', data=data)
return await self.handle_req('POST', 'proceedEmpty')