1#@!#!123s

: / opt / support / venv / lib / python3.13 / site-packages / rads /
Filename : semaphore.py
back
"""Ansible Semaphore API"""

import time
import requests


class AnsibleSemaphoreAPI:
    """Ansible Semaphore API"""

    def __init__(self, base_url: str, token: str):
        self.base_url = base_url
        self.token = token
        self.headers = {
            "Authorization": f"Bearer {self.token}",
            "Content-Type": "application/json",
        }

    def get(self, endpoint: str, timeout: int = 30) -> dict:
        """HTTP GET request to the Ansible Semaphore API

        Args:
            endpoint (str): API endpoint to access.
            timeout (int): request timeout. Defaults to 30.

        Returns:
            dict: API response
        """
        url = f"{self.base_url}/{endpoint}"
        response = requests.get(url, headers=self.headers, timeout=timeout)
        response.raise_for_status()
        return response.json()

    def post(self, endpoint: str, data: dict, timeout: int = 30):
        """HTTP POST request to the Ansible Semaphore API

        Args:
            endpoint (str): API endpoint to access.
            data (dict): data to serialize as JSON and send with the request.
            timeout (int): request timeout. Defaults to 30.

        Returns:
            dict: API response
        """
        url = f"{self.base_url}/{endpoint}"
        response = requests.post(
            url, headers=self.headers, json=data, timeout=timeout
        )
        response.raise_for_status()
        return response.json()

    def put(self, endpoint: str, data: dict, timeout: int = 30):
        """HTTP PUT request to the Ansible Semaphore API

        Args:
            endpoint (str): API endpoint to access.
            data (dict): data to serialize as JSON and send with the request.
            timeout (int): request timeout. Defaults to 30.

        Returns:
            dict: API response
        """
        url = f"{self.base_url}/{endpoint}"
        response = requests.put(
            url, headers=self.headers, json=data, timeout=timeout
        )
        response.raise_for_status()
        return response.json()

    def delete(self, endpoint: str, timeout: int = 30):
        """HTTP DELETE request to the Ansible Semaphore API

        Args:
            endpoint (str): API endpoint to access.
            timeout (int): request timeout. Defaults to 30.

        Returns:
            dict: API response
        """
        url = f"{self.base_url}/{endpoint}"
        response = requests.delete(url, headers=self.headers, timeout=timeout)
        response.raise_for_status()
        return response.json()

    def create_template(self, project_id: int, template_data: dict):
        """Creates a template in the specified project.

        Args:
            project_id (int): The ID of the project where the template will be
                created.
            template_data (dict): The data for the new template.

        Returns:
            dict: The created template's details.
        """
        endpoint = f"project/{project_id}/templates"
        return self.post(endpoint, template_data)

    def run_task(self, project_id: int, template_id: int, **kwargs):
        """Runs a task using the specified template and options.

        Args:
            project_id (int): The ID of the project where the task will be ran.
            template_id (int): The ID of the template to run.

        Returns:
            dict: The response from the API.
        """
        task_data = {
            "template_id": template_id,
            "debug": kwargs.get('debug', False),
            "dry_run": kwargs.get('dry_run', False),
            "diff": kwargs.get('diff', False),
            "playbook": kwargs.get('playbook', ""),
            "environment": kwargs.get('environment', ""),
            "limit": kwargs.get('limit', ""),
        }

        return self.post(f"project/{project_id}/tasks", task_data)

    def get_task(self, project_id: int, task_id: int):
        """Retrieves the details of a specific task.

        Args:
            project_id (int): The ID of the project where the template will be
                created.
            task_id (int): The ID of the task to retrieve.

        Returns:
            dict: The task details.
        """
        endpoint = f"/project/{project_id}/tasks/{task_id}"
        return self.get(endpoint)

    def wait_for_task(
        self, project_id: int, task_id: int, poll_interval: int = 5
    ):
        """Polls a task for its status until completion.

        Args:
            project_id (int): The ID of the project where the task was run.
            task_id (int): The ID of the task to poll.
            poll_interval (int): The interval in seconds between status checks.

        Returns:
            dict: The final task status.
        """
        while True:
            task_status = self.get_task(project_id, task_id)
            status = task_status['status']

            if status in ['success', 'error']:
                break

            print(
                f"Task {task_id} is {status}. Waiting for {poll_interval} "
                "seconds before next check..."
            )
            time.sleep(poll_interval)

        print(f"Task {task_id} completed with {status}.")
        return task_status