pbelskiy / ujenkins

Python client for Jenkins (sync + async)

Universal Python client for Jenkins

---

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.

1. 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/'}}

2. 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!

ujenkins's Issues

Jenkins version example gives error "TypeError: string indices must be integers"

When running the sync Jenkins version example (https://github.com/pbelskiy/ujenkins#usage) I get the following errror. What am i doing wrong?

Traceback (most recent call last):
  File "/root/test.py", line 8, in <module>
    example()
  File "/root/test.py", line 5, in example
    version = client.system.get_version()
  File "/usr/local/lib/python3.9/dist-packages/ujenkins/endpoints/system.py", line 40, in get_version
    return self.jenkins._request('GET', '/', callback=callback)
  File "/usr/local/lib/python3.9/dist-packages/ujenkins/adapters/sync.py", line 173, in _request
    self.crumb = self._get_crumb()
  File "/usr/local/lib/python3.9/dist-packages/ujenkins/adapters/sync.py", line 152, in _get_crumb
    self.crumb = {response['crumbRequestField']: response['crumb']}
TypeError: string indices must be integers

Better types definitions - pyright

Can ujenkins better support return types(mostly for async staff)? When I use AsyncJenkinsClient and call for example hw_jenkins.builds.get_artifact() the return type is Bytes so when I await for that then pyright will return error: "bytes" is not awaitable (reportGeneralTypeIssues). To resolve these issues it is required to do casting on each function call cast(Awaitable[bytes], hw_jenkins.builds.get_artifact()).

New release

Hello, Petr!
Are there any plans to build a new release with build.get() new args support?
Because I really need it :)
I would be very grateful for that.

add nodes.launch_agent()

POST /jenkins/computer/xxx/launchSlaveAgent

Jobs get method not retrieving all jobs

The get method only seems to retrieve top layer jobs and does not seem to recurse into the folders.

Bug with retries

JenkinsClient class can have retry parameter. But passing it causes errors. I suspect it may be due to urllib3 Retry class renaming method_whitelist to allowed_methods. Renaming it locally seems to fix the problem.

ujenkins==0.10.0
urllib3==2.0.3
python3.10.7

API Documentation Generation Issue

There is no information listed under the "API endpoints" section on Read the Docs.

When I ran a Sphinx build in my local environment, the descriptions were generated as intended.

Could it be that the build on Read the Docs is failing?
Thank you for checking into this matter.

add proxy parameter to client init

add proxy parameter to client init

method_whitelist is deprecated (urllib3)

https://urllib3.readthedocs.io/en/stable/reference/urllib3.util.html?highlight=method_whitelist

method_whitelist is deprecated

Readme decode error when trying to install from source on Windows

Hi,

like I already mentioned here, when trying to install the package from source on Windows, it gives the following error.

UnicodeDecodeError: 'charmap' codec can't decode byte 0x81 in position 738: character maps to <undefined>

I had to add the encoding in setup.py in line 18 to fix this.

...
with open('README.rst', encoding='utf-8') as readme_file:
...

remove Python 3.5 support

And use f-strings

rename login to user

Auth problem on Jenkins 2.361.2

When using user\password pair instead of recommended token, something strange happens:

If http://localhost:8080 is using it ok
if http://127.0.0.0:8080 leads 403

Feature: getting job artifacts

Is it possible somehow download job artifacts besides switching to aiohttp and doing downloads outside of ujenkins module?
If not would it be possible to create new function to download these artifacts, similar to getting console output?

fix GitHub actions

It duplicated for PR and push

docs: fix width of left panel

Or need to make paths shorter like get_info() instead of Build.get_info()

add table with different packages comparison

New Feature: Console Output via logText

Some of my console log output is 14mb per job. I want to be able to incrementally load the console output live.
My particular use-case, I am writing the stream to redis via a celery eventlet.

I think the feature would also be useful for running groovy scripts written locally via the api, similar to the VSCode plugin https://marketplace.visualstudio.com/items?itemName=tabeyti.jenkins-jack which I find very useful.

I was successful in replicating the same api calls one would see in the console output.

I had to create a new endpoint get_output_logtext to set a different content type and make it a POST, and I had to create a new callback that exposed the response headers as they are used to determine if there is more data coming, and the next index to ask for.

In the example below I use the branch to do a loop and create an async generator. I don't know how much you want to include so I tried to keep it as simple as possible.

If you are interested in this feature could you please guide me on how I should be refactoring it so that I could make the PR?

39c0f0e

example:

from ujenkins import AsyncJenkinsClient #this is GeneMyslinsky/ujenkins/streaming
from tm_auth.settings import settings
import nest_asyncio, asyncio
nest_asyncio.apply()

async def start_job(job_location, parameters=None):
    jenkins = AsyncJenkinsClient(settings['JENKINS_URL'], settings['JENKINS_USER'], settings['JENKINS_PASS'])
    args = {}
    if parameters: args = parameters

    queue = await jenkins.builds.start(job_location, parameters=args)
    qinfo = await jenkins.queue.get_info(queue)


    if qinfo['executable'] and qinfo['executable']['number']:
        number, url = qinfo['executable']['number'], qinfo['executable']['url']
    else:
        raise Exception("Job did not start")

    tries = 0
    while (tries := tries + 1) < 10:
        job = await jenkins.builds.get_info(job_location, number)
        if job['building']:
            break
        else:
            await asyncio.sleep(1)
    
    async for output in stream_job(job_location, number):
        print(output)
        
async def stream_job(path, number, poll_text=5):

    start = 0
    while True:
        output, headers = await jenkins.builds.get_output_logtext(path, number, start)
        yield output
        if headers.get('X-More-Data') == 'true':
            start = headers['X-Text-Size']
        else:
            break

        await asyncio.sleep(poll_text)