Source code for airflow.providers.jenkins.hooks.jenkins

#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
from __future__ import annotations

from typing import Any

import jenkins

from airflow.hooks.base import BaseHook


[docs]class JenkinsHook(BaseHook):
    """Hook to manage connection to jenkins server."""

[docs]    conn_name_attr = "conn_id"
[docs]    default_conn_name = "jenkins_default"
[docs]    conn_type = "jenkins"
[docs]    hook_name = "Jenkins"

    @classmethod
[docs]    def get_connection_form_widgets(cls) -> dict[str, Any]:
        """Return connection widgets to add to Jenkins connection form."""
        from flask_babel import lazy_gettext
        from wtforms import BooleanField

        return {
            "use_https": BooleanField(
                label=lazy_gettext("Use Https"),
                description="Specifies whether to use https scheme. Defaults to http",
            ),
        }

    @classmethod
[docs]    def get_ui_field_behaviour(cls) -> dict[str, Any]:
        """Return custom UI field behaviour for Jenkins connection."""
        return {
            "hidden_fields": ["schema", "extra"],
            "relabeling": {},
            "placeholders": {
                "login": "Login for the Jenkins service you would like to connect to",
                "password": "Password for the Jenkins service you would like to connect too",
                "host": "Host for your Jenkins server. Should NOT contain scheme (http:// or https://)",
                "port": "Specify a port number",
            },
        }

    def __init__(self, conn_id: str = default_conn_name) -> None:
        super().__init__()
        connection = self.get_connection(conn_id)
        self.connection = connection
        connection_prefix = "http"
        # connection.extra contains info about using https (true) or http (false)
        if connection.extra_dejson.get("use_https"):
            connection_prefix = "https"
        url = f"{connection_prefix}://{connection.host}:{connection.port}/{connection.schema}"
        self.log.info("Trying to connect to %s", url)
        self.jenkins_server = jenkins.Jenkins(url, connection.login, connection.password)

[docs]    def get_jenkins_server(self) -> jenkins.Jenkins:
        """Get jenkins server."""
        return self.jenkins_server

[docs]    def get_latest_build_number(self, job_name) -> int:
        self.log.info("Build number not specified, getting latest build info from Jenkins")
        job_info = self.jenkins_server.get_job_info(job_name)
        return job_info["lastBuild"]["number"]

[docs]    def get_build_result(self, job_name: str, build_number) -> str:
        build_info = self.jenkins_server.get_build_info(job_name, build_number)
        return build_info["result"]

[docs]    def get_build_building_state(self, job_name: str, build_number: int | None) -> bool:
        if not build_number:
            build_number_to_check = self.get_latest_build_number(job_name)
        else:
            build_number_to_check = build_number

        self.log.info("Getting build info for %s build number: #%s", job_name, build_number_to_check)
        build_info = self.jenkins_server.get_build_info(job_name, build_number_to_check)
        building = build_info["building"]
        return building