# encoding: utf-8
# pylint: disable=no-member
# type: ignore
from collections import namedtuple
from pathlib import PurePosixPath
[docs]
class UrlMixIn:
__slots__ = ()
def _url2name(self, url):
if not url.startswith(self.url):
raise ValueError(f'{url} is not in {self.url}')
return url.replace(self.url, '/').replace('/job/', '/').strip('/')
def _name2url(self, full_name):
if not full_name:
return self.url
full_name = full_name.strip('/').replace('/', '/job/')
return f'{self.url}job/{full_name}/'
def _parse_name(self, full_name):
if full_name.startswith(('http://', 'https://')):
full_name = self._url2name(full_name)
path = PurePosixPath(full_name)
parent = str(path.parent) if path.parent.name else ''
return parent, path.name
[docs]
class DeletionMixIn:
__slots__ = ()
[docs]
def delete(self):
self.handle_req('POST', 'doDelete')
[docs]
class ConfigurationMixIn:
__slots__ = ()
@property
def name(self):
return self.url.split('/')[-2]
[docs]
class DescriptionMixIn:
__slots__ = ()
[docs]
def set_description(self, text):
self.handle_req('POST', 'submitDescription',
params={'description': text})
[docs]
class RunScriptMixIn:
__slots__ = ()
[docs]
def run_script(self, script):
return self.handle_req('POST', 'scriptText',
data={'script': script}).text
[docs]
class EnableMixIn:
__slots__ = ()
[docs]
def enable(self):
return self.handle_req('POST', 'enable')
[docs]
def disable(self):
return self.handle_req('POST', 'disable')
[docs]
class RawJsonMixIn:
__slots__ = ()
[docs]
def api_json(self, tree='', depth=0):
return self.raw
Parameter = namedtuple('Parameter', ['class_name', 'name', 'value'])
[docs]
class ActionsMixIn:
__slots__ = ()
[docs]
def get_parameters(self):
parameters = []
for action in self.api_json()['actions']:
if 'parameters' in action:
parameters.extend(Parameter(raw['_class'], raw['name'], raw.get(
'value', '')) for raw in action['parameters'])
break
return parameters
[docs]
def get_causes(self):
return next((action['causes'] for action in self.api_json()['actions'] if 'causes' in action), [])
# async classes
[docs]
class AsyncDeletionMixIn:
__slots__ = ()
[docs]
async def delete(self):
await self.handle_req('POST', 'doDelete')
[docs]
class AsyncConfigurationMixIn:
__slots__ = ()
@property
def name(self):
return self.url.split('/')[-2]
[docs]
class AsyncDescriptionMixIn:
__slots__ = ()
[docs]
async def set_description(self, text):
await self.handle_req('POST', 'submitDescription',
params={'description': text})
[docs]
class AsyncRunScriptMixIn:
__slots__ = ()
[docs]
async def run_script(self, script):
return (await self.handle_req('POST', 'scriptText', data={'script': script})).text
[docs]
class AsyncEnableMixIn:
__slots__ = ()
[docs]
async def enable(self):
return await self.handle_req('POST', 'enable')
[docs]
async def disable(self):
return await self.handle_req('POST', 'disable')
[docs]
class AsyncRawJsonMixIn:
__slots__ = ()
[docs]
async def api_json(self, tree='', depth=0):
return self.raw
[docs]
class AsyncActionsMixIn:
__slots__ = ()
[docs]
async def get_parameters(self):
parameters = []
data = await self.api_json()
for action in data['actions']:
if 'parameters' in action:
parameters.extend(Parameter(raw['_class'], raw['name'], raw.get(
'value', '')) for raw in action['parameters'])
break
return parameters
[docs]
async def get_causes(self):
data = await self.api_json()
return next((action['causes'] for action in data['actions'] if 'causes' in action), [])