第4章 異なる言語での API 要求

本章では、curl、Ruby および Python を使用して Red Hat Satellite に API 要求を送信する方法を、例と併せて説明します。

4.1. curl を使用した API 要求

本セクションでは、Satellite API で curl を使用して、さまざまなタスクを実行する方法を説明します。

Red Hat Satellite は HTTPS と、デフォルトではホストの識別用に証明書を使用する必要があります。「SSL 認証の概要」の記載どおりに Satellite サーバーの証明書を追加していない場合には、--insecure オプションを使用して証明書チェックを省略することができます。

ユーザー認証の場合は、--user オプションを指定して、--user username:password 形式で、Satellite ユーザー認証情報を指定します。または、パスワードを含めない場合には、コマンドでパスワードを入力するようにプロンプトが表示されます。セキュリティーリスクを軽減するには、コマンドの一部としてパスワードを入力すると、シェルの履歴の一部として残ってしまうので、入力しないようにしてください。本セクションの例では、簡素化の目的でのみ、パスワードを含めています。

--silent オプションを使用する場合は、curl で進捗メーターやエラーメッセージは表示されない点に注意してください。

本章の例では、Python json.tool モジュールを使用して出力をフォーマットしています。

4.1.1. API 要求への JSON データの指定

データを API 要求で Satellite Server に渡すことができます。このデータは、JSON 形式でなければなりません。--data オプションで、JSON データを指定する場合には、--header オプションを使用して、以下の HTTP ヘッダーを設定する必要があります。

--header "Accept:application/json,version=2" \
--header "Content-Type:application/json"

以下のオプションのいずれかを使用して、--data オプションでデータを追加します。

  1. 引用符の付いた JSON 形式のデータを中括弧 {} 囲みます。JSON タイプのパラメーターの値を渡す場合には、バックスラッシュ \ を使用して、引用符 " をエスケープする必要があります。たとえば、中括弧内では "Example JSON Variable"\"Example JSON Variable\" のような形式にする必要があります。

    --date {"id":44, "smart_class_parameter":{"override":"true", "parameter_type":"json", "default_value":"{\"GRUB_CMDLINE_LINUX\": {\"audit\":\"1\",\"crashkernel\":\"true\"}}"}}
  2. 引用符なしの JSON 形式のデータは、@ マークとファイル名を指定します。

    --data @file.json

    JSON 形式データの外部ファイルを使用すると、以下のような利点があります。

    • 任意のテキストエディターを使用できる
    • 構文チェッカーを使用してミスを特定し、回避できる
    • ツールを使用して JSON データの妥当性を確認して、再フォーマットができる

JSON ファイルの検証

json_verify ツールを使用して、JSON ファイルの妥当性を確認します。

$ json_verify < test_file.json

4.1.2. リソース一覧の取得

本セクションは、 Satellite 6 API で curl を使用して、Satellite デプロイメントから情報を要求する方法を説明します。これらの例には、要求と応答の両方が含まれます。デプロイメントごとに、異なる結果が得られることが想定されます。

ユーザーの一覧表示

以下の例は、Satellite リソース (今回の例では、Satellite ユーザー) の一覧を返す基本的な要求です。このような要求では、メタデータでラップされたデータ一覧を返しますが、他の要求タイプでは実際のオブジェクトが返されます。

要求例:

$ curl --request GET --insecure --user sat_username:sat_password \
https://satellite.example.com/api/users | python -m json.tool

応答例:

{
    "page": 1,
    "per_page": 20,
    "results": [
        {
            "admin": false,
            "auth_source_id": 1,
            "auth_source_name": "Internal",
            "created_at": "2018-09-21 08:59:22 UTC",
            "default_location": null,
            "default_organization": null,
            "description": "",
            "effective_admin": false,
            "firstname": "",
            "id": 5,
            "last_login_on": "2018-09-21 09:03:25 UTC",
            "lastname": "",
            "locale": null,
            "locations": [],
            "login": "test",
            "mail": "example@domain.com",
            "organizations": [
                {
                    "id": 1,
                    "name": "Default Organization"
                }
            ],
            "ssh_keys": [],
            "timezone": null,
            "updated_at": "2018-09-21 09:04:45 UTC"
        },
        {
            "admin": true,
            "auth_source_id": 1,
            "auth_source_name": "Internal",
            "created_at": "2018-09-20 07:09:41 UTC",
            "default_location": null,
            "default_organization": {
                "description": null,
                "id": 1,
                "name": "Default Organization",
                "title": "Default Organization"
            },
            "description": "",
            "effective_admin": true,
            "firstname": "Admin",
            "id": 4,
            "last_login_on": "2018-12-07 07:31:09 UTC",
            "lastname": "User",
            "locale": null,
            "locations": [
                {
                    "id": 2,
                    "name": "Default Location"
                }
            ],
            "login": "admin",
            "mail": "root@example.com",
            "organizations": [
                {
                    "id": 1,
                    "name": "Default Organization"
                }
            ],
            "ssh_keys": [],
            "timezone": null,
            "updated_at": "2018-11-14 08:19:46 UTC"
        }
    ],
    "search": null,
    "sort": {
        "by": null,
        "order": null
    },
    "subtotal": 2,
    "total": 2
}

4.1.3. リソースの作成および変更

本章では、Satellite 6 API で curl を使用して Satellite Server のリソースを操作する方法を説明します。これらの API 呼び出しでは、API 呼び出しの json 形式でデータを渡す必要があります。詳細情報は、「API 要求への JSON データの指定」を参照してください。

ユーザーの作成

以下の例では、--data オプションを使用してユーザーを作成し、必要な情報を指定します。

要求例:

$ curl --header "Accept:application/json,version=2" \
--header "Content-Type:application/json" --request POST \
--user sat_username:sat_password --insecure \
--data "{\"firstname\":\"Test Name\",\"mail\":\"test@example.com\",\"login\":\"test_user\",\"password\":\"password123\",\"auth_source_id\":1}" \
https://satellite.example.com/api/users | python -m json.tool

ユーザーの変更

以下の例では、「ユーザーの作成」で作成した test_user のログインと名前を変更します。

要求例:

$ curl --header "Accept:application/json,version=2" \
--header "Content-Type:application/json" --request PUT \
--user sat_username:sat_password --insecure \
--data "{\"firstname\":\"New Test Name\",\"mail\":\"test@example.com\",\"login\":\"new_test_user\",\"password\":\"password123\",\"auth_source_id\":1}" \
https://satellite.example.com/api/users/8 | python -m json.tool

4.2. Ruby を使用した API 要求

本セクションでは、Satellite API で Ruby を使用して、さまざまなタスクを実行する方法を説明します。

重要

以下は、スクリプトおよびコマンドの例です。以下のスクリプトを慎重にレビューしてから使用するようにしてください。変数、ユーザー名、パスワード、その他の情報は、お使いのデプロイメントに適した値に置き換えてください。

4.2.1. Ruby を使用したオブジェクトの作成

以下のスクリプトは Red Hat Satellite 6 API と接続して、組織を作成し、その新規組織内に 3 つの環境を作成します。組織がすでに存在する場合には、このスクリプトはその組織を使用します。組織内に環境が 1 つでも存在する場合は、スクリプトによりエラーが送出されて、スクリプトは終了します。

#!/usr/bin/ruby

require 'rest-client'
require 'json'

url = 'https://satellite.example.com/api/v2/'
katello_url = "#{url}/katello/api/v2/"

$username = 'admin'
$password = 'changeme'

org_name = "MyOrg"
environments = [ "Development", "Testing", "Production" ]

# Performs a GET using the passed URL location
def get_json(location)
  response = RestClient::Request.new(
    :method => :get,
    :url => location,
    :user => $username,
    :password => $password,
    :headers => { :accept => :json,
    :content_type => :json }
  ).execute
  JSON.parse(response.to_str)
end

# Performs a POST and passes the data to the URL location
def post_json(location, json_data)
  response = RestClient::Request.new(
    :method => :post,
    :url => location,
    :user => $username,
    :password => $password,
    :headers => { :accept => :json,
    :content_type => :json},
    :payload => json_data
  ).execute
  JSON.parse(response.to_str)
end

# Creates a hash with ids mapping to names for an array of records
def id_name_map(records)
  records.inject({}) do |map, record|
    map.update(record['id'] => record['name'])
  end
end

# Get list of existing organizations
orgs = get_json("#{katello_url}/organizations")
org_list = id_name_map(orgs['results'])

if !org_list.has_value?(org_name)
  # If our organization is not found, create it
  puts "Creating organization: \t#{org_name}"
  org_id = post_json("#{katello_url}/organizations", JSON.generate({"name"=> org_name}))["id"]
else
  # Our organization exists, so let's grab it
  org_id = org_list.key(org_name)
  puts "Organization \"#{org_name}\" exists"
end

# Get list of organization's lifecycle environments
envs = get_json("#{katello_url}/organizations/#{org_id}/environments")
env_list = id_name_map(envs['results'])
prior_env_id = env_list.key("Library")

# Exit the script if at least one life cycle environment already exists
environments.each do |e|
  if env_list.has_value?(e)
    puts "ERROR: One of the Environments is not unique to organization"
    exit
  end
end

 # Create life cycle environments
environments.each do |environment|
  puts "Creating environment: \t#{environment}"
  prior_env_id = post_json("#{katello_url}/organizations/#{org_id}/environments", JSON.generate({"name" => environment, "organization_id" => org_id, "prior_id" => prior_env_id}))["id"]
end

4.2.2. Ruby での Apipie バインディングの使用

Apipie バインディングは、apipie で記述された API 呼び出しの Ruby バインディングです。このバインディングは、Satellite から API 定義を取得してキャッシュし、オンデマンドで API 呼び出しを生成します。以下の例では、組織を作成し、その新規組織内に 3 つの環境を作成します。組織がすでに存在する場合には、このスクリプトはその組織を使用します。組織内に環境が 1 つでも存在する場合は、スクリプトによりエラーが送出されて、スクリプトは終了します。

#!/usr/bin/tfm-ruby

require 'apipie-bindings'

org_name = "MyOrg"
environments = [ "Development", "Testing", "Production" ]

# Create an instance of apipie bindings
@api = ApipieBindings::API.new({
  :uri => 'https://satellite.example.com/',
  :username => 'admin',
  :password => 'changeme',
  :api_version => 2
})

# Performs an API call with default options
def call_api(resource_name, action_name, params = {})
  http_headers = {}
  apipie_options = { :skip_validation => true }
  @api.resource(resource_name).call(action_name, params, http_headers, apipie_options)
end

# Creates a hash with IDs mapping to names for an array of records
def id_name_map(records)
  records.inject({}) do |map, record|
    map.update(record['id'] => record['name'])
  end
end

# Get list of existing organizations
orgs = call_api(:organizations, :index)
org_list = id_name_map(orgs['results'])

if !org_list.has_value?(org_name)
  # If our organization is not found, create it
  puts "Creating organization: \t#{org_name}"
  org_id = call_api(:organizations, :create, {'organization' => { :name => org_name }})['id']
else
  # Our organization exists, so let's grab it
  org_id = org_list.key(org_name)
  puts "Organization \"#{org_name}\" exists"
end

# Get list of organization's life cycle environments
envs = call_api(:lifecycle_environments, :index, {'organization_id' => org_id})
env_list = id_name_map(envs['results'])
prior_env_id = env_list.key("Library")

# Exit the script if at least one life cycle environment already exists
environments.each do |e|
  if env_list.has_value?(e)
    puts "ERROR: One of the Environments is not unique to organization"
    exit
  end
end

 # Create life cycle environments
environments.each do |environment|
  puts "Creating environment: \t#{environment}"
  prior_env_id = call_api(:lifecycle_environments, :create, {"name" => environment, "organization_id" => org_id, "prior_id" => prior_env_id })['id']
end

4.3. Python を使用した API 要求

本セクションでは、Satellite API で Python を使用して、さまざまなタスクを実行する方法を説明します。

重要

以下は、スクリプトおよびコマンドの例です。以下のスクリプトを慎重にレビューしてから使用するようにしてください。変数、ユーザー名、パスワード、その他の情報は、お使いのデプロイメントに適した値に置き換えてください。

本セクションのスクリプト例では、REST API との対話に SSL 検証を使用しません。

4.3.1. Python を使用したオブジェクトの作成

以下のスクリプトは Red Hat Satellite 6 API と接続して、組織を作成し、その新規組織内に 3 つの環境を作成します。組織がすでに存在する場合には、このスクリプトはその組織を使用します。組織内に環境が 1 つでも存在する場合は、スクリプトによりエラーが送出されて、スクリプトは終了します。

Python 2 Example

#!/usr/bin/python

import json
import sys

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

# URL to your Satellite 6 server
URL = "https://satellite.example.com"
# URL for the API to your deployed Satellite 6 server
SAT_API = "%s/katello/api/v2/" % URL
# Katello-specific API
KATELLO_API = "%s/katello/api/" % URL
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 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(SAT_API + "organizations/" + ORG_NAME)

    # If our organization is not found, create it
    if org.get('error', None):
        org_id = post_json(
            SAT_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 "Organization '%s' exists." % ORG_NAME

    # Now, let's fetch all available life cycle environments for this org...
    envs = get_json(
        SAT_API + "organizations/" + str(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(
            SAT_API + "organizations/" + str(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()

4.3.2. Python を使用した APIからの情報要求

以下は、さまざまな API 要求に Pythonを使用するスクリプト例です。

Python 2 Example

#!/usr/bin/python
import json
import sys
try:
    import requests
except ImportError:
    print "Please install the python-requests module."
    sys.exit(-1)

SAT_API = 'https://satellite.example.com/api/v2/'
USERNAME = "admin"
PASSWORD = "password"
SSL_VERIFY = False   # Ignore SSL for now

def get_json(url):
    # Performs a GET 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:
        for host in hosts:
            print "ID: %-10d Name: %-30s IP: %-20s OS: %-30s" % (host['id'], host['name'], host['ip'], host['operatingsystem_name'])

def main():
    host = 'satellite.example.com'
    print "Displaying all info for host %s ..." % host
    display_all_results(SAT_API + 'hosts/' + host)

    print "Displaying all facts for host %s ..." % host
    display_all_results(SAT_API + 'hosts/%s/facts' % host)

    host_pattern = 'example'
    print "Displaying basic info for hosts matching pattern '%s'..." % host_pattern
    display_info_for_hosts(SAT_API + 'hosts?search=' + host_pattern)

    environment = 'production'
    print "Displaying basic info for hosts in environment %s..." % environment
    display_info_for_hosts(SAT_API + 'hosts?search=environment=' + environment)

    model = 'RHEV Hypervisor'
    print "Displaying basic info for hosts with model name %s..." % model
    display_info_for_hosts(SAT_API + 'hosts?search=model="' + model + '"')

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)

SAT = "satellite.example.com"
# URL for the API to your deployed Satellite 6 server
SAT_API = f"https://{SAT}/api/"
KATELLO_API = f"https://{SAT}/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 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 = SAT
    print(f"Displaying all info for host {host} ...")
    display_all_results(SAT_API + 'hosts/' + host)

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

    host_pattern = 'example'
    print(f"Displaying basic info for hosts matching pattern '{host_pattern}'...")
    display_info_for_hosts(SAT_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(SAT_API + 'hosts?search=environment=' + environment)


if __name__ == "__main__":
    main()