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¶

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, *, 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)¶

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)¶

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.

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

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

Async python client for Jenkins¶

Status¶

Installation¶

Documentation¶

Usage¶

Testing¶

Contributing¶

API endpoints¶

Core¶

Exceptions¶

Jobs¶

Builds¶

Nodes¶

Plugins¶

Views¶

Utils (helpers)¶

Async python client for Jenkins ¶