# encoding: utf-8
import asyncio
import threading
import typing
from importlib import import_module
from httpx import HTTPStatusError
from api4jenkins.mix import UrlMixIn
from .credential import AsyncCredentials, Credentials
from .exceptions import ItemNotFoundError
from .http import new_async_http_client, new_http_client
from .item import AsyncItem, Item
from .job import AsyncFolder, AsyncProject, Folder
from .node import AsyncNodes, Nodes
from .plugin import AsyncPluginsManager, PluginsManager
from .queue import AsyncQueue, Queue
from .system import AsyncSystem, System
from .user import AsyncUser, AsyncUsers, User, Users
from .view import AsyncViews, Views
EMPTY_FOLDER_XML = '''<?xml version='1.0' encoding='UTF-8'?>
<com.cloudbees.hudson.plugins.folder.Folder/>'''
[docs]
class Jenkins(Item, UrlMixIn):
r'''Constructs :class:`Jenkins <Jenkins>`.
:param url: URL of Jenkins server, ``str``
:param auth: (optional) Auth ``tuple`` to enable Basic/Digest/Custom HTTP Auth.
:param \*\*kwargs: other kwargs are same as `httpx.Client <https://www.python-httpx.org/api/#client>`_
Usage::
>>> from api4jenkins import Jenkins
>>> j = Jenkins('http://127.0.0.1:8080/', auth=('admin', 'admin'))
>>> print(j)
<Jenkins: http://127.0.0.1:8080/>
>>> j.version
'2.176.2'
'''
def __init__(self, url, **kwargs):
self.http_client = new_http_client(**kwargs)
self._crumb = None
self._auth = kwargs.get('auth')
self._sync_lock = threading.Lock()
super().__init__(self, url)
self.user = User(
self, f'{self.url}user/{self._auth[0]}/') if self._auth else None
[docs]
def get_job(self, full_name):
'''Get job by full name
:param full_name: ``str``, full name of job
:returns: Corresponding Job object or None
Usage::
>>> from api4jenkins import Jenkins
>>> j = Jenkins('http://127.0.0.1:8080/', auth=('admin', 'admin'))
>>> job = j.get_job('freestylejob')
>>> print(job)
<FreeStyleProject: http://127.0.0.1:8080/job/freestylejob/>
'''
folder, name = self._resolve_name(full_name)
if folder.exists():
return folder.get(name)
[docs]
def iter(self, depth=0):
'''Iterate jobs with depth
:param depth: ``int``, depth to iterate, default is 0
:returns: iterator of jobs
Usage::
>>> from api4jenkins import Jenkins
>>> j = Jenkins('http://127.0.0.1:8080/', auth=('admin', 'admin'))
>>> for job in j.iter():
... print(job)
<FreeStyleProject: http://127.0.0.1:8080/job/freestylejob/>
...
'''
yield from Folder(self, self.url)(depth)
[docs]
def create_job(self, full_name, xml, recursive=False):
'''Create new jenkins job with given xml configuration
:param full_name: ``str``, full name of job
:param xml: xml configuration string
:param recursive: (optional) Boolean, recursively create folder if not existed
Usage::
>>> from api4jenkins import Jenkins
>>> j = Jenkins('http://127.0.0.1:8080/', auth=('admin', 'admin'))
>>> xml = """<?xml version='1.1' encoding='UTF-8'?>
... <project>
... <builders>
... <hudson.tasks.Shell>
... <command>echo $JENKINS_VERSION</command>
... </hudson.tasks.Shell>
... </builders>
... </project>"""
>>> j.create_job('freestylejob', xml)
>>> job = j.get_job('freestylejob')
>>> print(job)
<FreeStyleProject: http://127.0.0.1:8080/job/freestylejob/>
'''
folder, name = self._resolve_name(full_name)
if recursive and not folder.exists():
self.create_job(folder.full_name, EMPTY_FOLDER_XML,
recursive=recursive)
return folder.create(name, xml)
[docs]
def copy_job(self, full_name, dest):
'''Create job by copying other job, the source job and dest job are in
same folder.
:param full_name: full name of source job
:param dest: name of new job
Usage::
>>> from api4jenkins import Jenkins
>>> j = Jenkins('http://127.0.0.1:8080/', auth=('admin', 'admin'))
>>> j.copy_job('folder/freestylejob', 'newjob')
>>> j.get_job('folder/newjob')
>>> print(job)
<FreeStyleProject: http://127.0.0.1:8080/job/folder/job/newjob/>
'''
folder, name = self._resolve_name(full_name)
return folder.copy(name, dest)
[docs]
def delete_job(self, full_name):
'''Delete job
:param full_name: ``str``, full name of job
Usage::
>>> from api4jenkins import Jenkins
>>> j = Jenkins('http://127.0.0.1:8080/', auth=('admin', 'admin'))
>>> job = j.get_job('freestylejob')
>>> print(job)
<FreeStyleProject: http://127.0.0.1:8080/job/freestylejob/>
>>> j.delete_job('freestylejob')
>>> job = j.get_job('freestylejob')
>>> print(job)
None
'''
if job := self.get_job(full_name):
job.delete()
[docs]
def build_job(self, full_name, **params):
'''Build job with/without params
:param full_name: ``str``, full name of job
:param params: parameters for building, support delay and remote token
:returns: ``QueueItem``
Usage::
>>> from api4jenkins import Jenkins
>>> j = Jenkins('http://127.0.0.1:8080/', auth=('admin', 'admin'))
>>> item = j.build_job('freestylejob')
>>> import time
>>> while not item.get_build():
... time.sleep(1)
>>> build = item.get_build()
>>> print(build)
<FreeStyleBuild: http://127.0.0.1:8080/job/freestylejob/1/>
>>> for line in build.progressive_output():
... print(line)
...
'''
job = self._get_job_and_check(full_name)
return job.build(**params)
def _get_job_and_check(self, full_name):
job = self.get_job(full_name)
if job is None:
raise ItemNotFoundError(f'No such job: {full_name}')
return job
[docs]
def rename_job(self, full_name, new_name):
job = self._get_job_and_check(full_name)
return job.rename(new_name)
[docs]
def move_job(self, full_name, new_full_name):
job = self._get_job_and_check(full_name)
return job.move(new_full_name)
[docs]
def duplicate_job(self, full_name, new_name, recursive=False):
job = self._get_job_and_check(full_name)
return job.duplicate(new_name, recursive)
[docs]
def is_name_safe(self, name):
resp = self.handle_req('GET', 'checkJobName', params={'value': name})
return 'is an unsafe character' not in resp.text
[docs]
def validate_jenkinsfile(self, content):
"""validate Jenkinsfile, see
https://www.jenkins.io/doc/book/pipeline/development/#linter
Args:
content (str): content of Jenkinsfile
Returns:
str: 'Jenkinsfile successfully validated.' if validate successful
or error message
"""
data = {'jenkinsfile': content}
return self.handle_req(
'POST', 'pipeline-model-converter/validate', data=data).text
def _resolve_name(self, full_name):
parent, name = self._parse_name(full_name)
return Folder(self, self._name2url(parent)), name
# def exists(self):
# '''Check if Jenkins server is up
# :returns: True or False
# '''
# try:
# self._request('HEAD', self.url)
# return True
# except Exception as e:
# return isinstance(e, (AuthenticationError, PermissionError))
@property
def crumb(self):
'''Crumb of Jenkins'''
if self._crumb is None:
with self._sync_lock:
if self._crumb is None:
try:
_crumb = self._request(
'GET', f'{self.url}crumbIssuer/api/json').json()
self._crumb = {
_crumb['crumbRequestField']: _crumb['crumb']}
except HTTPStatusError:
self._crumb = {}
return self._crumb
@property
def system(self):
'''An object for managing system operation.
see :class:`System <api4jenkins.system.System>`'''
return System(self, self.url)
@property
def plugins(self):
'''An object for managing plugins.
see :class:`PluginsManager <api4jenkins.plugin.PluginsManager>`'''
return PluginsManager(self, f'{self.url}pluginManager/')
@property
def version(self):
'''Version of Jenkins'''
return self.handle_req('HEAD', '').headers['X-Jenkins']
@property
def credentials(self):
'''An object for managing credentials.
see :class:`Credentials <api4jenkins.credential.Credentials>`'''
return Credentials(self, f'{self.url}credentials/store/system/')
@property
def views(self):
'''An object for managing views of main window.
see :class:`Views <api4jenkins.view.Views>`'''
return Views(self)
@property
def nodes(self):
'''An object for managing nodes.
see :class:`Nodes <api4jenkins.node.Nodes>`'''
return Nodes(self, f'{self.url}computer/')
@property
def queue(self):
'''An object for managing build queue.
see :class:`Queue <api4jenkins.queue.Queue>`'''
return Queue(self, f'{self.url}queue/')
@property
def users(self):
return Users(self, f'{self.url}asynchPeople/')
@property
def me(self):
return self.user
def __call__(self, depth):
yield from self.iter(depth)
def __getitem__(self, full_name):
return self.get_job(full_name)
def _patch_to(module, cls, func=None):
_module = import_module(module)
if func:
_class = getattr(_module, cls.__name__)
setattr(_class, func.__name__, func)
else:
setattr(_module, cls.__name__, cls)
[docs]
class AsyncJenkins(AsyncItem, UrlMixIn):
def __init__(self, url, **kwargs):
self.http_client = new_async_http_client(**kwargs)
self._crumb = None
self._async_lock = asyncio.Lock()
self._auth = kwargs.get('auth')
super().__init__(self, url)
self.user = AsyncUser(
self, f'{self.url}user/{self._auth[0]}/') if self._auth else None
[docs]
async def get_job(self, full_name):
folder, name = self._resolve_name(full_name)
if await folder.exists():
return await folder.get(name)
[docs]
async def aiter(self, depth=0):
async for job in AsyncFolder(self, self.url)(depth):
yield job
[docs]
async def create_job(self, full_name, xml, recursive=False):
folder, name = self._resolve_name(full_name)
if recursive and not await folder.exists():
await self.create_job(folder.full_name, EMPTY_FOLDER_XML,
recursive=recursive)
return await folder.create(name, xml)
[docs]
async def copy_job(self, full_name, dest):
folder, name = self._resolve_name(full_name)
return await folder.copy(name, dest)
[docs]
async def delete_job(self, full_name):
job = await self.get_job(full_name)
if job:
await job.delete()
[docs]
async def build_job(self, full_name, **params):
job = await self._get_job_and_check(full_name)
if not isinstance(job, AsyncProject):
raise AttributeError(f'{job} has no attribute build')
return await job.build(**params)
[docs]
async def rename_job(self, full_name, new_name):
job = await self._get_job_and_check(full_name)
return await job.rename(new_name)
[docs]
async def move_job(self, full_name, new_full_name):
job = await self._get_job_and_check(full_name)
return await job.move(new_full_name)
[docs]
async def duplicate_job(self, full_name, new_name, recursive=False):
job = await self._get_job_and_check(full_name)
return await job.duplicate(new_name, recursive)
async def _get_job_and_check(self, full_name: str):
job = await self.get_job(full_name)
if job is None:
raise ItemNotFoundError(f'No such job: {full_name}')
return job
[docs]
async def is_name_safe(self, name):
resp = await self.handle_req('GET', 'checkJobName', params={'value': name})
return 'is an unsafe character' not in resp.text
[docs]
async def validate_jenkinsfile(self, content):
data = await self.handle_req(
'POST', 'pipeline-model-converter/validate', data={'jenkinsfile': content})
return data.text
def _resolve_name(self, full_name):
parent, name = self._parse_name(full_name)
return AsyncFolder(self, self._name2url(parent)), name
# async def exists(self):
# try:
# await self._request('HEAD', self.url)
# return True
# except Exception as e:
# return isinstance(e, (AuthenticationError, PermissionError))
@property
async def crumb(self):
if self._crumb is None:
async with self._async_lock:
if self._crumb is None:
try:
_crumb = (await self._request('GET', f'{self.url}crumbIssuer/api/json')).json()
self._crumb = {
_crumb['crumbRequestField']: _crumb['crumb']}
except HTTPStatusError:
self._crumb = {}
return self._crumb
@property
def system(self):
return AsyncSystem(self, self.url)
@property
def plugins(self):
return AsyncPluginsManager(self, f'{self.url}pluginManager/')
@property
async def version(self):
return (await self.handle_req('HEAD', '')).headers['X-Jenkins']
@property
def credentials(self):
return AsyncCredentials(self, f'{self.url}credentials/store/system/')
@property
def views(self):
return AsyncViews(self)
@property
def nodes(self):
return AsyncNodes(self, f'{self.url}computer/')
@property
def queue(self):
return AsyncQueue(self, f'{self.url}queue/')
@property
def users(self):
return AsyncUsers(self, f'{self.url}asynchPeople/')
@property
def me(self):
return self.user
async def __call__(self, depth):
async for job in self.aiter(depth):
yield job
async def __getitem__(self, full_name):
return await self.get_job(full_name)