TaskLDAP

class tasks.task_ldap.TaskLDAP

Call a LDAP service.

Inputs

Name	Type	Default	Description
encoding	`str`	`utf-8`	The encoding of the return values of the LDAP server
host	`str`		The LDAP server to connect to
paged_search	`bool`	`False`	Perform a paged search until the enpoint is exhausted
password	`str`		The password used to bind to the LDAP server
port	`int`	`None`	The port number to use. If unset port 389 for ldap:// and port 636 for ldaps:// will be used
search	`dict`		Perform a LDAP search. Pass a dictionary containing the parameters of the search. e.g. {'search_base': 'o=test', 'search_filter': '(objectclass=)', 'attributes': ''}
use_ssl	`bool`	`True`	Will use ldap:// if set to `False`, ldaps:// otherwise
user	`str`		The username used to bind to the LDAP server

Outputs

Name	Type	Description
log	`list`
result	`object`
execution_id	`int`	The ID of the task execution
message	`str`	The ended message for the task. If the task ended with an error, the message will contain information about what went wrong
status	`str`	The ended status for the task. Either "success" or "error".

Constants

input_list = ['encoding', 'host', 'paged_search', 'password', 'port', 'search', 'use_ssl', 'user']output_list = ['log', 'result']version = 1

Methods

run ()

Example

import flow_api
def handler(system: flow_api.System, this: flow_api.Execution):
    this.task(
        'LDAP',
        host='my-ldap-server',
        user='cn=user,dc=example,dc=com',
        password='my-secret-password',
        search={
            'search_base': 'dc=example,dc=com',
            'search_filter': '(objectclass=*)',
            'attributes': '*',
        },
    )
    return this.success('all done')

TaskK8S

TaskPS