4.3. 带有 Python 的 API 请求
本节概述了如何将 Python 与 Satellite API 搭配使用来执行各种任务。
重要
以下是脚本和命令示例。确保您在使用前仔细检查这些脚本,并替换任何变量、用户名、密码和其他信息,以适应您自己的部署。
本节中的示例脚本不使用 SSL 验证与 REST API 交互。
4.3.1. 使用 Python 创建对象 复制链接链接已复制到粘贴板!
复制链接链接已复制到粘贴板!
此脚本连接到红帽卫星 6 API 并创建一个组织,然后在组织中创建三个环境。如果组织已存在,该脚本将使用该组织。如果组织中已存在任何环境,该脚本将引发错误并退出。
Python 2 示例
URL to your Satellite 6 server URL for the API to your deployed Satellite 6 server Katello-specific API Default credentials to login to Satellite 6 Ignore SSL for now Name of the organization to be either created or used Name for life cycle environments to be either created or used
#!/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()