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 Coverage status Version status Downloads status

Installation

pip install 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, pytest pytest-cov pytest-asyncio

docker run -d --name jenkins --restart always -p 8080:8080 jenkins/jenkins:lts
docker exec jenkins cat /var/jenkins_home/secrets/initialAdminPassword
python3 -m pytest -v --cov=aiojenkins --cov-report=term --cov-report=html

Contributing

Feel free to PR

API endpoints

Core

class aiojenkins.jenkins.Jenkins(host: str, login: Optional[str] = None, password: Optional[str] = None, *, retry: Optional[dict] = None)

Core library class.

It`s possible to use retry argument to prevent failures if server restarting or temporary network problems, anyway 500 ~ 599 HTTP statues is checked and triggers new retry attempt. To enable retry with default options just pass retry=dict(enabled=True)

Example: retry = dict(
attempts=50, # total attempts count default is 50 interval=0.2, # interval in seconds between attempts statuses=[403], # additional HTTP statuses for retry

)

cancel_quiet_down() → None

Cancel server quiet down period

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

Generate new API token.

Returns two values: * tokenValue - uses for authorization * tokenUuid - uses for revoke

is_ready() → bool

Determines is server loaded and ready for work

quiet_down() → None

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

restart() → None

Restart server immediately

revoke_token(token_uuid: str) → None

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

run_groovy_script(script: str) → str

Execute Groovy script on the server.

safe_restart() → None

Restart server when installation is complete and no jobs are running

wait_until_ready(sleep_interval_sec: float = 1.0) → None

Blocks until server is completely loaded

Exceptions

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

Jobs

class aiojenkins.jobs.Jobs(jenkins)
construct_config(**kwargs) → str

Jenkins job XML constructor

get_all() → dict

Get list of builds for specified job name. Returned example:

jobs = [
{‘name’: ‘test’, ‘url’: ‘http://localhost/job/test/’}, {‘name’: ‘test_builds’, ‘url’: ‘http://localhost/job/test_builds/’}

]

Builds

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

Extract job name and build number from build url

get_all(name: str) → list

Get list of builds for specified job name. Returned example:

builds = [
{‘number’: 1, ‘url’: ‘http://localhost/job/test/1/’}, {‘number’: 2, ‘url’: ‘http://localhost/job/test/2/’}

]

get_info(name: str, build_id: int) → dict

Get detailed information about specified build of job

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

Get console output of specified build

get_url_info(build_url: str) → dict

Extract job name and build number from url and return info

start(name: str, parameters: Optional[dict] = None, delay: 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/

Nodes

class aiojenkins.nodes.Nodes(jenkins)
get_all() → dict

Get available nodes on server. Example dict result: {

“master”: dict(…), “buildbot1”: 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.

Example: [{
‘job_name’: ‘test’, ‘number’: 1, ‘url’: ‘http://localhost:8080/job/test/1/

}]

get_config(name: str) → str

Return node config in XML format

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.

Example: [{
‘job_name’: ‘test’, ‘number’: 1, ‘url’: ‘http://localhost:8080/job/test/1/

}]

construct(**kwargs)

Jenkins node constructor, returns dict to be passed to create method.

Views

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

Create view

delete(name: str) → None

Delete view

get_all() → dict

Returns dict of all views and their properties.

get_config(name: str) → str

Return view config in XML format

is_exists(name: str) → bool

Check if view exists

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

Reconfigure view

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