Portainer Gist

portainer_stack.py

#!/usr/bin/env python3
"""Script to start/stop portainer stacks"""

import json
import os
import sys
from datetime import datetime
from enum import Enum
from typing import NamedTuple, TypedDict, Union
from urllib.parse import urljoin

import requests
from dotenv import load_dotenv

load_dotenv()  # take environment variables from .env.

START = "start"
STOP = "stop"
SCRIPT_OPTIONS = [START, STOP]
BASE_SERVER_URL = os.getenv("BASE_SERVER_URL", "")
SSL_CERT_PATH = os.getenv("SSL_CERT_PATH", "")
TIMEOUT = 180


class StackType(TypedDict):
    """
    Endpoint structure from api payload response

    Args:
        TypedDict (_type_): type hint
        Id (int): Stack Identifier
        Status (int): Stack status (1 - active, 2 - inactive)
        EndpointId (int): Environment(Endpoint) identifier.
    """

    EndpointId: int
    Id: int
    Status: int


class PortainerStack(NamedTuple):
    """Stack Object
    - stack_id: Stack Identifier
    - status: Stack status (1 - active, 2 - inactive)
    - endpoint_id: Environment(Endpoint) identifier.
    """

    endpoint_id: int
    stack_id: int
    status: int


class Status(Enum):
    """Status values for stacks"""

    STACK_ACTIVE = 1
    STACK_INACTIVE = 2


def get_new_token() -> str:
    """get token from admin user"""

    auth_url: str = urljoin(base=BASE_SERVER_URL, url="api/auth")
    body = {"username": os.getenv("USERNAME"), "password": os.getenv("PASSWORD")}
    token_response = requests.post(
        url=auth_url, json=body, verify=SSL_CERT_PATH, timeout=TIMEOUT
    )

    if token_response.status_code != 200:
        print("Failed to obtain token from the OAuth 2.0 server")
        print(token_response)
        print(token_response.status_code)
        sys.exit(1)

    return json.loads(token_response.text)["jwt"]


def list_stacks(token: str) -> list[StackType]:
    """List stacks in portainer"""

    stacks_url = urljoin(base=BASE_SERVER_URL, url="api/stacks")
    headers = {"Authorization": f"Bearer {token}"}
    stacks = requests.get(
        stacks_url, headers=headers, verify=SSL_CERT_PATH, timeout=TIMEOUT
    )

    return json.loads(stacks.text)


def get_htpc_stack_information(stack_record: StackType) -> PortainerStack:
    """returns stack information in case id changes,
    Expand by filtering for specific stack name in the future"""

    return PortainerStack(
        stack_id=stack_record.get("Id"),
        endpoint_id=stack_record.get("EndpointId"),
        status=stack_record.get("Status"),
    )


def stack_action(portainer_stack: PortainerStack, token: str, action: str) -> str:
    """funtion that stats/stops stacks"""
    stacks_url = urljoin(
        base=BASE_SERVER_URL, url=f"api/stacks/{portainer_stack.stack_id}/{action}"
    )
    params = {"endpointId": portainer_stack.endpoint_id}
    headers = {"Authorization": f"Bearer {token}"}

    print(
        "Attempting to "
        f"{action} stack: {portainer_stack.stack_id} endpoint_id: {portainer_stack.endpoint_id}"
    )
    stack_response = requests.post(
        url=stacks_url,
        headers=headers,
        params=params,
        verify=SSL_CERT_PATH,
        timeout=TIMEOUT,
    )

    if stack_response.status_code != 200:
        print("Failed to start stack")
        print(stack_response)
        print(stack_response.status_code)
        sys.exit(2)

    return json.loads(stack_response.text)


def parse_action(action: str) -> Union[None, str]:
    """validates proper options are passed"""
    if action in SCRIPT_OPTIONS:
        return action

    return None


def validate_action_status(action: str, portainer_stack: PortainerStack) -> bool:
    """validates if action is already executed"""
    if action == START and portainer_stack.status == Status.STACK_ACTIVE.value:
        print("Stack is already started")
        return True

    if action == STOP and portainer_stack.status == Status.STACK_INACTIVE.value:
        print("Stack is already stopped")
        return True

    return False


def main(arguments: list[str]) -> None:
    """log start time and any errors"""
    dt_string = datetime.now().strftime("%d/%m/%Y %H:%M:%S")
    print(f"Starting: {dt_string}")
    action_argument = None

    if arguments:
        action_argument = parse_action(action=arguments[0])

    token = get_new_token()
    stack_list = list_stacks(token=token)
    portainer_stack = get_htpc_stack_information(stack_record=stack_list[0])
    print("Status:", Status(portainer_stack.status).name)

    if action_argument and validate_action_status(
        action=action_argument, portainer_stack=portainer_stack
    ):
        return None

    if portainer_stack.status == Status.STACK_ACTIVE.value:
        print("Stack Available. Stoping stack.")
        response = stack_action(
            portainer_stack=portainer_stack,
            token=token,
            action=STOP,
        )
        print(response)

        return None

    print("Stack Unavailable. Starting stack.")
    response = stack_action(
        portainer_stack=portainer_stack,
        token=token,
        action=START,
    )
    print(response)
    return None


if __name__ == "__main__":
    main(sys.argv[1:])