5.3. 在 Python 中调用 API

Python 3 示例

#!/usr/bin/python3

import json
import sys

try:
    import requests
except ImportError:
    print("Please install the python-requests module.")
    sys.exit(-1)

# URL to your Satellite Server
URL = "https://satellite.example.com"
FOREMAN_API = f"{URL}/api/"
KATELLO_API = f"{URL}/katello/api/"
POST_HEADERS = {'content-type': 'application/json'}
# Default credentials to login to Satellite 6
USERNAME = "admin"
PASSWORD = "changeme"
# Ignore SSL for now
SSL_VERIFY = False

# Name of the organization to be either created or used
ORG_NAME = "MyOrg"
# Name for life cycle environments to be either created or used
ENVIRONMENTS = ["Development", "Testing", "Production"]


def get_json(location):
    """
    Performs a GET by using the passed URL location
    """
    r = requests.get(location, auth=(USERNAME, PASSWORD), verify=SSL_VERIFY)
    return r.json()


def post_json(location, json_data):
    """
    Performs a POST and passes the data to the URL location
    """
    result = requests.post(
        location,
        data=json_data,
        auth=(USERNAME, PASSWORD),
        verify=SSL_VERIFY,
        headers=POST_HEADERS
    )
    return result.json()


def main():
    """
    Main routine that creates or re-uses an organization and
    life cycle environments.
    If life cycle environments already
    exist, exit out.
    """

    # Check if our organization already exists
    org = get_json(f"{FOREMAN_API}/organizations/{ORG_NAME}")

    # If our organization is not found, create it
    if org.get('error', None):
        org_id = post_json(
            f"{FOREMAN_API}/organizations/",
            json.dumps({"name": ORG_NAME})
        )["id"]
        print("Creating organization:\t" + ORG_NAME)
    else:
        # Our organization exists, so let's grab it
        org_id = org['id']
        print(f"Organization '{ORG_NAME}' exists.")

    # Now, let's fetch all available life cycle environments for this org...
    envs = get_json(
        f"{KATELLO_API}/organizations/{org_id}/environments/"
    )

    # ...and add them to a dictionary, with respective 'Prior' environment
    prior_env_id = 0
    env_list = {}
    for env in envs['results']:
        env_list[env['id']] = env['name']
        prior_env_id = env['id'] if env['name'] == "Library" else prior_env_id

    # Exit the script if at least one life cycle environment already exists
    if all(environment in env_list.values() for environment in ENVIRONMENTS):
        print("ERROR: One of the Environments is not unique to organization")
        sys.exit(-1)

    # Create life cycle environments
    for environment in ENVIRONMENTS:
        new_env_id = post_json(
            f"{KATELLO_API}/organizations/{org_id}/environments/",
            json.dumps({
                "name": environment,
                "organization_id": org_id,
                "prior": prior_env_id
            })
        )["id"]

        print("Creating environment:\t" + environment)
        prior_env_id = new_env_id


if __name__ == "__main__":
    main()

Python 3 示例

#!/usr/bin/env python3

import json
import sys

try:
    import requests
except ImportError:
    print("Please install the python-requests module.")
    sys.exit(-1)

HOSTNAME = "satellite.example.com"
# URL for the API to your Satellite Server
FOREMAN_API = f"https://{HOSTNAME}/api/"
KATELLO_API = f"https://{HOSTNAME}/katello/api/v2/"

POST_HEADERS = {'content-type': 'application/json'}
# Default credentials to login to Satellite 6
USERNAME = "admin"
PASSWORD = "password"
# Ignore SSL for now
SSL_VERIFY = False
#SSL_VERIFY = "./path/to/CA-certificate.crt" # Put the path to your CA certificate here to allow SSL_VERIFY


def get_json(url):
    # Performs a GET by using the passed URL location
    r = requests.get(url, auth=(USERNAME, PASSWORD), verify=SSL_VERIFY)
    return r.json()

def get_results(url):
    jsn = get_json(url)
    if jsn.get('error'):
        print("Error: " + jsn['error']['message'])
    else:
        if jsn.get('results'):
            return jsn['results']
        elif 'results' not in jsn:
            return jsn
        else:
            print("No results found")
    return None

def display_all_results(url):
    results = get_results(url)
    if results:
        print(json.dumps(results, indent=4, sort_keys=True))

def display_info_for_hosts(url):
    hosts = get_results(url)
    if hosts:
        print(f"{'ID':10}{'Name':40}{'IP':30}{'Operating System':30}")
        for host in hosts:
            print(f"{str(host['id']):10}{host['name']:40}{str(host['ip']):30}{str(host['operatingsystem_name']):30}")

def display_info_for_subs(url):
    subs = get_results(url)
    if subs:
        print(f"{'ID':10}{'Name':90}{'Start Date':30}")
        for sub in subs:
            print(f"{str(sub['id']):10}{sub['name']:90}{str(sub['start_date']):30}")

def main():
    host = HOSTNAME
    print(f"Displaying all info for host {host} ...")
    display_all_results(FOREMAN_API + 'hosts/' + host)

    print(f"Displaying all facts for host {host} ...")
    display_all_results(FOREMAN_API + f'hosts/{host}/facts')

    host_pattern = 'example'
    print(f"Displaying basic info for hosts matching pattern '{host_pattern}'...")
    display_info_for_hosts(FOREMAN_API + 'hosts?per_page=1&search=name~' + host_pattern)

    print(f"Displaying basic info for subscriptions")
    display_info_for_subs(KATELLO_API + 'subscriptions')

    environment = 'production'
    print(f"Displaying basic info for hosts in environment {environment}...")
    display_info_for_hosts(FOREMAN_API + 'hosts?search=environment=' + environment)


if __name__ == "__main__":
    main()