Custom Module Development

Overview

Learn how to create custom Ansible modules when existing modules don't meet your needs.

Basic Module Structure

Simple Module Example

#!/usr/bin/python

from ansible.module_utils.basic import AnsibleModule

def run_module():
    # Define module arguments
    module_args = dict(
        name=dict(type='str', required=True),
        state=dict(type='str', default='present', choices=['present', 'absent']),
        properties=dict(type='dict', default=dict())
    )

    # Create module instance
    module = AnsibleModule(
        argument_spec=module_args,
        supports_check_mode=True
    )

    # Initialize result dictionary
    result = dict(
        changed=False,
        original_message='',
        message=''
    )

    # Module logic here
    if module.params['state'] == 'present':
        result['changed'] = True
        result['message'] = f"Created {module.params['name']}"

    # Return results
    module.exit_json(**result)

def main():
    run_module()

if __name__ == '__main__':
    main()

Advanced Features

Complex Module Example

#!/usr/bin/python

from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils.common.dict_transformations import dict_merge
from ansible.module_utils._text import to_native
import traceback

class CustomError(Exception):
    pass

def setup_resource(module):
    name = module.params['name']
    properties = module.params['properties']

    try:
        # Resource setup logic here
        return True, {"status": "created", "name": name}
    except CustomError as e:
        module.fail_json(msg=f"Error setting up resource: {to_native(e)}")
    except Exception as e:
        module.fail_json(msg=f"Unexpected error: {to_native(e)}", 
                        exception=traceback.format_exc())

def run_module():
    module_args = dict(
        name=dict(type='str', required=True),
        state=dict(
            type='str', 
            default='present', 
            choices=['present', 'absent']
        ),
        properties=dict(
            type='dict',
            default=dict(),
            options=dict(
                timeout=dict(type='int', default=30),
                retries=dict(type='int', default=3),
                custom_option=dict(type='str')
            )
        ),
        validate=dict(type='bool', default=True)
    )

    module = AnsibleModule(
        argument_spec=module_args,
        supports_check_mode=True,
        required_if=[
            ('state', 'present', ['properties'])
        ]
    )

    result = dict(
        changed=False,
        original_message='',
        message='',
        resource={}
    )

    if module.check_mode:
        module.exit_json(**result)

    if module.params['state'] == 'present':
        changed, resource = setup_resource(module)
        result['changed'] = changed
        result['resource'] = resource

    module.exit_json(**result)

if __name__ == '__main__':
    main()

Module with API Integration

#!/usr/bin/python

import requests
from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils.urls import fetch_url

class APIClient:
    def __init__(self, module):
        self.module = module
        self.base_url = module.params['api_url']
        self.token = module.params['api_token']

    def make_request(self, method, endpoint, data=None):
        headers = {
            'Authorization': f'Bearer {self.token}',
            'Content-Type': 'application/json'
        }

        url = f"{self.base_url}/{endpoint}"

        response, info = fetch_url(
            self.module,
            url,
            method=method,
            data=data,
            headers=headers
        )

        if info['status'] >= 400:
            self.module.fail_json(
                msg=f"API request failed: {info['msg']}",
                status_code=info['status']
            )

        return response.read()

def run_module():
    module_args = dict(
        api_url=dict(type='str', required=True),
        api_token=dict(type='str', required=True, no_log=True),
        resource_id=dict(type='str', required=True),
        action=dict(
            type='str',
            choices=['get', 'create', 'update', 'delete'],
            required=True
        ),
        data=dict(type='dict')
    )

    module = AnsibleModule(
        argument_spec=module_args,
        supports_check_mode=True
    )

    client = APIClient(module)
    # Module logic here

Testing Modules

Unit Testing

#!/usr/bin/python
# test_custom_module.py

import unittest
from unittest.mock import patch
from ansible.module_utils import basic
from ansible.module_utils.common.text.converters import to_bytes
import json

def set_module_args(args):
    args = json.dumps({'ANSIBLE_MODULE_ARGS': args})
    basic._ANSIBLE_ARGS = to_bytes(args)

class TestCustomModule(unittest.TestCase):
    def setUp(self):
        self.mock_module = patch.multiple(basic.AnsibleModule,
                                        exit_json=exit_json,
                                        fail_json=fail_json)
        self.mock_module.start()

    def tearDown(self):
        self.mock_module.stop()

    def test_module_fail_when_required_args_missing(self):
        with self.assertRaises(AnsibleFailJson) as result:
            set_module_args({})
            custom_module.main()

    def test_module_success(self):
        set_module_args({
            'name': 'test_resource',
            'state': 'present'
        })
        with self.assertRaises(AnsibleExitJson) as result:
            custom_module.main()

        self.assertTrue(result.exception.args[0]['changed'])

Integration Testing

# integration_tests/test_custom_module.yml
- hosts: localhost
  gather_facts: no
  tasks:
    - name: Test custom module
      custom_module:
        name: test_resource
        state: present
      register: result

    - name: Verify result
      assert:
        that:
          - result is changed
          - result.resource.name == 'test_resource'

Documentation

Module Documentation

#!/usr/bin/python
# Copyright: (c) 2024, Your Name <[email protected]>
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)

DOCUMENTATION = r'''
---
module: custom_module

short_description: Manage custom resources

version_added: "1.0.0"

description:
    - Create, update, or delete custom resources
    - Manages resource lifecycle and properties

options:
    name:
        description: Resource name
        required: true
        type: str
    state:
        description: Desired state of the resource
        choices: [ present, absent ]
        default: present
        type: str
    properties:
        description: Resource properties
        type: dict
        default: {}

author:
    - Your Name (@yourgithub)
'''

EXAMPLES = r'''
# Create a resource
- name: Create new resource
  custom_module:
    name: myresource
    state: present
    properties:
      key1: value1
      key2: value2

# Delete a resource
- name: Remove resource
  custom_module:
    name: myresource
    state: absent
'''

RETURN = r'''
resource:
    description: Resource information
    type: dict
    returned: always
    sample: {
        "name": "myresource",
        "status": "created",
        "properties": {
            "key1": "value1"
        }
    }
changed:
    description: Whether the resource was changed
    type: bool
    returned: always
'''

Best Practices

Error Handling

try:
    result = do_something()
except Exception as e:
    module.fail_json(
        msg=f"Operation failed: {to_native(e)}",
        exception=traceback.format_exc()
    )

Input Validation

def validate_input(module):
    name = module.params['name']
    if len(name) < 3:
        module.fail_json(
            msg="Name must be at least 3 characters"
        )

Return Values

result = {
    'changed': True,
    'resource': {
        'id': 'resource_id',
        'name': module.params['name'],
        'properties': module.params['properties']
    },
    'warnings': [],
    'diff': {
        'before': before_state,
        'after': after_state
    }
}