Plugins¶

from ansible.module_utils._text import to_native

try:
    cause_an_exception()
except Exception as e:
    raise AnsibleError('Something happened, this was original exception: %s' % to_native(e))

from ansible.module_utils._text import to_text
result_string = to_text(result_string)

DOCUMENTATION = '''
    callback: test
    type: stdout
    short_description: test callback
    version_added: historical
    description:
        - test callback description
    options:
      remote_addr:
        description:
            - Address
        default: inventory_hostname
        vars:
            - name: ansible_host
        type: str
'''

from ansible.plugins.action import ActionBase

class ActionModule(ActionBase):
    pass

module_return = self._execute_module(module_name='<NAME_OF_MODULE>',
                                     module_args=module_args,
                                     task_vars=task_vars, tmp=tmp)

#!/usr/bin/python
# Make coding more python3-ish, this is required for contributions to Ansible
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type

from ansible.plugins.action import ActionBase
from datetime import datetime


class ActionModule(ActionBase):
    def run(self, tmp=None, task_vars=None):
        super(ActionModule, self).run(tmp, task_vars)
        module_args = self._task.args.copy()
        module_return = self._execute_module(module_name='setup',
                                             module_args=module_args,
                                             task_vars=task_vars, tmp=tmp)
        ret = dict()
        remote_date = None
        if not module_return.get('failed'):
            for key, value in module_return['ansible_facts'].items():
                if key == 'ansible_date_time':
                    remote_date = value['iso8601']

        if remote_date:
            remote_date_obj = datetime.strptime(remote_date, '%Y-%m-%dT%H:%M:%SZ')
            time_delta = datetime.now() - remote_date_obj
            ret['delta_seconds'] = time_delta.seconds
            ret['delta_days'] = time_delta.days
            ret['delta_microseconds'] = time_delta.microseconds

        return dict(ansible_facts=dict(ret))

from ansible.plugins.loader import cache_loader
[...]
plugin = cache_loader.get('custom_cache', **cache_kwargs)

from ansible.plugins.cache import BaseCacheModule

class CacheModule(BaseCacheModule):
    def __init__(self, *args, **kwargs):
        super(CacheModule, self).__init__(*args, **kwargs)
        self._connection = self.get_option('_uri')
        self._prefix = self.get_option('_prefix')
        self._timeout = self.get_option('_timeout')

ansible-doc -l -t cache

from ansible.plugins.callback import CallbackBase

class CallbackModule(CallbackBase):
    pass

def v2_on_any(self,*args,**kwargs):

def v2_runner_on_failed(self,result,ignore_errors=False):
def v2_runner_on_ok(self,result):
def v2_runner_on_skipped(self,result):
def v2_runner_on_unreachable(self,result):
def v2_runner_on_async_poll(self,result):
def v2_runner_on_async_ok(self,result):
def v2_runner_on_async_failed(self,result):

def v2_playbook_on_start(self,playbook):
def v2_playbook_on_notify(self,handler,host):
def v2_playbook_on_no_hosts_matched(self):
def v2_playbook_on_no_hosts_remaining(self):
def v2_playbook_on_task_start(self,task,is_conditional):
def v2_playbook_on_cleanup_task_start(self,task):
def v2_playbook_on_handler_task_start(self,task):
def v2_playbook_on_vars_prompt(self,varname,private=True,prompt=None,encrypt=None,confirm=False,salt_size=None,salt=None,def ault=None,unsafe=None):
def v2_playbook_on_vars_prompt(self,varname,private=True,prompt=None,encrypt=None,confirm=False,salt_size=None,salt=None,def ault=None,unsafe=None):
self.playbook_on_vars_prompt(varname,private,prompt,encrypt,confirm,salt_size,salt,def ault,unsafe)
def v2_playbook_on_import_for_host(self,result,imported_file):
def v2_playbook_on_not_import_for_host(self,result,missing_file):
def v2_playbook_on_play_start(self,play):
def v2_playbook_on_stats(self,stats):

def v2_on_file_diff(self,result):
def v2_playbook_on_include(self,included_file):
def v2_runner_item_on_ok(self,result):
def v2_runner_item_on_failed(self,result):
def v2_runner_item_on_skipped(self,result):
def v2_runner_retry(self,result):
def v2_runner_on_start(self,host,task):

print(result._task) # 输出任务名
print(result._host) # 输出主机名
print(result._result) # 输出任务执行的结果
print(result.is_changed) # 是否更改
print(result.is_failed)  # 是否失败

# Make coding more python3-ish, this is required for contributions to Ansible
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type

# not only visible to ansible-doc, it also 'declares' the options the plugin requires and how to configure them.
DOCUMENTATION = '''
  callback: timer
  callback_type: aggregate
  requirements:
    - whitelist in configuration
  short_description: Adds time to play stats
  version_added: "2.0"
  description:
      - This callback just adds total play duration to the play stats.
  options:
    format_string:
      description: format of the string shown to user at play end
      ini:
        - section: callback_timer
          key: format_string
      env:
        - name: ANSIBLE_CALLBACK_TIMER_FORMAT
      default: "Playbook run took %s days, %s hours, %s minutes, %s seconds"
'''
from datetime import datetime

from ansible.plugins.callback import CallbackBase


class CallbackModule(CallbackBase):
    """
    This callback module tells you how long your plays ran for.
    """
    CALLBACK_VERSION = 2.0
    CALLBACK_TYPE = 'aggregate'
    CALLBACK_NAME = 'namespace.collection_name.timer'

    # only needed if you ship it and don't want to enable by default
    CALLBACK_NEEDS_WHITELIST = True

    def __init__(self):

        # make sure the expected objects are present, calling the base's __init__
        super(CallbackModule, self).__init__()

        # start the timer when the plugin is loaded, the first play should start a few milliseconds after.
        self.start_time = datetime.now()

    def _days_hours_minutes_seconds(self, runtime):
        ''' internal helper method for this callback '''
        minutes = (runtime.seconds // 60) % 60
        r_seconds = runtime.seconds - (minutes * 60)
        return runtime.days, runtime.seconds // 3600, minutes, r_seconds

    # this is only event we care about for display, when the play shows its summary stats; the rest are ignored by the base class
    def v2_playbook_on_stats(self, stats):
        end_time = datetime.now()
        runtime = end_time - self.start_time

        # Shows the usage of a config option declared in the DOCUMENTATION variable. Ansible will have set it when it loads the plugin.
        # Also note the use of the display object to print to screen. This is available to all callbacks, and you should use this over printing yourself
        self._display.display(self._plugin_options['format_string'] % (self._days_hours_minutes_seconds(runtime)))

ansible-doc -l -t callback

ansible-doc -l -t connection

ansible-doc -l -t inventory

# python 3 headers, required if submitting to Ansible
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type

DOCUMENTATION = """
        lookup: file
        author: Daniel Hokka Zakrisson <daniel@hozac.com>
        version_added: "0.9"
        short_description: read file contents
        description:
            - This lookup returns the contents from a file on the Ansible controller's file system.
        options:
          _terms:
            description: path(s) of files to read
            required: True
        notes:
          - if read in variable context, the file can be interpreted as YAML if the content is valid to the parser.
          - this lookup does not understand globing --- use the fileglob lookup instead.
"""
from ansible.errors import AnsibleError, AnsibleParserError
from ansible.plugins.lookup import LookupBase
from ansible.utils.display import Display

display = Display()


class LookupModule(LookupBase):

    def run(self, terms, variables=None, **kwargs):


        # lookups in general are expected to both take a list as input and output a list
        # this is done so they work with the looping construct 'with_'.
        ret = []
        for term in terms:
            display.debug("File lookup term: %s" % term)

            # Find the file in the expected search path, using a class method
            # that implements the 'expected' search path for Ansible plugins.
            lookupfile = self.find_file_in_search_path(variables, 'files', term)

            # Don't use print or your own logging, the display class
            # takes care of it in a unified way.
            display.vvvv(u"File lookup using %s as file" % lookupfile)
            try:
                if lookupfile:
                    contents, show_data = self._loader._get_file_contents(lookupfile)
                    ret.append(contents.rstrip())
                else:
                    # Always use ansible error classes to throw 'final' exceptions,
                    # so the Ansible engine will know how to deal with them.
                    # The Parser error indicates invalid options passed
                    raise AnsibleParserError()
            except AnsibleParserError:
                raise AnsibleError("could not locate file in lookup: %s" % term)

        return ret

---
- hosts: all
  vars:
     contents: "{{ lookup('namespace.collection_name.file', '/etc/foo.txt') }}"

  tasks:

     - debug:
         msg: the value of foo.txt is {{ contents }} as seen today {{ lookup('pipe', 'date +"%Y-%m-%d"') }}

ansible-doc -l -t lookup

ansible-doc -l -t shell

ansible-doc -l -t strategy

ansible-doc -l -t vars