Async python client for Jenkins¶
Asynchronous python package of Jenkins API endpoints based on aiohttp.
Also pay attention to brand new package with same API set but with sync and async interfaces:
https://github.com/pbelskiy/ujenkins
Installation¶
pip3 install aiojenkins
Usage¶
Start new build:
import asyncio
import aiojenkins
jenkins = aiojenkins.Jenkins('http://your_server/jenkins', 'user', '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()
Documentation¶
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
Contributing¶
Feel free to PR
API endpoints¶
Core¶
-
class
aiojenkins.jenkins.
Jenkins
(host: str, user: Optional[str] = None, password: Optional[str] = None, *, verify: bool = True, timeout: Optional[float] = None, retry: Optional[dict] = None)¶ Core library class.
Parameters: - host (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))
- factor:
- statuses:
List[int]
HTTP statues retries on. (default [])
Example:
retry = dict( total=10, factor=1, statuses=[500] )
- total:
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, build 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, default is 1 second. Returns: None
Exceptions¶
-
exception
aiojenkins.exceptions.
JenkinsError
(message: Optional[str] = None, status: Optional[int] = None)¶
-
exception
aiojenkins.exceptions.
JenkinsNotFoundError
(message: Optional[str] = None, status: Optional[int] = 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
-
static
construct_config
(**kwargs) → str¶ Jenkins job XML constructor, cannot be used for folder creating yet.
-
Builds¶
-
class
aiojenkins.builds.
Builds
(jenkins)¶ List of Jenkins tags which can be used insted of build_id number.
- lastBuild
- lastCompletedBuild
- lastFailedBuild
- lastStableBuild
- lastSuccessfulBuild
- lastUnstableBuild
- lastUnsuccessfulBuild
-
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: Union[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
-
get_all
(name: str) → list¶ Get list of builds for specified job.
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). Returns: list of build for specified job. Return type: List
-
get_info
(name: str, build_id: Union[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: Union[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_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: Union[int, str]) → bool¶ Check if specified build id of job exists.
Parameters: - name (str) – Job name or path (if in folder).
- build_id (int) – Build number or some of standard tags like lastBuild.
Returns: exists or not.
Return type: bool
-
start
(name: str, parameters: Optional[dict] = None, delay: int = 0, **kwargs) → 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 (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]
-
stop
(name: str, build_id: Union[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
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: str = '') → None¶ Disable node if enabled.
Parameters: - name (str) – Node name.
- message (str) – Reason message.
Returns: None
-
enable
(name: str) → None¶ Enable node if disabled.
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
-
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
-
reconfigure
(name: str, config: str) → None¶ Reconfigure node.
Parameters: - name (str) – Node name.
- config (str) – New 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
-
static
construct
(**kwargs) → dict¶ Construct XML node config.
Returns: node XML config. Return type: dict
-
Plugins¶
Queue¶
-
class
aiojenkins.queue.
Queue
(jenkins)¶ -
cancel
(item_id: int) → None¶ Cancel enqueued item (build) identifier.
Parameters: item_id (int) – enqueued item identifier. Returns: None
-
get_all
() → 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
-
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: Optional[str] = None, parameters: Optional[List[dict]] = None, commands: Optional[List[str]] = None) → str¶ Constructs an XML for job creating depends on arguments.
Parameters: - description (Optional[str]) – Job description.
- parameters (Optional[List[dict]]) –
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 (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.
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¶ Construct node config.
Parameters: - name – 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
-
aiojenkins.utils.
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]