Async python client for Jenkins

Asynchronous python library of Jenkins API endpoints based on aiohttp. Initial version of aiojenkins.

Public API is still unstable.

Status

Build Status Docs status Coverage status Version status Downloads status

Installation

pip3 install -U aiojenkins

Documentation

Read the Docs

Usage

Start new build:

import asyncio
import aiojenkins

jenkins = aiojenkins.Jenkins('http://your_server/jenkins', 'login', 'password')

async def example():
    await jenkins.builds.start('job_name', dict(parameter='test'))

loop = asyncio.get_event_loop()
try:
    loop.run_until_complete(example())
finally:
    loop.run_until_complete(jenkins.close())
    loop.close()

Please look at tests directory for more examples.

Testing

Currently tests aren’t using any mocking. I am testing locally with dockerized LTS Jenkins ver. 2.222.3

Prerequisites: docker, tox

docker run -d --name jenkins --restart always -p 8080:8080 jenkins/jenkins:lts
docker exec jenkins cat /var/jenkins_home/secrets/initialAdminPassword
chromium http://localhost:8080  # create admin:admin
tox

Or Jenkins 1.554

docker run -d --name jenkins-1.554 --restart always -p 8081:8080 jenkins:1.554

Contributing

Feel free to PR

API endpoints

Core

class aiojenkins.jenkins.Jenkins(host: str, login: Optional[str] = None, password: Optional[str] = None, *, loop: Optional[asyncio.events.AbstractEventLoop] = None, verify: bool = True, timeout: Optional[float] = None, retry: Optional[dict] = None)

Core library class.

Parameters:
  • host (str) – URL of jenkins server.
  • login (Optional[str]) – Login, user name.
  • password (Optional[str]) – Password for login.
  • loop (Optional[AbstractEventLoop]) – Asyncio current event loop.
  • verify (Optional[bool]) – Verify SSL (default: true).
  • timeout (Optional[int]) – HTTP request timeout.
  • retry (Optional[dict]) –

    Retry options to prevent failures if server restarting or temporary network problem. - total: int Total retries count. (default 0) - factor: int Sleep between retries (default 0)

    {factor} * (2 ** ({number of total retries} - 1))
    • statuses: List[int] HTTP statues retries on. (default [])

    Example: .. code-block:: python

    retry = dict(
    attempts=10, factor=1, statuses=[500]

    )
Returns:

Jenkins instance
cancel_quiet_down() → None

Cancel server quiet down period.

Returns:None
close() → None

Finalize client, close http session.

Returns:None
generate_token(name: str) → Tuple[str, str]

Generate new API token.

Parameters:name (str) – name of token.
Returns:
tokenValue - uses for authorization,
tokenUuid - uses for revoke
Return type:Tuple[str, str]
get_status() → dict

Get server status.

Returns:jenkins server details.
Return type:dict
get_version() → aiojenkins.jenkins.JenkinsVersion

Get server version.

Returns:named tuple with minor, major, patch version.
Return type:JenkinsVersion
is_ready() → bool

Determines is server loaded and ready for work.

Returns:ready state.
Return type:bool
quiet_down() → None

Start server quiet down period, new builds will not be started.

Returns:None
restart() → None

Restart server immediately.

Returns:None
revoke_token(token_uuid: str) → None

Revoke API token, please note that uuid is used, not value.

Parameters:token_uuid (str) – uuid of token to be revoked.
Returns:None
run_groovy_script(script: str) → str

Execute Groovy script on the server.

Parameters:script (str) – script content.
Returns:output of script.
Return type:str
safe_restart() → None

Restart server when installation is complete and no jobs are running.

Returns:None
wait_until_ready(sleep_interval_sec: float = 1.0) → None

Blocks until server is completely loaded.

Parameters:sleep_interval_sec (float) – delay between checks.
Returns:None

Exceptions

exception aiojenkins.exceptions.JenkinsError(message=None, status=None)
exception aiojenkins.exceptions.JenkinsNotFoundError(message=None, status=None)

Jobs

class aiojenkins.jobs.Jobs(jenkins)
copy(name: str, new_name: str) → None

Copy specified job.

Parameters:
  • name (str) – job name or path (within folder).
  • new_name (str) – new job name.
Returns:

None
create(name: str, config: str) → None

Create new jenkins job (project).

Parameters:
  • name (str) – job name.
  • config (str) – XML config of new job. It`s convenient way to use get_config() to get existing job config and change it on your taste, or to use construct_config() method.
Returns:

None
delete(name: str) → None

Delete existed jenkins job (project).

Parameters:name (str) – job name. For job in folder just use /.
Returns:None
disable(name: str) → None

Disable specified job.

Parameters:name (str) – job name.
Returns:None
enable(name: str) → None

Enable specified job.

Parameters:name (str) – job name.
Returns:None
get_all() → Dict[str, dict]

Get dict of all existed jobs in system, including jobs in folder.

Returns: Dict[str, dict] - name and job properties.

Example: {
‘test’: {
‘name’: ‘test’, ‘url’: ‘http://localhost/job/test/’

}, ‘folder/foo’: {

‘name’: ‘folder/job’, ‘url’: ‘http://localhost/job/folder/job/foo/’

}

}

get_config(name: str) → str

Get XML config of specified job.

Parameters:name (str) – job name.
Returns:XML config
Return type:str
get_info(name: str) → dict

Get detailed information of specified job.

Parameters:name (str) – job name.
Returns:job details.
Return type:dict
is_exists(name: str) → bool

Check if specified job exists.

Parameters:name (str) – job name.
Returns:job exists or not
Return type:bool
reconfigure(name: str, config: str) → None

Reconfigure specified job name.

Parameters:
  • name (str) – job name or path (within folder).
  • config (str) – XML config of new job. It`s convenient way to use get_config() to get existing job config and change it on your taste, or to use construct_config() method.
Returns:

None
rename(name: str, new_name: str) → None

Rename specified job name.

Parameters:
  • name (str) – job name or path (within folder).
  • new_name (str) – new job name.
Returns:

None
construct_config(**kwargs) → str

Jenkins job XML constructor, cannot be used for folder creating yet.

Builds

class aiojenkins.builds.Builds(jenkins)
static parse_url(build_url: str) → Tuple[str, int]

Extract job name and build number from url.

Parameters:build_url (str) – build url.
Returns:job name and build id.
Return type:Tuple[str, int]
delete(name: str, build_id: int) → None

Delete specified build.

Parameters:
  • name (str) – job name or path (if in folder).
  • build_id (int) – build identifier.
Returns:

None
get_all(name: str) → list

Get list of builds for specified job.

Parameters:name (str) – job name or path (if in folder).
Returns:list of build for specified job.
builds = [
{‘number’: 1, ‘url’: ‘http://localhost/job/test/1/’}, {‘number’: 2, ‘url’: ‘http://localhost/job/test/2/’}

]
Return type:List
get_info(name: str, build_id: int) → dict

Get detailed information about specified build number of job.

Parameters:
  • name (str) – job name or path (if in folder).
  • build_id (int) – build identifier.
Returns:

information about build.
Return type:

Dict
get_output(name: str, build_id: int) → str

Get console output of specified build.

Parameters:
  • name (str) – job name or path (if in folder).
  • build_id (int) – build identifier.
Returns:

build output.
Return type:

str
get_url_info(build_url: str) → dict

Extract job name and build number from url and return info about build.

Parameters:build_url (str) – job name or path (if in folder).
Returns:information about build.
Return type:Dict
is_exists(name: str, build_id: int) → bool

Check if specified build id of job exists.

Parameters:
  • name (str) – job name or path (if in folder).
  • build_id (int) – build identifier.
Returns:

exists or not.
Return type:

bool
start(name: str, parameters: Optional[dict] = None, delay: Optional[int] = 0) → Optional[int]

Enqueue new build with delay (default is 0 seconds, means immediately)

Note about delay (quiet-period): https://www.jenkins.io/blog/2010/08/11/quiet-period-feature/

Parameters:
  • name (str) – job name or path (if in folder).
  • parameters (int) – parameters of triggering build.
  • delay (int) – delay before start.
Returns:

queue item id.
Return type:

Optional[int]
stop(name: str, build_id: int) → None

Stop specified build.

Parameters:
  • name (str) – job name or path (if in folder).
  • build_id (int) – build identifier.
Returns:

None

Nodes

class aiojenkins.nodes.Nodes(jenkins)
create(name: str, config: dict) → None

Create new node.

Parameters:
  • name (str) – node name.
  • config (str) – XML config for new node.
Returns:

None
delete(name: str) → None

Delete node.

Parameters:name (str) – node name.
Returns:None
disable(name: str, message: Optional[str] = '') → None

Disable node.

Parameters:
  • name (str) – node name.
  • message (Optional[str]) – reason message.
Returns:

None
enable(name: str) → None

Enable node.

Parameters:name (str) – node name.
Returns:None
get_all() → Dict[str, dict]

Get available nodes on server.

Returns:node name, and it`s detailed information.
Example: {
“master”: dict(…), “buildbot1”: dict(…)

}
Return type:Dict[str, dict]
get_all_builds(name: str) → List[dict]

Return list of all detalizied builds for node name, actually it parsed from RSS feed. Ascending builds sort.

Parameters:name (str) – node name.
Returns:list of all builds for specified node.
Example: [{
‘job_name’: ‘test’, ‘number’: 1, ‘url’: ‘http://localhost:8080/job/test/1/’

}]
Return type:List[dict]
get_config(name: str) → str

Return node config in XML format.

Parameters:name (str) – node name.
Returns:node config.
Return type:str
get_failed_builds(name: str) → List[dict]

Return list of detalizied failed builds for node name. Actually it parsed from RSS feed. usefull for build restart. Ascending builds sort.

Parameters:name (str) – node name.
Returns:builds and their information.
Example: [{
‘job_name’: ‘test’, ‘number’: 1, ‘url’: ‘http://localhost:8080/job/test/1/’

}]
Return type:List[dict]
get_info(name: str) → dict

Get node detailed information.

Parameters:name (str) – node name.
Returns:detailed node information.
Return type:dict
is_exists(name: str) → bool

Check is node exist.

Parameters:name (str) – node name.
Returns:node existing.
Return type:bool
reconfigure(name: str, config: str) → None

Reconfigure node.

Parameters:
  • name (str) – node name.
  • config (str) – bew XML config for node.
Returns:

None
update_offline_reason(name: str, message: str) → None

Update reason message of disabled node.

Parameters:
  • name (str) – node name.
  • message (str) – reason message.
Returns:

None
construct(**kwargs) → dict

Construct XML node config.

Returns:node XML config.
Return type:bool

Plugins

class aiojenkins.plugins.Plugins(jenkins)
get_all(depth: Optional[int] = 2) → Dict[str, dict]

Get dict of all existed plugins in the system.

Returns:Dict[str, dict] - plugin name and plugin properties.

Views

class aiojenkins.views.Views(jenkins)
create(name: str, config: str) → None

Create view using XML config.

Parameters:
  • name (str) – view name.
  • config (str) – XML config.
Returns:

None
delete(name: str) → None

Delete view.

Parameters:name (str) – view name.
Returns:None
get_all() → Dict[str, dict]

Get all views and their details.

Returns:Dict[str, dict] - plugin name and plugin properties.
get_config(name: str) → str

Return view config in XML format.

Parameters:name (str) – view name.
Returns:XML config of view.
Return type:str
is_exists(name: str) → bool

Check if view exists.

Parameters:name (str) – view name.
Returns:view existing
Return type:bool
reconfigure(name: str, config: str) → None

Reconfigure view.

Parameters:
  • name (str) – view name.
  • config (str) – XML config.
Returns:

None

Utils (helpers)

aiojenkins.utils.construct_job_config(*, description: str = None, parameters: List[dict] = None, commands: List[str] = None) → str

Constructs an XML for job creating depends on arguments.

Parameters:
  • description (-) – Job description
  • parameters (-) –

    Parameters for job, note that name is mandatory field.

    Example: [
    dict(name=’param1’), dict(name=’param2’, description=’helpfull information’), dict(name=’param3’, default=’default command value’),

    ]

  • commands (-) –
    List of commands which will be joined as one string by note
    that entire command block will run in one shell instance.
    Example: [
    ‘echo 1’, ‘sleep 5’,

    ]
Returns:

Prettified XML ready to submit on Jenkins
Return type:

  • str
aiojenkins.utils.construct_node_config(*, name: str, remote_fs: str = '/tmp', executors: int = 2) → dict
Parameters:
  • name (-) – Node name
  • remote_fs (-) – Remote node root directory
  • executors (-) – Number of node executors
Returns:

return ready to use dict with nodes.create()
Return type:

  • dict
aiojenkins.utils.parse_build_url(build_url: str) → Tuple[str, int]

Extract job name and build number from build url