awxkit python API: add example code

[azrdev]

ISSUE TYPE

• Feature Idea

SUMMARY

There are some few examples for using awxkit as a CLI tool in https://github.com/ansible/awx/blob/393369a/awxkit/awxkit/cli/docs/source/examples.rst

What is lacking, however, are examples for importing awxkit into own python code to interface the API. Given the dynamic nature of the library it's not easy to grasp from looking at its source code, either.

I'm aware that the awxkit python API is not considered stable (though I don't recall where I found that statement), but that might not be a hindrance for using it, so some example would be useful.

[azrdev]

I gathered so much from https://github.com/ansible/awx/issues/4492#issuecomment-522827735:

from awxkit import api, config, utils

config.base_url = "https://awx.example.com"
config.credentials = utils.PseudoNamespace({'default': {'username': 'admin', 'password': '12345'}})

connection = api.Api()
connection.load_session().get()
api_v2 = connection.available_versions.v2.get()

# dir(api_v2) == the members of the object are all the top-level /api/v2/ endpoints
# these have python methods `get`, `put`, `patch` etc.
# e.g.
print(api_v2.hosts.get())

[nodje]

Well done finding some python usage example!

Actually I think the issue title should be "provide Python API documentation" because there's simply none! And it's really lacking.

[manoj-maracheea-cko]

Thanks to you @azrdev,

I am able to launch a job like the one below. I don't know if this is the best way to go. I am also looking for getting the status of the job and monitor it. so I can decide what the next move depend on the status.

# Importing necessary Classes and Modules from awxkit.
from awxkit import api, config, utils, cli
from awxkit.api import ApiV2, job_templates, JobTemplateLaunch, base, pages, jobs
from awxkit.api.resources import resources

# Variables for config and login credentials.

config.base_url = base_url
config.credentials = utils.PseudoNamespace(
    {'default': {'username': user, 'password': password}})

# Initializing the connections

connection = ApiV2()
connection.load_session().get()
result = connection

# Getting the job_template results that contain the related name [ Use for the related URL ] using the jobs templates name

test = result.job_templates.get(name="configurations").results[0]

# Payload to launch with limit.

payload = {"limit": "xx.xx.xx.xxx"}


# Launch the Job, test as the related name and the payload 

launch = job_templates.JobTemplate.launch(test, payload = payload)
# Launch return the Job number and status.

Anyone can help with monitoring the job status. Thanking you in advance. I will keep trying, if I have something I will update it here.

[azrdev]

@manoj-maracheea-cko you'd look at job_events, either polling or maybe awxkit offers some notification/listening functionality.

btw, could you wrap your whole code block in ~~~ or ``` for readability?

[manoj-maracheea-cko]

@azrdev thanks I will try the job_events. and update here.

[nachobytes]

@manoj-maracheea-cko @azrdev Your comments helped me get started and I decided to write an article about this. Hope you find it useful. Open to suggestions as well.

https://medium.com/@nachobytes/awxkit-and-python-for-ansible-automation-effe1b637154

[manoj-maracheea-cko]

Great article @nachobytes, I may add more info here later this might help others as well. Great things you did.

I am happy that this has helped you, and wish we always can share our knowledge with others. Keep it up.

[manoj-maracheea-cko]

@nachobytes , find below what I was able to do.

from awxkit import api, config, utils, cli
from awxkit.api import ApiV2, job_templates, JobTemplateLaunch, base, pages, jobs
from awxkit.api.resources import resources
from awxkit.cli import custom
import json
import os

file = os.path.abspath(os.path.dirname(__file__) + "/.env.json")
try:
    data = json.load(open(file))
    base_url = data["base_url"]
    user = data["user"]
    password = data["password"]
except FileNotFoundError:
    base_url = os.getenv("BASE_URL")
    user = os.getenv("USER")
    password = os.getenv("PASSWORD")

config.base_url = base_url
config.credentials = utils.PseudoNamespace(
    {'default': {'username': user, 'password': password}})

connection = ApiV2()
connection.load_session().get()
result = connection


def run_jobs(payload, template_name):
    test = result.job_templates.get(name=template_name).results[0]
    launch = job_templates.JobTemplate.launch(test, payload=payload)
    job_id = launch.get().id
    job_url = base_url+f"/#/jobs/playbook/{job_id}/output"
   job_status = True
   while job_status or job_running:
        # Get the job status
        status = jobs.JobEvent.get(test).status 
        if  status == "successful":
             print("Job completed Successful")
             job_running = False
       elif status == "failed":
             print("Check job URL: ",  job_url  )
             job_running = False

From here on I think, anyone can do what is needed. as a primary starting point.

[isuftin]

Figured while this ticket is still open, I'd add a little more example. This updates a specific extra_var in a job template or workflow template

import datetime
import os
import json
import sys
from awxkit import config, utils
from awxkit.api import ApiV2
from awxkit.api.resources import resources
from awxkit.exceptions import Unauthorized

USERNAME = 'my-rundeck-service-account'
password = os.getenv('RD_OPTION_SERVICE_ACCOUNT_PASSWORD')
job_template_name = os.getenv('RD_OPTION_WORKFLOW_JOB_TEMPLATE_NAME')

# Create authenticated session
config.base_url = 'https://my-ansible-automation-platform.domain.name.com.gov.net.org.whatever/'
config.credentials = utils.PseudoNamespace(
    {'default': {'username': USERNAME, 'password': password}})
session_connection = ApiV2().load_session().get()
resources = session_connection.get(resources)

try:
    resources.me.get()
except Unauthorized:
    print('Invalid username or password was used to log in')
    sys.exit(1)

print(f'Looking for template by the name of {job_template_name}')
unified_job_templates = resources.unified_job_templates.get(
    name=job_template_name)
if unified_job_templates.count == 0:
    print(f'AAP template by name of {job_template_name} doesnt exist')
    sys.exit(1)

print(f'Found template by the name of {job_template_name}')
unified_job_template = unified_job_templates.results[0]

extra_vars = json.loads(unified_job_template.extra_vars)
print(f'Current variables: \n{json.dumps(extra_vars, indent=4, sort_keys=True)}')

now = datetime.datetime.now()
extra_vars['current_time'] = now.strftime("%m/%d/%Y, %H:%M:%S")

print(f'New variables: \n{json.dumps(extra_vars, indent=4, sort_keys=True)}')
unified_job_template.extra_vars = json.dumps(extra_vars)
unified_job_template.patch()