Universal Python client for Jenkins ¶

Coverage status Version status Downloads status

Python client for Jenkins which supports both sync and async syntax with same interface.

Comparison to other packages
Name	Sync	Async	Python
ujenkins	YES	YES	3.6+
aiojenkins	NO	YES	3.5+
python-jenkins	YES	NO	3.4+
jenkinsapi	YES	NO	3.4+

Installation¶

Latest release from PyPI

pip3 install ujenkins

Or latest developing version

pip3 install git+https://github.com/pbelskiy/ujenkins

Usage¶

Get Jenkins version using sync client:

from ujenkins import JenkinsClient

def example():
    client = JenkinsClient('http://server', 'user', 'password')
    version = client.system.get_version()
    print(version)

example()

With async client (be careful AsyncJenkinsClient must be called inside async function):

import asyncio
from ujenkins import AsyncJenkinsClient

async def example():
    client = AsyncJenkinsClient('http://server', 'user', 'password')
    version = await client.system.get_version()
    print(version)
    await client.close()

asyncio.run(example())

Examples¶

In all code examples below client instance is created by:

from ujenkins import JenkinsClient
client = JenkinsClient('http://server', 'user', 'password')

Get timestamp of latest build¶

client.builds.get_info('job', 'lastBuild')['timestamp']

Get url of started build¶

Be careful, JenkinsNotFoundError could be raise in case build with same arg already enqueued.

item_id = client.builds.start('my_job')
while True:
    time.sleep(5)
    try:
        info = client.queue.get_info(item_id)
        print(info['executable']['url'])
        break
    except (KeyError, TypeError):
        pass  # wait for build will be started

Get all jobs¶

Basically client.jobs.get() returns jobs from root (depth = 0), in case you want receive all the jobs, there are few approaches for it.

Set needed depth, experimentally 10 is enough.

jobs = client.jobs.get(depth=10)

Output:

{'folder': {'_class': 'com.cloudbees.hudson.plugins.folder.Folder',
            'jobs': [{'_class': 'hudson.model.FreeStyleProject',
                    'color': 'notbuilt',
                    'name': 'job_in_folder1',
                    'url': 'http://localhost:8080/job/folder/job/job_in_folder1/'},
                    {'_class': 'com.cloudbees.hudson.plugins.folder.Folder',
                    'jobs': [{'_class': 'hudson.model.FreeStyleProject',
                                'color': 'notbuilt',
                                'name': 'sub_job_in_subfolder',
                                'url': 'http://localhost:8080/job/folder/job/subfolder/job/sub_job_in_subfolder/'}],
                    'name': 'subfolder',
                    'url': 'http://localhost:8080/job/folder/job/subfolder/'}],
            'name': 'folder',
            'url': 'http://localhost:8080/job/folder/'},
'job': {'_class': 'hudson.model.FreeStyleProject',
        'color': 'blue',
        'name': 'job',
        'url': 'http://localhost:8080/job/job/'}}

Or just write your code to recursively form it, example is below.

def get_all_jobs(url: str = '', parent: str = '') -> Dict[str, dict]:
    jobs = {}

    for name, prop in client.jobs.get(url).items():
        jobs[parent + name] = prop
        if 'Folder' in prop.get('_class', ''):
            jobs.update(get_all_jobs(prop['url'], parent + name + '/'))

    return jobs

all_jobs = get_all_jobs()

Working with build artifacts¶

# get content of artifact (bytes)
content = client.builds.get_artifact('my_job', 31, 'photo.jpg')
with open('/tmp/photo.jpg', 'wb') as f:
    w.write(content)

# enumerate artifacts
artifacts = client.builds.get_list_artifacts('my_job', 31)
for artifact in artifacts:
    # get content and manually save it
    content = client.builds.get_artifact('my_job', 31, artifact['path'])

    # or absolute url could be used for external download
    print(artifact['url'])
    # >> 'http://server/job/my_job/31/artifact/photo.jpg'

Documentation¶

Read the Docs

Testing¶

Prerequisites: tox

Then just run tox, all dependencies and checks will run automatically

tox

Contributing¶

Any contributions are welcome!

API endpoints¶

Core¶

class ujenkins.JenkinsClient(url: str, user: str | None = None, password: str | None = None, *, verify: bool = True, timeout: float | None = None, retry: dict | None = None)¶

Jenkins sync client class.

Parameters:

url (str) – URL of Jenkins server.
user (Optional[str]) – User name.
password (Optional[str]) – Password for user.
verify (Optional[bool]) – Verify SSL (default: true).
timeout (Optional[float]) – HTTP request timeout.

retry (Optional[dict]) –

Retry options to prevent failures if server restarting or temporary network problem. Disabled by default use total > 0 to enable.

total: int Total retries count.
factor: int Sleep between retries (default 1)
{factor} * (2 ** ({number of total retries} - 1))
statuses: List[int] HTTP statues retries on. (default [])
methods: List[str] list of HTTP methods to retry, idempotent
methods are used by default.

Example:

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

With factor = 1

Retry number	Sleep
1	0.5 seconds
2	1.0 seconds
3	2.0 seconds
4	4.0 seconds
5	8.0 seconds
6	16.0 seconds
7	32.0 seconds
8	1.1 minutes
9	2.1 minutes
10	4.3 minutes
11	8.5 minutes
12	17.1 minutes
13	34.1 minutes
14	1.1 hours
15	2.3 hours
16	4.6 hours
17	9.1 hours
18	18.2 hours
19	36.4 hours
20	72.8 hours

Returns:

Client instance

close() → None¶: Close client session

class ujenkins.AsyncJenkinsClient(url: str, user: str | None = None, password: str | None = None, *, verify: bool = True, timeout: float | None = None, retry: dict | None = None)¶

Jenkins async client class.

Parameters:

url (str) – URL of Jenkins server.
user (Optional[str]) – User name.
password (Optional[str]) – Password for user.
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. Disabled by default use total > 0 to enable.
- total: int Total retries count.
- factor: int Sleep between retries (default 1)
  {factor} * (2 ** ({number of total retries} - 1))
- statuses: List[int] HTTP statues retries on. (default [])
- methods: List[str] list of HTTP methods to retry, idempotent
  methods are used by default.
Example:
```
retry = dict(
    total=10,
    factor=1,
    statuses=[500]
)
```

Returns:

AsyncClient instance

async close() → None¶: Close client session

class ujenkins.core.Response(status, headers, text, content)¶

Create new instance of Response(status, headers, text, content)

status: int¶: Alias for field number 0

headers: CaseInsensitiveDict | CIMultiDictProxy¶: Alias for field number 1

text: str¶: Alias for field number 2

content: bytes | None¶: Alias for field number 3

Builds¶

class ujenkins.endpoints.builds.Builds(jenkins)¶

List of Jenkins tags which can be used insted of build_id number.

lastBuild
lastCompletedBuild
lastFailedBuild
lastStableBuild
lastSuccessfulBuild
lastUnstableBuild
lastUnsuccessfulBuild

get(name: str, *, fields: List[str] | None = None, start: int | None = None, end: int | None = None) → List[dict]¶

Get list of builds for specified job.

Available fields in jenkins:

_class
actions
artifacts
building
description
displayName
duration
estimatedDuration
executor
fingerprint
fullDisplayName
id
keepLog
number
queueId
result
timestamp
url
changeSets
culprits
inProgress
nextBuild
previousBuild

Use * inside the list to get all fields. Example:

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

Parameters:

name (str) – Job name or path (if in folder).
fields (Optional[List]) – List of fields to return. Possible values are mentioned in the available fields. If empty, default fields will be returned.
start (Optional[int]) – The start index of the builds to retrieve. Used with the ‘end’ parameter to specify a range. Defaults to None.
end (Optional[int]) – The end index of the builds to retrieve. Used with the ‘start’ parameter to specify a range. Defaults to None.

Returns:

list of builds for specified job.

Return type:

List[dict]

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

Get detailed information about specified build number of job.

Parameters:

name (str) – Job name or path (if in folder).
build_id (int) – Build number or some of standard tags like lastBuild.

Returns:

information about build.

Return type:

dict

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

Get console output of specified build.

Parameters:

name (str) – Job name or path (if in folder).
build_id (int) – Build number or some of standard tags like lastBuild.

Returns:

build output.

Return type:

str

get_artifact(name: str, build_id: int | str, path: str) → bytes¶

Get artifact content of specified build.

Parameters:

name (str) – Job name or path (if in folder).
build_id (int) – Build number or some of standard tags like lastBuild.
path (str) – Relative path to build artifact, could be used from get_info() or get_list_artifacts(), or just well known name.

Returns:

artifact content.

Return type:

bytes

get_list_artifacts(name: str, build_id: int | str) → List[dict]¶

Get list of build artifacts.

Example

[
    {
        'name': 'photo.jpg',
        'path': 'photo.jpg',
        'url': 'http://server/job/my_job/31/artifact/photo.jpg'
    }
]

Parameters:

name (str) – Job name or path (if in folder).
build_id (int) – Build number or some of standard tags like lastBuild.

Returns:

list of build artifacts.

Return type:

List[dict]

start(name: str, parameters: Any | None = None, delay: int = 0, **kwargs: Any) → int | None¶

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 (Optional[Any]) –

Parameters of triggering build as dict or argument, also parameters can be passed as kwargs.

Examples:

start(..., parameters=dict(a=1, b='string'))
start(..., a=1, b='string')
start(..., parameters=1)
start(..., parameters(a=1, b='string'), c=3)

delay (int) – Delay before start, default is 0, no delay.

Returns:

queue item id.

Return type:

Optional[int]

Raises:

JenkinsNotFoundError – in case build with same arg already enqueued.

stop(name: str, build_id: int | str) → None¶

Stop specified build.

Parameters:

name (str) – Job name or path (if in folder).
build_id (int) – Build number or some of standard tags like lastBuild.

Returns:

None

delete(name: str, build_id: int | str) → None¶

Delete specified build.

Parameters:

name (str) – Job name or path (if in folder).
build_id (int) – Build number or some of standard tags like lastBuild.

Returns:

None

Jobs¶

class ujenkins.endpoints.jobs.Jobs(jenkins)¶

get(url: str = '', depth: int | None = None) → Dict[str, dict]¶

Get jobs in selected folder, by default root is used.

Parameters:

url (str) – URL from job property, by default root is used.
depth (Optional[int]) – Get jobs recursively including folders from selected URL using depth value, default is None which means no recursion.

Example:

{'folder': {'_class': 'com.cloudbees.hudson.plugins.folder.Folder',
            'name': 'folder',
            'url': 'http://localhost/job/folder/'},
 'project': {'_class': 'hudson.model.FreeStyleProject',
             'color': 'blue',
             'name': 'project',
             'url': 'http://localhost/job/project/'}}

Returns:: full name and job properties.
Return type:: Dict[str, dict]

get_info(name: str) → dict¶

Get detailed information of specified job.

Parameters:: name (str) – Job name.
Returns:: job information.
Return type:: dict

get_config(name: str) → str¶

Get XML config of a specified job.

Parameters:: name (str) – Job name.
Returns:: XML config
Return type:: str

is_exists(name: str) → bool¶

Checks if the job exists.

Parameters:: name (str) – Job name.
Returns:: job exists.
Return type:: bool

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

Create new jenkins job.

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

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

delete(name: str) → None¶

Delete existed jenkins job.

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

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

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

enable(name: str) → None¶

Enable specified job.

Parameters:: name (str) – Job name.
Returns:: None

disable(name: str) → None¶

Disable specified job.

Parameters:: name (str) – Job name.
Returns:: None

Nodes¶

class ujenkins.endpoints.nodes.Nodes(jenkins)¶

get() → Dict[str, dict]¶

Get all available nodes on server.

Returns:: node name, and it`s detailed information.
Return type:: Dict[str, dict]

Example:

response = {
    "master": dict(...),
    "buildbot1": dict(...)
}

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.
Return type:: List[dict]

Example:

response = [{
    'job_name': 'test',
    'number': 1,
    'url': 'http://localhost:8080/job/test/1/'
}]

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.
Return type:: List[dict]

Example:

response = [{
    'job_name': 'test',
    'number': 1,
    'url': 'http://localhost:8080/job/test/1/'
}]

get_info(name: str) → dict¶

Get node detailed information.

Parameters:: name (str) – Node name.
Returns:: detailed node information.
Return type:: dict

get_config(name: str) → str¶

Return node config in XML format.

Parameters:: name (str) – Node name.
Returns:: node config.
Return type:: str

is_exists(name: str) → bool¶

Check is node exist.

Parameters:: name (str) – Node name.
Returns:: node existing.
Return type:: bool

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

Create new node.

Parameters:

name (str) – Node name.
config (dict) – Config for new node, see ujenkins.helpers.construct_node_config

Returns:

None

Raises:

JenkinsError – in case node already exists.

delete(name: str) → None¶

Delete node.

Parameters:: name (str) – Node name.
Returns:: None

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

Reconfigure node.

Parameters:

name (str) – Node name.
config (str) – New XML config for node.

Returns:

None

enable(name: str) → None¶

Enable node if it disabled.

Parameters:: name (str) – Node name.
Returns:: None

disable(name: str, message: str = '') → None¶

Disable node if it enabled.

Parameters:

name (str) – Node name.
message (Optional[str]) – Reason message.

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

launch_agent(name: str) → None¶

Launch agent on node, for example in case when disconnected.

State of connection can be determinated by get_info(…) method, which contains custom property defined by packages: _disconnected.

Parameters:: name (str) – Node name.
Returns:: None

Plugins¶

class ujenkins.endpoints.plugins.Plugins(jenkins)¶

get(depth: int = 2) → Dict[str, dict]¶

Get dict of all existed plugins in the system.

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

Queue¶

class ujenkins.endpoints.queue.Queue(jenkins)¶

get() → Dict[int, dict]¶

Get server queue.

Returns:: id item in queue, and it’s detailed information.
Return type:: Dict[int, dict]

get_info(item_id: int) → dict¶

Get info about enqueued item (build) identifier.

Parameters:: item_id (int) – enqueued item identifier.
Returns:: identifier information.
Return type:: dict

cancel(item_id: int) → None¶

Cancel enqueued item (build) identifier.

Parameters:: item_id (int) – enqueued item identifier.
Returns:: None

System¶

class ujenkins.endpoints.system.System(jenkins)¶

get_status() → dict¶

Get server status.

Returns:: jenkins server details.
Return type:: dict

get_version() → 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

cancel_quiet_down() → None¶

Cancel server quiet down period.

Returns:: None

restart() → None¶

Restart server immediately.

Returns:: None

safe_restart() → None¶

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

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]

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

Views¶

class ujenkins.endpoints.views.Views(jenkins)¶

get() → Dict[str, dict]¶

Get views and their details.

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

is_exists(name: str) → bool¶

Check if view exists.

Parameters:: name (str) – View name.
Returns:: view existing.
Return type:: bool

get_config(name: str) → str¶

Return view config in XML format.

Parameters:: name (str) – View name.
Returns:: XML config of view.
Return type:: str

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

Create view using XML config.

Parameters:

name (str) – View name.
config (str) – XML config.

Returns:

None

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

Reconfigure 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

Helpers¶

ujenkins.helpers.construct_job_config(*, description: str | None = None, parameters: List[dict] | None = None, commands: List[str] | None = None) → str¶

Constructs an XML for job creating depends on arguments.

Example:

parameters = [
    dict(name='param1'),
    dict(name='param2', description='helpfull information'),
    dict(name='param3', default='default command value'),
]

commands = [
    'echo 1',
    'sleep 5',
]

Parameters:

description (Optional[str]) – Job description.
parameters (Optional[List[dict]]) – Parameters for job, note that name is mandatory field.
commands (Optional[List[str]]) – List of commands which will be joined as one string by note that entire command block will run in one shell instance.

Returns:

Prettified XML ready to submit on Jenkins.

Return type:

str

ujenkins.helpers.construct_node_config(*, name: str, remote_fs: str = '/tmp', executors: int = 2) → dict¶

Construct node config.

Parameters:

name (str) – Node name.
remote_fs (str) – Remote node root directory.
executors (int) – Number of node executors

Returns:

return ready to use dict with nodes.create()

Return type:

dict

ujenkins.helpers.parse_build_url(build_url: str) → Tuple[str, int]¶

Extract job name and build number from build url.

Parameters:: build_url (str) – URL to build.
Returns:: job name and build number.
Return type:: Tuple[str, int]

Exceptions¶

exception ujenkins.exceptions.JenkinsError(message=None, status=None)¶: Core library exception

exception ujenkins.exceptions.JenkinsNotFoundError(message=None, status=None)¶: Raises when request return HTTP code 404 (not found)

Universal Python client for Jenkins ¶

Installation¶

Usage¶

Examples¶

Get timestamp of latest build¶

Get url of started build¶

Get all jobs¶

Working with build artifacts¶

Documentation¶

Testing¶

Contributing¶

API endpoints¶

Core¶

Builds¶

Jobs¶

Nodes¶

Plugins¶

Queue¶

System¶

Views¶

Helpers¶

Exceptions¶

Home page

Navigation

Universal Python client for Jenkins¶

Installation¶

Usage¶

Examples¶

Get timestamp of latest build¶

Get url of started build¶

Get all jobs¶

Working with build artifacts¶

Documentation¶

Testing¶

Contributing¶

API endpoints¶

Core¶

Builds¶

Jobs¶

Nodes¶

Plugins¶

Queue¶

System¶

Views¶

Helpers¶

Exceptions¶

Universal Python client for Jenkins ¶