Jenkins lockable-resources CLI manager.

.bash_completion

_lockres_completions()
{
    local actions
    actions="list lock unlock"
    case $COMP_CWORD in
        1)
            COMPREPLY=( $(compgen -W "${actions}" -- "${COMP_WORDS[COMP_CWORD]}") )
            ;;
        2)
            names=$(lockres list-names)
            COMPREPLY=( $(compgen -W "${names}" -- "${COMP_WORDS[COMP_CWORD]}") )
            ;;
    esac
    return 0
}

complete -F _lockres_completions lockres

manage-resources.py

from __future__ import annotations

import argparse
import os
import ast
import sys
from pathlib import Path
from datetime import datetime
from time import time
from typing import Any

import dotenv
import httpx
from prettytable import PrettyTable

LOCKABLE_RESOURCES_URL = "https://jenkins.n7space.com/lockable-resources/"
RESOURCE_CACHE_FILE = Path(__file__).parent / ".resource-names"
RESOURCE_CACHE_LIFETIME = 60 * 60  # 1 hour, in seconds


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Lockable Resources Manager for Jenkins."
    )

    parser.add_argument(
        "action",
        choices=["list", "list-names", "lock", "unlock"],
        help="Action to perform on lockable resources.",
    )

    parser.add_argument(
        "resource_name",
        nargs="?",
        help="Name of the resource to lock or unlock. Required for 'lock' and 'unlock' actions.",
    )

    args = parser.parse_args()

    if args.action in ["lock", "unlock"] and not args.resource_name:
        parser.error(
            "The 'resource_name' argument is required when performing a 'lock' or 'unlock' action."
        )

    return args


def get_lockable_resources(user: str, api_key: str) -> list[dict[str, Any]]:
    response = httpx.get(
        f"{LOCKABLE_RESOURCES_URL}/api/python", auth=httpx.BasicAuth(user, api_key)
    )

    if response.status_code != httpx.codes.OK:
        print(
            f"Error: could not retrieve resources (response status code: {response.status_code})"
        )
        sys.exit(1)

    resources = ast.literal_eval(response.text)
    return resources["resources"]


def request_resource_lock(user: str, api_key: str, resource_name: str) -> None:
    response = httpx.post(
        f"{LOCKABLE_RESOURCES_URL}/reserve?resource={resource_name}",
        auth=httpx.BasicAuth(user, api_key),
    )

    if response.status_code == httpx.codes.LOCKED:
        print("Error: could not lock resource (it's already in use!)")
        sys.exit(3)
    elif response.status_code != httpx.codes.FOUND:
        print(
            f"Error: could not lock resource (response status code: {response.status_code})"
        )
        sys.exit(4)


def request_resource_unlock(user: str, api_key: str, resource_name: str) -> None:
    response = httpx.post(
        f"{LOCKABLE_RESOURCES_URL}/unreserve?resource={resource_name}",
        auth=httpx.BasicAuth(user, api_key),
    )

    if (
        response.status_code != httpx.codes.FORBIDDEN
        and response.status_code != httpx.codes.FOUND
    ):
        print(
            f"Error: could not unlock resource. Response code: {response.status_code}"
        )
        print(response.text)
        sys.exit(5)


def list_resources(user: str, api_key: str):
    table = PrettyTable(["Name", "Locked by", "Locked at"])
    for resource in get_lockable_resources(user, api_key):
        if resource["free"]:
            table.add_row([resource["name"], "", ""])
        else:
            lock_time = datetime.fromtimestamp(
                resource["reservedTimestamp"] / 1000
            )  # jenkins uses milliseconds
            if resource["reserved"]:
                table.add_row(
                    [
                        resource["name"],
                        resource["reservedBy"],
                        lock_time.strftime("%Y-%m-%d %H:%M"),
                    ]
                )
            else:
                table.add_row(
                    [
                        resource["name"],
                        resource["buildName"],
                        lock_time.strftime("%Y-%m-%d %H:%M"),
                    ]
                )

    print(table)


def resource_cache_needs_update() -> bool:
    if not RESOURCE_CACHE_FILE.exists():
        return True

    last_modified = os.path.getmtime(RESOURCE_CACHE_FILE)
    return (time() - last_modified) > RESOURCE_CACHE_LIFETIME


def get_resource_names(user: str, api_key: str) -> str:
    resources_list = []
    for resource in get_lockable_resources(user, api_key):
        if not resource["ephemeral"]:
            resources_list.append(resource["name"])

    return " ".join(resources_list)


def list_resource_names(user: str, api_key: str):
    if resource_cache_needs_update():
        RESOURCE_CACHE_FILE.write_text(get_resource_names(user, api_key))
    print(RESOURCE_CACHE_FILE.read_text())


def lock_resource(user: str, api_key: str, resource_name: str):
    request_resource_lock(user, api_key, resource_name)
    print(f"Resource {resource_name} locked!")


def unlock_resource(user: str, api_key: str, resource_name: str):
    request_resource_unlock(user, api_key, resource_name)
    print(f"Resource {resource_name} unlocked!")


def main():
    args = parse_args()

    dotenv.load_dotenv()
    jenkins_api_key = os.getenv("JENKINS_API_KEY")
    jenkins_user = os.getenv("JENKINS_USER")

    if jenkins_api_key is None or jenkins_user is None:
        print("""Missing JENKINS_API_KEY/JENKINS_USER env var!
    Generate a new API key via Jenkins dashboard, and save it with your username
    as JENKINS_API_KEY and JENKINS_USER in `.env` file stored next to this script.""")
        sys.exit(1)

    if args.action == "list":
        list_resources(jenkins_user, jenkins_api_key)
    elif args.action == "list-names":
        list_resource_names(jenkins_user, jenkins_api_key)
    elif args.action == "lock":
        lock_resource(jenkins_user, jenkins_api_key, args.resource_name)
    elif args.action == "unlock":
        unlock_resource(jenkins_user, jenkins_api_key, args.resource_name)


if __name__ == "__main__":
    main()

requirements.txt

# This file was autogenerated by uv via the following command:
#    uv export
anyio==4.9.0 \
    --hash=sha256:673c0c244e15788651a4ff38710fea9675823028a6f08a5eda409e0c9840a028 \
    --hash=sha256:9f76d541cad6e36af7beb62e978876f3b41e3e04f2c1fbf0884604c0a9c4d93c
    # via httpx
certifi==2025.1.31 \
    --hash=sha256:3d5da6925056f6f18f119200434a4780a94263f10d1c21d032a6f6b2baa20651 \
    --hash=sha256:ca78db4565a652026a4db2bcdf68f2fb589ea80d0be70e03929ed730746b84fe
    # via
    #   httpcore
    #   httpx
dotenv==0.9.9 \
    --hash=sha256:29cf74a087b31dafdb5a446b6d7e11cbce8ed2741540e2339c69fbef92c94ce9
    # via jenkins
h11==0.14.0 \
    --hash=sha256:8f19fbbe99e72420ff35c00b27a34cb9937e902a8b810e2c88300c6f0a3b699d \
    --hash=sha256:e3fe4ac4b851c468cc8363d500db52c2ead036020723024a109d37346efaa761
    # via httpcore
httpcore==1.0.8 \
    --hash=sha256:5254cf149bcb5f75e9d1b2b9f729ea4a4b883d1ad7379fc632b727cec23674be \
    --hash=sha256:86e94505ed24ea06514883fd44d2bc02d90e77e7979c8eb71b90f41d364a1bad
    # via httpx
httpx==0.28.1 \
    --hash=sha256:75e98c5f16b0f35b567856f597f06ff2270a374470a5c2392242528e3e3e42fc \
    --hash=sha256:d909fcccc110f8c7faf814ca82a9a4d816bc5a6dbfea25d6591d6985b8ba59ad
    # via jenkins
idna==3.10 \
    --hash=sha256:12f65c9b470abda6dc35cf8e63cc574b1c52b11df2c86030af0ac09b01b13ea9 \
    --hash=sha256:946d195a0d259cbba61165e88e65941f16e9b36ea6ddb97f00452bae8b1287d3
    # via
    #   anyio
    #   httpx
prettytable==3.16.0 \
    --hash=sha256:3c64b31719d961bf69c9a7e03d0c1e477320906a98da63952bc6698d6164ff57 \
    --hash=sha256:b5eccfabb82222f5aa46b798ff02a8452cf530a352c31bddfa29be41242863aa
    # via jenkins
python-dotenv==1.1.0 \
    --hash=sha256:41f90bc6f5f177fb41f53e87666db362025010eb28f60a01c9143bfa33a2b2d5 \
    --hash=sha256:d7c01d9e2293916c18baf562d95698754b0dbbb5e74d457c45d4f6561fb9d55d
    # via dotenv
sniffio==1.3.1 \
    --hash=sha256:2f6da418d1f1e0fddd844478f41680e794e6051915791a034ff65e5f100525a2 \
    --hash=sha256:f4324edc670a0f49750a81b895f35c3adb843cca46f0530f79fc1babb23789dc
    # via anyio
typing-extensions==4.13.2 ; python_full_version < '3.13' \
    --hash=sha256:a439e7c04b49fec3e5d3e2beaa21755cadbbdc391694e28ccdd36ca4a1408f8c \
    --hash=sha256:e6c81219bd689f51865d9e372991c540bda33a0379d5573cddb9a3a23f7caaef
    # via anyio
wcwidth==0.2.13 \
    --hash=sha256:3da69048e4540d84af32131829ff948f1e022c1c6bdb8d6102117aac784f6859 \
    --hash=sha256:72ea0c06399eb286d978fdedb6923a9eb47e1c486ce63e9b4e64fc18303972b5
    # via prettytable

shell_utils.sh

export JENKINS_UTILS_DIR="$HOME/jenkins"
export JENKINS_LOCKRES_SCRIPT="main.py"
export JENKINS_UTILS_PYTHON="./.venv/bin/python"

function lockres() {
    (cd $JENKINS_UTILS_DIR && $JENKINS_UTILS_PYTHON $JENKINS_LOCKRES_SCRIPT $@)
}