Intro
If you have Ansible automation within your organization that needs to communicate with a custom API, you can solve this in several ways. Perhaps a few curl commands in bash lines with `ansible.builtin.shell`. It’s not pretty, but you could create a task with inputs. A better approach would be to create a module. It sounds complicated, but it’s actually quite manageable if you have a little Python knowledge.
The example API
For this example, we have a simple API that we need to call from Ansible. The power of Ansible is that it only makes changes when necessary (idempotent). So, you need to take this into account in your module. Compare existing values with what (possibly) needs to be changed.
We’re using a Docker container to run a demo API. You can start it with:
docker run -p 5041:8080 opvolger/demo-api-ansible
Afterward, you can view the Swagger definition at the following URL:
It’s a key-value store, with the key always being a single uppercase letter and the value being a number. This API allows you to retrieve, modify, or add a current value.
Create the module
An Ansible module must be written in Python (for Linux). This should be placed under your playbook folder in the “library” folder. You can find more information on the Ansible docs: https://docs.ansible.com/ansible/latest/dev_guide/developing_modules_general.html
Figure out what your input variables should be
We have a simple API that can get, set, change and delete a key-value pair. The key must be a letter and the value a number. The API needs authentication with username and password or with a token.
In Ansible we need to build a module that can set/get a character-key/number-value and clear all characters. This is what we need in the playbooks.
So we need:
- the arguments character and number (key and value)
- 3 actions in the module: set, get and clear
- authentication with username/password OR the use of a token
- the endpoint where we can call the API.
Something like this:
- name: Set number on a character with username and password
api_demo:
endpoint: http://localhost:5041/
username: user
password: password
character: "A"
number: 4
action: set
delegate_to: localhostStart with the documentation
If you know how to call the module. We can start with the documentation and empty implementation. We use the class AnsibleModule for communication back to Ansible and set the variable `DOCUMENTATION`, `EXAMPLES` and `RETURN`.
Here is my implementation:
"""Ansible module that has only documentation."""
from ansible.module_utils.basic import AnsibleModule
DOCUMENTATION = r'''
---
module: api_demo
author:
- Bas Magré (@opvolger)
short_description: The ability to create, remove and manage a list of characters that have numbers
version_added: 0.0.1
description: "The ability to create, remove and manage a list of characters that have numbers. This is just a demo!"
options:
endpoint:
description: The uri of the API
type: str
required: true
sample: 'http://localhost:5041/'
username:
description: Username to get a token
type: str
required: false
sample: 'user'
password:
description: Password to get a token
type: str
required: false
sample: 'password'
token:
description: Use a token direct (without username/password)
type: str
required: false
sample: 'secret'
character:
description: Character (key)
type: str
required: false
sample: 'A'
number:
description: Number (value)
type: int
required: false
sample: 5
action:
description: The action to perform
type: str
required: true
default: get
choices: [ get, set, clear ]
sample: 'set'
'''
EXAMPLES = r'''
- name: Clear all set character and numbers with a token
api_demo:
endpoint: http://localhost:5041/
token: secret
action: clear
delegate_to: localhost
- name: Set number on a character with username and password
api_demo:
endpoint: http://localhost:5041/
username: user
password: password
character: "A"
number: 4
action: set
delegate_to: localhost
- name: Get number from set character with a token
api_demo:
endpoint: http://localhost:5041/
token: secret
character: "A"
action: get
delegate_to: localhost
'''
RETURN = r'''
exists:
description: if the character is set
returned: success
type: bool
sample: True
number:
description: The number that is set or get
type: int
sample: 5
'''
def run_module() -> None:
"""The Ansible module."""
# define the available arguments/parameters that a user can pass to the module
module_args = {
'endpoint': {'type': 'str', 'required': True},
'username': {'type': 'str', 'required': False},
'password': {'type': 'str', 'required': False, 'no_log': True},
'token': {'type': 'str', 'required': False, 'no_log': True},
'character': {'type': 'str', 'required': False},
'number': {'type': 'int', 'required': False},
'action': {'type': 'str', 'required': True, 'choices': ['get', 'set', 'clear']}
}
# the AnsibleModule object will be our abstraction working with Ansible
# this includes instantiation, a couple of common attr would be the
# args/params passed to the execution, as well as if the module
# supports check mode
module = AnsibleModule(
argument_spec=module_args,
)
# seed the result dict in the object
# we primarily care about changed and state
# change is if this module effectively modified the target
# state will include any data that you want your module to pass back
# for consumption, for example, in a subsequent task
result = {
'changed': False,
'rc': 1,
'stdout': None,
'stderr': None
}
result['rc'] = 0 # we are at the end, no errors occurred
module.exit_json(**result)
def main() -> None:
"""Main function to run Ansible Module."""
run_module()
if __name__ == '__main__':
main()Now we can test if the documentation is correct with the commands.
# test doc generation
cd ansible-playbook
ANSIBLE_LIBRARY=./library ansible-doc -t module api_demo_start_docThis should not give an error, but should display your module’s documentation.
Continue with arguments
Our module can work with an username/password or with a token. There are more arguments that should or shouldn’t be used in combination.
- if we use an username we need a password
- we need or username/password or token
- we can’t user username/password and token together
- if we use the get command we need a character
- if we use the set command we need a character and number
We can implement this in the module like this:
# use username with password
check_required_together = [
('username', 'password')
]
# use username/password or token is needed
check_required_one_of = [('username', 'token')]
# use username/password or token, only one
check_mutually_exclusive = [('username', 'token')]
# if action == get, we need the character argument
# if action == set, we need the character and number arguments
check_required_if = [
('action', 'get', ['character']),
('action', 'set', ['character', 'number'])
]
# the AnsibleModule object will be our abstraction working with Ansible
# this includes instantiation, a couple of common attr would be the
# args/params passed to the execution, as well as if the module
# supports check mode
module = AnsibleModule(
argument_spec=module_args,
supports_check_mode=True,
required_if=check_required_if,
required_together=check_required_together,
required_one_of=check_required_one_of,
mutually_exclusive=check_mutually_exclusive
)You can put the arrays directly in the constructor of AnsibleModule, but I like to make comments on my variable. For more information see the Ansible docs https://docs.ansible.com/ansible/latest/reference_appendices/module_utils.html
Make a class for the interaction
To maintain a clear overview and simplify development, you can encapsulate the entire interaction within a class. The advantage of a separate class is that you can essentially reuse this code outside of the Ansible module. Ideally, you could create a separate module for this and maintain it as a Python packages (https://packaging.python.org/en/latest/tutorials/packaging-projects/), provided you can/are allowed to make this code public.
For this example, I will eventually include the class in my Ansible module. But it could also have been an import, of course.
The module:
"""Module providing calls to the demo api."""
from typing import List
from urllib.parse import urljoin
import json
import requests
class DemoApi:
"""
A simple demo class where the API logic is written
:param username: user that connect to API
:param password: password from the user
:param token: token can be user instead of username/password
:param uri: the endpoint of the API
:raises HTTPError: if one occurred
"""
def __init__(self, username: str, password: str, token: str, uri: str):
self.uri = uri
self.session = requests.session()
self.__connect(username, password, token)
def __connect(self, username: str, password: str, token: str):
if token:
self.session.headers.update({'X-Auth-Token': token})
else:
response = self.session.post(urljoin(self.uri, "token"), json={
"username": username, "password": password})
response.raise_for_status()
self.session.headers.update({'X-Auth-Token': response.text})
def reset(self, character: str) -> None:
"""
Reset will remove character from the set of characters that are set
:param character: character to reset
:raises HTTPError: if one occurred
"""
response = self.session.delete(
urljoin(self.uri, f"character/{character}"))
response.raise_for_status()
def set(self, character: str, number: int) -> None:
"""
Set the number on a character
:param character: character to set
:param number: the number that will be given to the character
:raises HTTPError: if one occurred
"""
response = self.session.put(
urljoin(self.uri, f"character/{character}?number={number}"))
response.raise_for_status()
def update(self, character: str, number: int) -> None:
"""
Update the number on a character
:param character: character to update
:param number: the number that will be given to the character
:raises HTTPError: if one occurred
"""
response = self.session.post(
urljoin(self.uri, f"character/{character}?number={number}"))
response.raise_for_status()
def get(self, character: str) -> int:
"""
Get the number that is set on a character
:param character: character where you want the number from
:returns: the number that will be given to the character
:raises HTTPError: if one occurred
"""
response = self.session.get(
urljoin(self.uri, f"character/{character}"))
response.raise_for_status()
return json.loads(response.text)
def list(self) -> List[str]:
"""
Get the list of characters that are set
:returns: the list of characters that have a number
:raises HTTPError: if one occurred
"""
response = self.session.get(urljoin(self.uri, "character"))
response.raise_for_status()
return json.loads(response.text)and some tests, to test the code we made
"""Test module for DemoApi."""
import unittest
from demoapi import DemoApi
class TestApi(unittest.TestCase):
"""Test Class"""
def setUp(self):
self.demo_api = DemoApi('user', 'password', None,
'http://localhost:5041/')
# clear all characters (if any)
characters = self.demo_api.list()
for character in characters:
self.demo_api.reset(character)
def test_username(self) -> None:
"""Test module with username/password
yes i know, only test 1 thing every test...
"""
characters = self.demo_api.list()
for character in characters:
self.demo_api.reset(character)
# set A and B
self.demo_api.set('A', 5)
self.demo_api.set('B', 7)
# check value of A
check = self.demo_api.get('A')
assert check == 5
# check list
check = self.demo_api.list()
assert len(check) == 2
assert 'A' in check
assert 'B' in check
def test_token(self) -> None:
"""Test module with token."""
self.demo_api = DemoApi(None, None, 'secret', 'http://localhost:5041/')
# set A and B
self.demo_api.set('A', 5)
# check value of A
check = self.demo_api.get('A')
assert check == 5
if __name__ == '__main_':
unittest.main()Combine it all in one Ansible Module
Now we can copy the class in the Ansible module and connect them together.
You can use the module to ask if the user is in check mode (https://docs.ansible.com/ansible/latest/playbook_guide/playbooks_checkmode.html). That is in the property `module.check_mode`. if that is true, do not make changes or don’t support it (supports_check_mode=False) in the initialization of the AnsibleModule.
This is the finished module:
"""Ansible module that interact with demo api."""
from urllib.parse import urljoin
from typing import List
import json
import re
import requests
from ansible.module_utils.basic import AnsibleModule
class DemoApi:
"""
A simple demo class where the API logic is written
:param username: user that connect to API
:param password: password from the user
:param token: token can be user instead of username/password
:param uri: the endpoint of the API
:raises HTTPError: if one occurred
"""
def __init__(self, username: str, password: str, token: str, uri: str):
self.uri = uri
self.session = requests.session()
self.__connect(username, password, token)
def __connect(self, username: str, password: str, token: str):
if token:
self.session.headers.update({'X-Auth-Token': token})
else:
response = self.session.post(urljoin(self.uri, "token"), json={
"username": username, "password": password})
response.raise_for_status()
self.session.headers.update({'X-Auth-Token': response.text})
def reset(self, character: str) -> None:
"""
Reset will remove character from the set of characters that are set
:param character: character to reset
:raises HTTPError: if one occurred
"""
response = self.session.delete(
urljoin(self.uri, f"character/{character}"))
response.raise_for_status()
def set(self, character: str, number: int) -> None:
"""
Set the number on a character
:param character: character to set
:param number: the number that will be given to the character
:raises HTTPError: if one occurred
"""
response = self.session.put(
urljoin(self.uri, f"character/{character}?number={number}"))
response.raise_for_status()
def update(self, character: str, number: int) -> None:
"""
Update the number on a character
:param character: character to update
:param number: the number that will be given to the character
:raises HTTPError: if one occurred
"""
response = self.session.post(
urljoin(self.uri, f"character/{character}?number={number}"))
response.raise_for_status()
def get(self, character: str) -> int:
"""
Get the number that is set on a character
:param character: character where you want the number from
:returns: the number that will be given to the character
:raises HTTPError: if one occurred
"""
response = self.session.get(
urljoin(self.uri, f"character/{character}"))
response.raise_for_status()
return json.loads(response.text)
def list(self) -> List[str]:
"""
Get the list of characters that are set
:returns: the list of characters that have a number
:raises HTTPError: if one occurred
"""
response = self.session.get(urljoin(self.uri, "character"))
response.raise_for_status()
return json.loads(response.text)
DOCUMENTATION = r'''
---
module: api_demo
author:
- Bas Magré (@opvolger)
short_description: The ability to create, remove and manage a list of characters that have numbers
version_added: 0.0.1
description: "The ability to create, remove and manage a list of characters that have numbers. This is just a demo!"
options:
endpoint:
description: The uri of the API
type: str
required: true
sample: 'http://localhost:5041/'
username:
description: Username to get a token
type: str
required: false
sample: 'user'
password:
description: Password to get a token
type: str
required: false
sample: 'password'
token:
description: Use a token direct (without username/password)
type: str
required: false
sample: 'secret'
character:
description: Character (key)
type: str
required: false
sample: 'A'
number:
description: Number (value)
type: int
required: false
sample: 5
action:
description: The action to perform
type: str
required: true
default: get
choices: [ get, set, clear ]
sample: 'set'
'''
EXAMPLES = r'''
- name: Clear all set character and numbers with a token
api_demo:
endpoint: http://localhost:5041/
token: secret
action: clear
delegate_to: localhost
- name: Set number on a character with username and password
api_demo:
endpoint: http://localhost:5041/
username: user
password: password
character: "A"
number: 4
action: set
delegate_to: localhost
- name: Get number from set character with a token
api_demo:
endpoint: http://localhost:5041/
token: secret
character: "A"
action: get
delegate_to: localhost
'''
RETURN = r'''
exists:
description: if the character is set
returned: success
type: bool
sample: True
number:
description: The number that is set or get
type: int
sample: 5
'''
def run_module() -> None:
"""The Ansible module."""
# define the available arguments/parameters that a user can pass to the module
module_args = {
'endpoint': {'type': 'str', 'required': True},
'username': {'type': 'str', 'required': False},
'password': {'type': 'str', 'required': False, 'no_log': True},
'token': {'type': 'str', 'required': False, 'no_log': True},
'character': {'type': 'str', 'required': False},
'number': {'type': 'int', 'required': False},
'action': {'type': 'str', 'required': True, 'choices': ['get', 'set', 'clear']}
}
# use username with password
check_required_together = [
('username', 'password')
]
# use username/password or token is needed
check_required_one_of = [('username', 'token')]
# use username/password or token, only one
check_mutually_exclusive = [('username', 'token')]
# if action == get, we need the character argument
# if action == set, we need the character and number arguments
check_required_if = [
('action', 'get', ['character']),
('action', 'set', ['character', 'number'])
]
# the AnsibleModule object will be our abstraction working with Ansible
# this includes instantiation, a couple of common attr would be the
# args/params passed to the execution, as well as if the module
# supports check mode
module = AnsibleModule(
argument_spec=module_args,
supports_check_mode=True,
required_if=check_required_if,
required_together=check_required_together,
required_one_of=check_required_one_of,
mutually_exclusive=check_mutually_exclusive
)
# seed the result dict in the object
# we primarily care about changed and state
# change is if this module effectively modified the target
# state will include any data that you want your module to pass back
# for consumption, for example, in a subsequent task
result = {
'changed': False,
'rc': 1,
'diff': None
}
endpoint = module.params['endpoint']
username = module.params['username']
password = module.params['password']
token = module.params['token']
character = module.params['character']
number = module.params['number']
action = module.params['action']
# input checks
if character is not None and not re.fullmatch(r"[A-Z]", character):
module.fail_json(
msg=f'character: "{character}" must be an alpha letter and in upper case', **result)
if number is not None and not 1 <= number <= 255:
module.fail_json(msg='number must be between 1 and 255', **result)
demo_api = DemoApi(username, password, token, endpoint)
# actions
if action == 'get':
# only get from API that is in the list
character_list = demo_api.list()
if character in character_list:
result['number'] = demo_api.get(character)
result['exists'] = True
else:
result['exists'] = False
elif action == 'set':
character_list = demo_api.list()
if character in character_list:
current_number = demo_api.get(character)
if current_number != number:
# if the user is working with this module in only check mode,
# we do not want to make any changes to the environment.
if not module.check_mode:
demo_api.update(character, number)
result['changed'] = True
result['diff'] = {'before': {
'character': character,
'number': current_number
},
'after': {
'character': character,
'number': number
}
}
else:
# if the user is working with this module in only check mode,
# we do not want to make any changes to the environment.
if not module.check_mode:
demo_api.set(character, number)
result['changed'] = True
result['diff'] = {'before': {
'character': None,
'number': None
},
'after': {
'character': character,
'number': number
}
}
result['number'] = number
result['exists'] = True
elif action == 'clear':
character_list = demo_api.list()
for character_clear in character_list:
# if the user is working with this module in only check mode
# we do not want to make any changes to the environment.
if not module.check_mode:
demo_api.reset(character_clear)
result['changed'] = True
result['exists'] = False
result['diff'] = {'before': {
'character_list': character_list
},
'after': {
'character_list': None
}
}
result['rc'] = 0 # we are at the end, no errors occurred
module.exit_json(**result)
def main() -> None:
"""Main function to run Ansible Module."""
run_module()
if __name__ == '__main__':
main()Our work is done. We have a working Ansible module!
More information on GitHub
You can find code examples and further explanations on how to debug these Ansible modules in our GitHub repo: https://github.com/Babelvis/ansible-library-demo
Comments are closed