# encoding: utf-8
import json
import xml.etree.ElementTree as ET
from functools import partial
from pathlib import PurePosixPath
from urllib.parse import unquote_plus
from .credential import AsyncCredentials, Credentials
from .item import AsyncItem, Item, append_slash, new_item, snake
from .mix import (AsyncConfigurationMixIn, AsyncDeletionMixIn,
AsyncDescriptionMixIn, AsyncEnableMixIn, ConfigurationMixIn,
DeletionMixIn, DescriptionMixIn, EnableMixIn)
from .queue import AsyncQueueItem, QueueItem
from .view import Views
[docs]
class NameMixIn:
# pylint: disable=no-member
@property
def full_name(self):
return unquote_plus(self.jenkins._url2name(self.url))
@property
def full_display_name(self):
return unquote_plus(self.full_name.replace('/', ' ยป '))
[docs]
class Job(Item, ConfigurationMixIn, DescriptionMixIn, DeletionMixIn, NameMixIn):
[docs]
def move(self, path):
path = path.strip('/')
params = {'destination': f'/{path}',
'json': json.dumps({'destination': f'/{path}'})}
resp = self.handle_req('POST', 'move/move', data=params)
self.url = resp.headers['Location']
return resp
[docs]
def rename(self, name):
resp = self.handle_req('POST', 'confirmRename',
params={'newName': name})
self.url = append_slash(resp.headers['Location'])
return resp
[docs]
def duplicate(self, path, recursive=False):
self.jenkins.create_job(path, self.configure(), recursive=recursive)
@property
def parent(self):
path = PurePosixPath(self.full_name)
if path.parent.name == '':
return self.jenkins
return self.jenkins.get_job(str(path.parent))
def _make_query(depth):
query = 'jobs[url]'
for _ in range(int(depth)):
query = f'jobs[url,{query}]'
return query
def _iter_jobs(jenkins, item):
yield new_item(jenkins, __name__, item)
if jobs := item.get('jobs'):
for job in jobs:
yield from _iter_jobs(jenkins, job)
[docs]
class Folder(Job):
[docs]
def create(self, name, xml):
return self.handle_req('POST', 'createItem', params={'name': name},
headers=self.headers, content=xml)
[docs]
def get(self, name):
for item in self.api_json(tree='jobs[name,url]')['jobs']:
if name == item['name']:
return self._new_item(__name__, item)
return None
[docs]
def iter(self, depth=0):
for item in self.api_json(tree=_make_query(depth))['jobs']:
yield from _iter_jobs(self.jenkins, item)
[docs]
def copy(self, src, dest):
params = {'name': dest, 'mode': 'copy', 'from': src}
return self.handle_req('POST', 'createItem', params=params)
[docs]
def reload(self):
return self.handle_req('POST', 'reload')
@property
def views(self):
return Views(self)
@property
def credentials(self):
return Credentials(self.jenkins, f'{self.url}credentials/store/folder/')
def __call__(self, depth):
yield from self.iter(depth)
[docs]
class WorkflowMultiBranchProject(Folder, EnableMixIn):
[docs]
def scan(self, delay=0):
return self.handle_req('POST', 'build', params={'delay': delay})
[docs]
def get_scan_log(self):
with self.handle_stream('GET', 'indexing/consoleText') as resp:
yield from resp.iter_lines()
@property
def buildable(self):
return ET.XML(self.configure()).find('disabled').text == 'false'
[docs]
class OrganizationFolder(WorkflowMultiBranchProject):
pass
def _set_get_methods(job, func):
for key in ['firstBuild', 'lastBuild', 'lastCompletedBuild',
'lastFailedBuild', 'lastStableBuild', 'lastUnstableBuild',
'lastSuccessfulBuild', 'lastUnsuccessfulBuild']:
setattr(job, snake(f'get_{key}'), partial(func, key))
def _get_build(job, api_json, number):
for item in api_json['builds']:
if number in [item['number'], item['displayName']]:
return job._new_item('api4jenkins.build', item)
[docs]
class Project(Job, EnableMixIn):
def __init__(self, jenkins, url):
super().__init__(jenkins, url)
def _get_build_by_key(key):
item = self.api_json(tree=f'{key}[url]')[key]
if item:
return self._new_item('api4jenkins.build', item)
_set_get_methods(self, _get_build_by_key)
[docs]
def build(self, **params):
reserved = ['token', 'delay']
if not params or all(k in reserved for k in params):
entry = 'build'
else:
entry = 'buildWithParameters'
resp = self.handle_req('POST', entry, params=params)
return QueueItem(self.jenkins, resp.headers['Location'])
[docs]
def get(self, number):
return _get_build(self, self.api_json(tree='builds[number,displayName,url]'), number)
[docs]
def iter(self):
for item in self.api_json(tree='builds[number,url]')['builds']:
yield self._new_item('api4jenkins.build', item)
[docs]
def iter_all_builds(self):
for item in self.api_json(tree='allBuilds[number,url]')['allBuilds']:
yield self._new_item('api4jenkins.build', item)
[docs]
def set_next_build_number(self, number):
self.handle_req('POST', 'nextbuildnumber/submit',
params={'nextBuildNumber': number})
[docs]
def get_parameters(self):
params = []
for p in self.api_json()['property']:
if 'parameterDefinitions' in p:
params = p['parameterDefinitions']
return params
@property
def building(self):
builds = self.api_json(tree='builds[building]')['builds']
return any(b['building'] for b in builds)
[docs]
def filter_builds_by_result(self, *, result):
"""filter build by build results, avaliable results are:
'SUCCESS', 'UNSTABLE', 'FAILURE', 'NOT_BUILT', 'ABORTED'
see: https://javadoc.jenkins-ci.org/hudson/model/Result.html
"""
expect = ['SUCCESS', 'UNSTABLE', 'FAILURE', 'NOT_BUILT', 'ABORTED']
if result not in expect:
raise ValueError(f'Expect one of {expect}')
yield from filter(lambda build: build.result == result, self)
[docs]
class WorkflowJob(Project):
pass
[docs]
class MatrixProject(Project):
pass
[docs]
class FreeStyleProject(Project):
pass
[docs]
class MavenModuleSet(Project):
pass
[docs]
class ExternalJob(Project):
pass
[docs]
class MultiJobProject(Project):
pass
[docs]
class IvyModuleSet(Project):
pass
[docs]
class BitbucketSCMNavigator(Project):
pass
[docs]
class GitHubSCMNavigator(Project):
pass
[docs]
class PipelineMultiBranchDefaultsProject(Project):
pass
# async class
[docs]
class AsyncJob(AsyncItem, AsyncConfigurationMixIn, AsyncDescriptionMixIn, AsyncDeletionMixIn, NameMixIn):
[docs]
async def move(self, path):
path = path.strip('/')
params = {'destination': f'/{path}',
'json': json.dumps({'destination': f'/{path}'})}
resp = await self.handle_req('POST', 'move/move', data=params)
self.url = resp.headers['Location']
return resp
[docs]
async def rename(self, name):
resp = await self.handle_req('POST', 'confirmRename',
params={'newName': name})
self.url = append_slash(resp.headers['Location'])
return resp
[docs]
async def duplicate(self, path, recursive=False):
await self.jenkins.create_job(path, await self.configure(), recursive=recursive)
@property
async def parent(self):
path = PurePosixPath(self.full_name)
if path.parent.name == '':
return self.jenkins
return await self.jenkins.get_job(str(path.parent))
[docs]
class AsyncFolder(AsyncJob):
[docs]
async def create(self, name, xml):
return await self.handle_req('POST', 'createItem', params={'name': name},
headers=self.headers, content=xml)
[docs]
async def get(self, name):
resp = await self.api_json(tree='jobs[name,url]')
for item in resp['jobs']:
if name == item['name']:
return self._new_item(__name__, item)
return None
[docs]
async def aiter(self, depth=0):
for item in (await self.api_json(tree=_make_query(depth)))['jobs']:
for job in _iter_jobs(self.jenkins, item):
yield job
[docs]
async def copy(self, src, dest):
params = {'name': dest, 'mode': 'copy', 'from': src}
return await self.handle_req('POST', 'createItem', params=params)
[docs]
async def reload(self):
return await self.handle_req('POST', 'reload')
@property
def views(self):
return Views(self)
@property
def credentials(self):
return AsyncCredentials(self.jenkins,
f'{self.url}credentials/store/folder/')
async def __call__(self, depth):
async for job in self.aiter(depth):
yield job
[docs]
class AsyncWorkflowMultiBranchProject(AsyncFolder, AsyncEnableMixIn):
[docs]
async def scan(self, delay=0):
return await self.handle_req('POST', 'build', params={'delay': delay})
[docs]
async def get_scan_log(self):
async with self.handle_stream('GET', 'indexing/consoleText') as resp:
async for line in resp.aiter_lines():
yield line
@property
async def buildable(self):
return ET.XML(await self.configure()).find('disabled').text == 'false'
[docs]
class AsyncOrganizationFolder(AsyncWorkflowMultiBranchProject):
pass
[docs]
class AsyncProject(AsyncJob, AsyncEnableMixIn):
def __init__(self, jenkins, url):
super().__init__(jenkins, url)
async def _get_build_by_key(key):
item = (await self.api_json(tree=f'{key}[url]'))[key]
if item:
return self._new_item('api4jenkins.build', item)
_set_get_methods(self, _get_build_by_key)
[docs]
async def build(self, **params):
reserved = ['token', 'delay']
if not params or all(k in reserved for k in params):
entry = 'build'
else:
entry = 'buildWithParameters'
resp = await self.handle_req('POST', entry, params=params)
return AsyncQueueItem(self.jenkins, resp.headers['Location'])
[docs]
async def get(self, number):
return _get_build(self, await self.api_json(tree='builds[number,displayName,url]'), number)
[docs]
async def aiter(self):
data = await self.api_json(tree='builds[number,url]')
for item in data['builds']:
yield self._new_item('api4jenkins.build', item)
[docs]
async def iter_all_builds(self):
data = await self.api_json(tree='allBuilds[number,url]')
for item in data['allBuilds']:
yield self._new_item('api4jenkins.build', item)
[docs]
async def set_next_build_number(self, number):
await self.handle_req('POST', 'nextbuildnumber/submit',
params={'nextBuildNumber': number})
[docs]
async def get_parameters(self):
params = []
for p in (await self.api_json())['property']:
if 'parameterDefinitions' in p:
params = p['parameterDefinitions']
return params
@property
async def building(self):
data = await self.api_json(tree='builds[building]')
return any(b['building'] for b in data['builds'])
[docs]
async def filter_builds_by_result(self, *, result):
"""filter build by build results, avaliable results are:
'SUCCESS', 'UNSTABLE', 'FAILURE', 'NOT_BUILT', 'ABORTED'
see: https://javadoc.jenkins-ci.org/hudson/model/Result.html
"""
expect = ['SUCCESS', 'UNSTABLE', 'FAILURE', 'NOT_BUILT', 'ABORTED']
if result not in expect:
raise ValueError(f'Expect one of {expect}')
async for build in self:
if await build.result == result:
yield build
[docs]
class AsyncWorkflowJob(AsyncProject):
pass
[docs]
class AsyncMatrixProject(AsyncProject):
pass
[docs]
class AsyncFreeStyleProject(AsyncProject):
pass
[docs]
class AsyncMavenModuleSet(AsyncProject):
pass
[docs]
class AsyncExternalJob(AsyncProject):
pass
[docs]
class AsyncMultiJobProject(AsyncProject):
pass
[docs]
class AsyncIvyModuleSet(AsyncProject):
pass
[docs]
class AsyncBitbucketSCMNavigator(AsyncProject):
pass
[docs]
class AsyncGitHubSCMNavigator(AsyncProject):
pass
[docs]
class AsyncPipelineMultiBranchDefaultsProject(AsyncProject):
pass