signature:
def modify ( self , pk = None , create_on_missing = False , ** kwargs ) :
    """Modify an existing notification template .
 Not all required configuration - related fields ( required according to
 notification _ type ) should be provided .
 Fields in the resource ' s ` identity ` tuple can be used in lieu of a
 primary key for a lookup ; in such a case , only other fields are
 written .
 To modify unique fields , you must use the primary key for the lookup .
 = = = = = API DOCS = = = = =
 Modify an already existing object .
 : param pk : Primary key of the resource to be modified .
 : type pk : int
 : param create _ on _ missing : Flag that if set , a new object is created if ` ` pk ` ` is not set and objects
 matching the appropriate unique criteria is not found .
 : type create _ on _ missing : bool
 : param ` * * kwargs ` : Keyword arguments which , all together , will be used as PATCH body to modify the
 resource object . if ` ` pk ` ` is not set , key - value pairs of ` ` * * kwargs ` ` which are
 also in resource ' s identity will be used to lookup existing reosource .
 : returns : A dictionary combining the JSON output of the modified resource , as well as two extra fields :
 " changed " , a flag indicating if the resource is successfully updated ; " id " , an integer which
 is the primary key of the updated object .
 : rtype : dict
 = = = = = API DOCS = = = = ="""

implementation:
# Create the resource if needed .
if pk is None and create_on_missing :
    try :
        self . get ( ** copy . deepcopy ( kwargs ) )
    except exc . NotFound :
        return self . create ( ** kwargs )
# Modify everything except notification type and configuration
config_item = self . _separate ( kwargs )
notification_type = kwargs . pop ( 'notification_type' , None )
debug . log ( 'Modify everything except notification type and' ' configuration' , header = 'details' )
part_result = super ( Resource , self ) . modify ( pk = pk , create_on_missing = create_on_missing , ** kwargs )
# Modify notification type and configuration
if notification_type is None or notification_type == part_result [ 'notification_type' ] :
    for item in part_result [ 'notification_configuration' ] :
        if item not in config_item or not config_item [ item ] :
            to_add = part_result [ 'notification_configuration' ] [ item ]
            if not ( to_add == '$encrypted$' and item in Resource . encrypted_fields ) :
                config_item [ item ] = to_add
if notification_type is None :
    kwargs [ 'notification_type' ] = part_result [ 'notification_type' ]
else :
    kwargs [ 'notification_type' ] = notification_type
self . _configuration ( kwargs , config_item )
debug . log ( 'Modify notification type and configuration' , header = 'details' )
result = super ( Resource , self ) . modify ( pk = pk , create_on_missing = create_on_missing , ** kwargs )
# Update ' changed ' field to give general changed info
if 'changed' in result and 'changed' in part_result :
    result [ 'changed' ] = result [ 'changed' ] or part_result [ 'changed' ]
return result


signature:
def delete ( self , pk = None , fail_on_missing = False , ** kwargs ) :
    """Remove the given notification template .
 Note here configuration - related fields like
 ' notification _ configuration ' and ' channels ' will not be
 used even provided .
 If ` fail _ on _ missing ` is True , then the object ' s not being found is
 considered a failure ; otherwise , a success with no change is reported .
 = = = = = API DOCS = = = = =
 Remove the given object .
 : param pk : Primary key of the resource to be deleted .
 : type pk : int
 : param fail _ on _ missing : Flag that if set , the object ' s not being found is considered a failure ; otherwise ,
 a success with no change is reported .
 : type fail _ on _ missing : bool
 : param ` * * kwargs ` : Keyword arguments used to look up resource object to delete if ` ` pk ` ` is not provided .
 : returns : dictionary of only one field " changed " , which is a flag indicating whether the specified resource
 is successfully deleted .
 : rtype : dict
 = = = = = API DOCS = = = = ="""

implementation:
self . _separate ( kwargs )
return super ( Resource , self ) . delete ( pk = pk , fail_on_missing = fail_on_missing , ** kwargs )


signature:
def list ( self , all_pages = False , ** kwargs ) :
    """Return a list of notification templates .
 Note here configuration - related fields like
 ' notification _ configuration ' and ' channels ' will not be
 used even provided .
 If one or more filters are provided through keyword arguments ,
 filter the results accordingly .
 If no filters are provided , return all results .
 = = = = = API DOCS = = = = =
 Retrieve a list of objects .
 : param all _ pages : Flag that if set , collect all pages of content from the API when returning results .
 : type all _ pages : bool
 : param page : The page to show . Ignored if all _ pages is set .
 : type page : int
 : param query : Contains 2 - tuples used as query parameters to filter resulting resource objects .
 : type query : list
 : param ` * * kwargs ` : Keyword arguments list of available fields used for searching resource objects .
 : returns : A JSON object containing details of all resource objects returned by Tower backend .
 : rtype : dict
 = = = = = API DOCS = = = = ="""

implementation:
self . _separate ( kwargs )
return super ( Resource , self ) . list ( all_pages = all_pages , ** kwargs )


signature:
def get ( self , pk = None , ** kwargs ) :
    """Return one and exactly one notification template .
 Note here configuration - related fields like
 ' notification _ configuration ' and ' channels ' will not be
 used even provided .
 Lookups may be through a primary key , specified as a positional
 argument , and / or through filters specified through keyword arguments .
 If the number of results does not equal one , raise an exception .
 = = = = = API DOCS = = = = =
 Retrieve one and exactly one object .
 : param pk : Primary key of the resource to be read . Tower CLI will only attempt to read * that * object
 if ` ` pk ` ` is provided ( not ` ` None ` ` ) .
 : type pk : int
 : param ` * * kwargs ` : Keyword arguments used to look up resource object to retrieve if ` ` pk ` ` is not provided .
 : returns : loaded JSON of the retrieved resource object .
 : rtype : dict
 = = = = = API DOCS = = = = ="""

implementation:
self . _separate ( kwargs )
return super ( Resource , self ) . get ( pk = pk , ** kwargs )


signature:
def config_from_environment ( ) :
    """Read tower - cli config values from the environment if present , being
 careful not to override config values that were explicitly passed in ."""

implementation:
kwargs = { }
for k in CONFIG_OPTIONS :
    env = 'TOWER_' + k . upper ( )
    v = os . getenv ( env , None )
    if v is not None :
        kwargs [ k ] = v
return kwargs


signature:
def with_global_options ( method ) :
    """Apply the global options that we desire on every method within
 tower - cli to the given click command ."""

implementation:
# Create global options for the Tower host , username , and password .
# These are runtime options that will override the configuration file
# settings .
method = click . option ( '-h' , '--tower-host' , help = 'The location of the Ansible Tower host. ' 'HTTPS is assumed as the protocol unless "http://" is explicitly ' 'provided. This will take precedence over a host provided to ' '`tower config`, if any.' , required = False , callback = _apply_runtime_setting , is_eager = True , expose_value = False ) ( method )
method = click . option ( '-t' , '--tower-oauth-token' , help = 'OAuth2 token to use to authenticate to Ansible Tower. ' 'This will take precedence over a token provided to ' '`tower config`, if any.' , required = False , callback = _apply_runtime_setting , is_eager = True , expose_value = False ) ( method )
method = click . option ( '-u' , '--tower-username' , help = 'Username to use to authenticate to Ansible Tower. ' 'This will take precedence over a username provided to ' '`tower config`, if any.' , required = False , callback = _apply_runtime_setting , is_eager = True , expose_value = False ) ( method )
method = click . option ( '-p' , '--tower-password' , help = 'Password to use to authenticate to Ansible Tower. ' 'This will take precedence over a password provided to ' '`tower config`, if any. If value is ASK you will be ' 'prompted for the password' , required = False , callback = _apply_runtime_setting , is_eager = True , expose_value = False ) ( method )
# Create a global verbose / debug option .
method = click . option ( '-f' , '--format' , help = 'Output format. The "human" format is intended for humans ' 'reading output on the CLI; the "json" and "yaml" formats ' 'provide more data, and "id" echos the object id only.' , type = click . Choice ( [ 'human' , 'json' , 'yaml' , 'id' ] ) , required = False , callback = _apply_runtime_setting , is_eager = True , expose_value = False ) ( method )
method = click . option ( '-v' , '--verbose' , default = None , help = 'Show information about requests being made.' , is_flag = True , required = False , callback = _apply_runtime_setting , is_eager = True , expose_value = False ) ( method )
method = click . option ( '--description-on' , default = None , help = 'Show description in human-formatted output.' , is_flag = True , required = False , callback = _apply_runtime_setting , is_eager = True , expose_value = False ) ( method )
# Create a global SSL warning option .
method = click . option ( '--insecure' , default = None , help = 'Turn off insecure connection warnings. Set config verify_ssl ' 'to make this permanent.' , is_flag = True , required = False , callback = _apply_runtime_setting , is_eager = True , expose_value = False ) ( method )
# Create a custom certificate specification option .
method = click . option ( '--certificate' , default = None , help = 'Path to a custom certificate file that will be used throughout' ' the command. Overwritten by --insecure flag if set.' , required = False , callback = _apply_runtime_setting , is_eager = True , expose_value = False ) ( method )
method = click . option ( '--use-token' , default = None , help = 'Turn on Tower\'s token-based authentication. No longer supported ' 'in Tower 3.3 and above.' , is_flag = True , required = False , callback = _apply_runtime_setting , is_eager = True , expose_value = False ) ( method )
# Manage the runtime settings context
method = runtime_context_manager ( method )
# Okay , we ' re done adding options ; return the method .
return method


signature:
def pop_option ( function , name ) :
    """Used to remove an option applied by the @ click . option decorator .
 This is useful for when you want to subclass a decorated resource command
 and * don ' t * want all of the options provided by the parent class '
 implementation ."""

implementation:
for option in getattr ( function , '__click_params__' , tuple ( ) ) :
    if option . name == name :
        function . __click_params__ . remove ( option )


signature:
def _read ( self , fp , fpname ) :
    """Read the configuration from the given file .
 If the file lacks any section header , add a [ general ] section
 header that encompasses the whole thing ."""

implementation:
# Attempt to read the file using the superclass implementation .
# Check the permissions of the file we are considering reading
# if the file exists and the permissions expose it to reads from
# other users , raise a warning
if os . path . isfile ( fpname ) :
    file_permission = os . stat ( fpname )
    if fpname != os . path . join ( tower_dir , 'tower_cli.cfg' ) and ( ( file_permission . st_mode & stat . S_IRGRP ) or ( file_permission . st_mode & stat . S_IROTH ) ) :
        warnings . warn ( 'File {0} readable by group or others.' . format ( fpname ) , RuntimeWarning )
# If it doesn ' t work because there ' s no section header , then
# create a section header and call the superclass implementation
# again .
try :
    return configparser . ConfigParser . _read ( self , fp , fpname )
except configparser . MissingSectionHeaderError :
    fp . seek ( 0 )
    string = '[general]\n%s' % fp . read ( )
    flo = StringIO ( string )
    # flo = = file - like object
    return configparser . ConfigParser . _read ( self , flo , fpname )


signature:
def set_or_reset_runtime_param ( self , key , value ) :
    """Maintains the context of the runtime settings for invoking
 a command .
 This should be called by a click . option callback , and only
 called once for each setting for each command invocation .
 If the setting exists , it follows that the runtime settings are
 stale , so the entire runtime settings are reset ."""

implementation:
if self . _runtime . has_option ( 'general' , key ) :
    self . _runtime = self . _new_parser ( )
if value is None :
    return
settings . _runtime . set ( 'general' , key . replace ( 'tower_' , '' ) , six . text_type ( value ) )


signature:
def runtime_values ( self , ** kwargs ) :
    """= = = = = API DOCS = = = = =
 Context manager that temporarily override runtime level configurations .
 : param kwargs : Keyword arguments specifying runtime configuration settings .
 : type kwargs : arbitrary keyword arguments
 : returns : N / A
 : Example :
 > > > import tower _ cli
 > > > from tower _ cli . conf import settings
 > > > with settings . runtime _ values ( username = ' user ' , password = ' pass ' ) :
 > > > print ( tower _ cli . get _ resource ( ' credential ' ) . list ( ) )
 = = = = = API DOCS = = = = ="""

implementation:
# Coerce all values to strings ( to be coerced back by configparser
# later ) and defenestrate any None values .
for k , v in copy . copy ( kwargs ) . items ( ) : # If the value is None , just get rid of it .
    if v is None :
        kwargs . pop ( k )
        continue
    # Remove these keys from the cache , if they are present .
    self . _cache . pop ( k , None )
    # Coerce values to strings .
    kwargs [ k ] = six . text_type ( v )
# Replace the ` self . _ runtime ` INI parser with a new one , using
# the context manager ' s kwargs as the " defaults " ( there can never
# be anything other than defaults , but that isn ' t a problem for our
# purposes because we ' re using our own precedence system ) .
# Ensure that everything is put back to rights at the end of the
# context manager call .
old_runtime_parser = self . _runtime
try :
    self . _runtime = Parser ( defaults = kwargs )
    self . _runtime . add_section ( 'general' )
    yield self
finally : # Revert the runtime configparser object .
    self . _runtime = old_runtime_parser
    # Remove the keys from the cache again , since the settings
    # have been reverted .
    for key in kwargs :
        self . _cache . pop ( k , None )


signature:
def launch ( self , job_template = None , monitor = False , wait = False , timeout = None , no_input = True , extra_vars = None , ** kwargs ) :
    """Launch a new job based on a job template .
 Creates a new job in Ansible Tower , immediately starts it , and
 returns back an ID in order for its status to be monitored .
 = = = = = API DOCS = = = = =
 Launch a new job based on a job template .
 : param job _ template : Primary key or name of the job template to launch new job .
 : type job _ template : str
 : param monitor : Flag that if set , immediately calls ` ` monitor ` ` on the newly launched job rather
 than exiting with a success .
 : type monitor : bool
 : param wait : Flag that if set , monitor the status of the job , but do not print while job is in progress .
 : type wait : bool
 : param timeout : If provided with ` ` monitor ` ` flag set , this attempt will time out after the given number
 of seconds .
 : type timeout : int
 : param no _ input : Flag that if set , suppress any requests for input .
 : type no _ input : bool
 : param extra _ vars : yaml formatted texts that contains extra variables to pass on .
 : type extra _ vars : array of strings
 : param diff _ mode : Specify diff mode for job template to run .
 : type diff _ mode : bool
 : param limit : Specify host limit for job template to run .
 : type limit : str
 : param tags : Specify tagged actions in the playbook to run .
 : type tags : str
 : param skip _ tags : Specify tagged actions in the playbook to omit .
 : type skip _ tags : str
 : param job _ type : Specify job type for job template to run .
 : type job _ type : str
 : param verbosity : Specify verbosity of the playbook run .
 : type verbosity : int
 : param inventory : Specify machine credential for job template to run .
 : type inventory : str
 : param credential : Specify machine credential for job template to run .
 : type credential : str
 : returns : Result of subsequent ` ` monitor ` ` call if ` ` monitor ` ` flag is on ; Result of subsequent
 ` ` wait ` ` call if ` ` wait ` ` flag is on ; Result of subsequent ` ` status ` ` call if none of
 the two flags are on .
 : rtype : dict
 = = = = = API DOCS = = = = ="""

implementation:
# Get the job template from Ansible Tower .
# This is used as the baseline for starting the job .
tags = kwargs . get ( 'tags' , None )
jt_resource = get_resource ( 'job_template' )
jt = jt_resource . get ( job_template )
# Update the job data by adding an automatically - generated job name ,
# and removing the ID .
data = { }
if tags :
    data [ 'job_tags' ] = tags
# Initialize an extra _ vars list that starts with the job template
# preferences first , if they exist
extra_vars_list = [ ]
if 'extra_vars' in data and len ( data [ 'extra_vars' ] ) > 0 : # But only do this for versions before 2.3
    debug . log ( 'Getting version of Tower.' , header = 'details' )
    r = client . get ( '/config/' )
    if LooseVersion ( r . json ( ) [ 'version' ] ) < LooseVersion ( '2.4' ) :
        extra_vars_list = [ data [ 'extra_vars' ] ]
# Add the runtime extra _ vars to this list
if extra_vars :
    extra_vars_list += list ( extra_vars )
# accept tuples
# If the job template requires prompting for extra variables ,
# do so ( unless - - no - input is set ) .
if jt . get ( 'ask_variables_on_launch' , False ) and not no_input and not extra_vars : # If JT extra _ vars are JSON , echo them to user as YAML
    initial = parser . process_extra_vars ( [ jt [ 'extra_vars' ] ] , force_json = False )
    initial = '\n' . join ( ( '# Specify extra variables (if any) here as YAML.' , '# Lines beginning with "#" denote comments.' , initial , ) )
    extra_vars = click . edit ( initial ) or ''
    if extra_vars != initial :
        extra_vars_list = [ extra_vars ]
# Data is starting out with JT variables , and we only want to
# include extra _ vars that come from the algorithm here .
data . pop ( 'extra_vars' , None )
# Replace / populate data fields if prompted .
modified = set ( )
for resource in PROMPT_LIST :
    if jt . pop ( 'ask_' + resource + '_on_launch' , False ) and not no_input :
        resource_object = kwargs . get ( resource , None )
        if type ( resource_object ) == types . Related :
            resource_class = get_resource ( resource )
            resource_object = resource_class . get ( resource ) . pop ( 'id' , None )
        if resource_object is None :
            debug . log ( '{0} is asked at launch but not provided' . format ( resource ) , header = 'warning' )
        elif resource != 'tags' :
            data [ resource ] = resource_object
            modified . add ( resource )
# Dump extra _ vars into JSON string for launching job
if len ( extra_vars_list ) > 0 :
    data [ 'extra_vars' ] = parser . process_extra_vars ( extra_vars_list , force_json = True )
# Create the new job in Ansible Tower .
start_data = { }
endpoint = '/job_templates/%d/launch/' % jt [ 'id' ]
if 'extra_vars' in data and len ( data [ 'extra_vars' ] ) > 0 :
    start_data [ 'extra_vars' ] = data [ 'extra_vars' ]
if tags :
    start_data [ 'job_tags' ] = data [ 'job_tags' ]
for resource in PROMPT_LIST :
    if resource in modified :
        start_data [ resource ] = data [ resource ]
# There ' s a non - trivial chance that we are going to need some
# additional information to start the job ; in particular , many jobs
# rely on passwords entered at run - time .
# If there are any such passwords on this job , ask for them now .
debug . log ( 'Asking for information necessary to start the job.' , header = 'details' )
job_start_info = client . get ( endpoint ) . json ( )
for password in job_start_info . get ( 'passwords_needed_to_start' , [ ] ) :
    start_data [ password ] = getpass ( 'Password for %s: ' % password )
# Actually start the job .
debug . log ( 'Launching the job.' , header = 'details' )
self . _pop_none ( kwargs )
kwargs . update ( start_data )
job_started = client . post ( endpoint , data = kwargs )
# Get the job ID from the result .
job_id = job_started . json ( ) [ 'id' ]
# If returning json indicates any ignored fields , display it in
# verbose mode .
if job_started . text == '' :
    ignored_fields = { }
else :
    ignored_fields = job_started . json ( ) . get ( 'ignored_fields' , { } )
has_ignored_fields = False
for key , value in ignored_fields . items ( ) :
    if value and value != '{}' :
        if not has_ignored_fields :
            debug . log ( 'List of ignored fields on the server side:' , header = 'detail' )
            has_ignored_fields = True
        debug . log ( '{0}: {1}' . format ( key , value ) )
# Get some information about the running job to print
result = self . status ( pk = job_id , detail = True )
result [ 'changed' ] = True
# If we were told to monitor the job once it started , then call
# monitor from here .
if monitor :
    return self . monitor ( job_id , timeout = timeout )
elif wait :
    return self . wait ( job_id , timeout = timeout )
return result


signature:
def launch ( self , monitor = False , wait = False , timeout = None , ** kwargs ) :
    """Launch a new ad - hoc command .
 Runs a user - defined command from Ansible Tower , immediately starts it ,
 and returns back an ID in order for its status to be monitored .
 = = = = = API DOCS = = = = =
 Launch a new ad - hoc command .
 : param monitor : Flag that if set , immediately calls ` ` monitor ` ` on the newly launched command rather
 than exiting with a success .
 : type monitor : bool
 : param wait : Flag that if set , monitor the status of the job , but do not print while job is in progress .
 : type wait : bool
 : param timeout : If provided with ` ` monitor ` ` flag set , this attempt will time out after the given number
 of seconds .
 : type timeout : int
 : param ` * * kwargs ` : Fields needed to create and launch an ad hoc command .
 : returns : Result of subsequent ` ` monitor ` ` call if ` ` monitor ` ` flag is on ; Result of subsequent ` ` wait ` `
 call if ` ` wait ` ` flag is on ; dictionary of " id " and " changed " if none of the two flags are on .
 : rtype : dict
 : raises tower _ cli . exceptions . TowerCLIError : When ad hoc commands are not available in Tower backend .
 = = = = = API DOCS = = = = ="""

implementation:
# This feature only exists for versions 2.2 and up
r = client . get ( '/' )
if 'ad_hoc_commands' not in r . json ( ) :
    raise exc . TowerCLIError ( 'Your host is running an outdated version' 'of Ansible Tower that can not run ' 'ad-hoc commands (2.2 or earlier)' )
# Pop the None arguments because we have no . write ( ) method in
# inheritance chain for this type of resource . This is needed
self . _pop_none ( kwargs )
# Actually start the command .
debug . log ( 'Launching the ad-hoc command.' , header = 'details' )
result = client . post ( self . endpoint , data = kwargs )
command = result . json ( )
command_id = command [ 'id' ]
# If we were told to monitor the command once it started , then call
# monitor from here .
if monitor :
    return self . monitor ( command_id , timeout = timeout )
elif wait :
    return self . wait ( command_id , timeout = timeout )
# Return the command ID and other response data
answer = OrderedDict ( ( ( 'changed' , True ) , ( 'id' , command_id ) , ) )
answer . update ( result . json ( ) )
return answer


signature:
def _auto_help_text ( self , help_text ) :
    """Given a method with a docstring , convert the docstring
 to more CLI appropriate wording , and also disambiguate the
 word " object " on the base class docstrings ."""

implementation:
# Delete API docs if there are any .
api_doc_delimiter = '=====API DOCS====='
begin_api_doc = help_text . find ( api_doc_delimiter )
if begin_api_doc >= 0 :
    end_api_doc = help_text . rfind ( api_doc_delimiter ) + len ( api_doc_delimiter )
    help_text = help_text [ : begin_api_doc ] + help_text [ end_api_doc : ]
# Convert the word " object " to the appropriate type of
# object being modified ( e . g . user , organization ) .
an_prefix = ( 'a' , 'e' , 'i' , 'o' )
if not self . resource_name . lower ( ) . startswith ( an_prefix ) :
    help_text = help_text . replace ( 'an object' , 'a %s' % self . resource_name )
if self . resource_name . lower ( ) . endswith ( 'y' ) :
    help_text = help_text . replace ( 'objects' , '%sies' % self . resource_name [ : - 1 ] , )
help_text = help_text . replace ( 'object' , self . resource_name )
# Convert some common Python terms to their CLI equivalents .
help_text = help_text . replace ( 'keyword argument' , 'option' )
help_text = help_text . replace ( 'raise an exception' , 'abort with an error' )
# Convert keyword arguments specified in docstrings enclosed
# by backticks to switches .
for match in re . findall ( r'`([\w_]+)`' , help_text ) :
    option = '--%s' % match . replace ( '_' , '-' )
    help_text = help_text . replace ( '`%s`' % match , option )
# Done ; return the new help text .
return help_text


signature:
def _echo_method ( self , method ) :
    """Given a method , return a method that runs the internal
 method and echos the result ."""

implementation:
@ functools . wraps ( method )
def func ( * args , ** kwargs ) : # Echo warning if this method is deprecated .
    if getattr ( method , 'deprecated' , False ) :
        debug . log ( 'This method is deprecated in Tower 3.0.' , header = 'warning' )
    result = method ( * args , ** kwargs )
    # If this was a request that could result in a modification
    # of data , print it in Ansible coloring .
    color_info = { }
    if isinstance ( result , dict ) and 'changed' in result :
        if result [ 'changed' ] :
            color_info [ 'fg' ] = 'yellow'
        else :
            color_info [ 'fg' ] = 'green'
    # Piece together the result into the proper format .
    format = getattr ( self , '_format_%s' % ( getattr ( method , 'format_freezer' , None ) or settings . format ) )
    output = format ( result )
    # Perform the echo .
    secho ( output , ** color_info )
return func


signature:
def _format_yaml ( self , payload ) :
    """Convert the payload into a YAML string with proper
 indentation and return it ."""

implementation: return parser . ordered_dump ( payload , Dumper = yaml . SafeDumper , default_flow_style = False )


signature:
def _format_id ( self , payload ) :
    """Echos only the id"""

implementation:
if 'id' in payload :
    return str ( payload [ 'id' ] )
if 'results' in payload :
    return ' ' . join ( [ six . text_type ( item [ 'id' ] ) for item in payload [ 'results' ] ] )
raise MultipleRelatedError ( 'Could not serialize output with id format.' )


signature:
def _format_human ( self , payload ) :
    """Convert the payload into an ASCII table suitable for
 printing on screen and return it ."""

implementation:
page = None
total_pages = None
# What are the columns we will show ?
columns = [ field . name for field in self . resource . fields if field . display or settings . description_on and field . name == 'description' ]
columns . insert ( 0 , 'id' )
# Save a dictionary - by - name of fields for later use
fields_by_name = { }
for field in self . resource . fields :
    fields_by_name [ field . name ] = field
# Sanity check : If there is a " changed " key in our payload
# and little else , we print a short message and not a table .
# this specifically applies to deletion
if 'changed' in payload and 'id' not in payload :
    return 'OK. (changed: {0})' . format ( six . text_type ( payload [ 'changed' ] ) . lower ( ) , )
# Sanity check : If there is no ID and no results , then this
# is unusual output ; keep our table formatting , but plow
# over the columns - as - keys stuff above .
# this originally applied to launch / status / update methods
# but it may become deprecated
if 'id' not in payload and 'results' not in payload :
    columns = [ i for i in payload . keys ( ) ]
# Get our raw rows into a standard format .
if 'results' in payload :
    raw_rows = payload [ 'results' ]
    if payload . get ( 'count' , 0 ) > len ( payload [ 'results' ] ) :
        prev = payload . get ( 'previous' , 0 ) or 0
        page = prev + 1
        count = payload [ 'count' ]
        if payload . get ( 'next' , None ) :
            total_pages = math . ceil ( count / len ( raw_rows ) )
        else :
            total_pages = page
else :
    raw_rows = [ payload ]
# If we have no rows to display , return this information
# and don ' t do any further processing .
if not raw_rows :
    return 'No records found.'
# Determine the width for each column .
widths = { }
for col in columns :
    widths [ col ] = max ( len ( col ) , * [ len ( self . get_print_value ( i , col ) ) for i in raw_rows ] )
    fd = fields_by_name . get ( col , None )
    if fd is not None and fd . col_width is not None :
        widths [ col ] = fd . col_width
# It ' s possible that the column widths will exceed our terminal
# width ; if so , reduce column widths accordingly .
# TODO : Write this .
# Put together the divider row .
# This is easy and straightforward : it ' s simply a table divider
# using the widths calculated .
divider_row = ''
for col in columns :
    divider_row += '=' * widths [ col ] + ' '
divider_row . rstrip ( )
# Put together the header row .
# This is also easy and straightforward ; simply center the
# headers ( which str . format does for us ! ) .
header_row = ''
for col in columns :
    header_row += ( '{0:^%d}' % widths [ col ] ) . format ( col ) + ' '
header_row . rstrip ( )
# Piece together each row of data .
data_rows = [ ]
for raw_row in raw_rows :
    data_row = ''
    for col in columns :
        template = six . text_type ( '{0:%d}' ) % widths [ col ]
        value = self . get_print_value ( raw_row , col )
        # Right - align certain native data types
        if isinstance ( raw_row . get ( col , 'N/A' ) , ( bool , int ) ) :
            template = template . replace ( '{0:' , '{0:>' )
        # Truncate the cell entry if exceeds manually
        # specified column width limit
        fd = fields_by_name . get ( col , None )
        if fd is not None and fd . col_width is not None :
            str_value = template . format ( value or '' )
            if len ( str_value ) > fd . col_width :
                value = str_value [ : fd . col_width ]
        data_row += template . format ( value or '' ) + ' '
    data_rows . append ( data_row . rstrip ( ) )
# Result the resulting table .
response = '\n' . join ( ( divider_row , header_row , divider_row , '\n' . join ( data_rows ) , divider_row , ) )
# Don ' t print page numbers for 1 page results
if page and total_pages != 1 :
    response += '(Page %d of %d.)' % ( page , total_pages )
if payload . get ( 'changed' , False ) :
    response = 'Resource changed.\n' + response
return response


signature:
def get_command ( self , ctx , name ) :
    """Retrieve the appropriate method from the Resource ,
 decorate it as a click command , and return that method ."""

implementation:
# Sanity check : Does a method exist corresponding to this
# command ? If not , None is returned for click to raise
# exception .
if not hasattr ( self . resource , name ) :
    return None
# Get the method .
method = getattr ( self . resource , name )
# Get any attributes that were given at command - declaration
# time .
attrs = getattr ( method , '_cli_command_attrs' , { } )
# If the help message comes from the docstring , then
# convert it into a message specifically for this resource .
help_text = inspect . getdoc ( method )
attrs [ 'help' ] = self . _auto_help_text ( help_text or '' )
# On some methods , we ignore the defaults , which are intended
# for writing and not reading ; process this .
ignore_defaults = attrs . pop ( 'ignore_defaults' , False )
# Wrap the method , such that it outputs its final return
# value rather than returning it .
new_method = self . _echo_method ( method )
# Soft copy the " _ _ click _ params _ _ " , if any exist .
# This is the internal holding method that the click library
# uses to store @ click . option and @ click . argument directives
# before the method is converted into a command .
# Because self . _ echo _ method uses @ functools . wraps , this is
# actually preserved ; the purpose of copying it over is
# so we can get our resource fields at the top of the help ;
# the easiest way to do this is to load them in before the
# conversion takes place . ( This is a happy result of Armin ' s
# work to get around Python ' s processing decorators
# bottom - to - top . )
click_params = getattr ( method , '__click_params__' , [ ] )
new_method . __click_params__ = copy ( click_params )
new_method = with_global_options ( new_method )
# Write options based on the fields available on this resource .
fao = attrs . pop ( 'use_fields_as_options' , True )
if fao :
    for field in reversed ( self . resource . fields ) :
        if not field . is_option :
            continue
        # If we got an iterable rather than a boolean ,
        # then it is a list of fields to use ; check for
        # presence in that list .
        if not isinstance ( fao , bool ) and field . name not in fao :
            continue
        # Create the initial arguments based on the
        # option value . If we have a different key to use
        # ( which is what gets routed to the Tower API ) ,
        # ensure that is the first argument .
        args = [ field . option ]
        if field . key :
            args . insert ( 0 , field . key )
        # short name aliases for common flags
        short_fields = { 'name' : 'n' , 'description' : 'd' , 'inventory' : 'i' , 'extra_vars' : 'e' }
        if field . name in short_fields :
            args . append ( '-' + short_fields [ field . name ] )
        # Apply the option to the method .
        option_help = field . help
        if isinstance ( field . type , StructuredInput ) :
            option_help += ' Use @ to get JSON or YAML from a file.'
        if field . required :
            option_help = '[REQUIRED] ' + option_help
        elif field . read_only :
            option_help = '[READ ONLY] ' + option_help
        option_help = '[FIELD]' + option_help
        click . option ( * args , default = field . default if not ignore_defaults else None , help = option_help , type = field . type , show_default = field . show_default , multiple = field . multiple , is_eager = False ) ( new_method )
# Make a click Command instance using this method
# as the callback , and return it .
cmd = click . command ( name = name , cls = ActionSubcommand , ** attrs ) ( new_method )
# If this method has a ` pk ` positional argument ,
# then add a click argument for it .
code = six . get_function_code ( method )
if 'pk' in code . co_varnames :
    click . argument ( 'pk' , nargs = 1 , required = False , type = str , metavar = '[ID]' ) ( cmd )
# Done ; return the command .
return cmd


signature:
def update ( self , inventory_source , monitor = False , wait = False , timeout = None , ** kwargs ) :
    """Update the given inventory source .
 = = = = = API DOCS = = = = =
 Update the given inventory source .
 : param inventory _ source : Primary key or name of the inventory source to be updated .
 : type inventory _ source : str
 : param monitor : Flag that if set , immediately calls ` ` monitor ` ` on the newly launched inventory update
 rather than exiting with a success .
 : type monitor : bool
 : param wait : Flag that if set , monitor the status of the inventory update , but do not print while it is
 in progress .
 : type wait : bool
 : param timeout : If provided with ` ` monitor ` ` flag set , this attempt will time out after the given number
 of seconds .
 : type timeout : int
 : param ` * * kwargs ` : Fields used to override underlyingl inventory source fields when creating and launching
 an inventory update .
 : returns : Result of subsequent ` ` monitor ` ` call if ` ` monitor ` ` flag is on ; Result of subsequent ` ` wait ` `
 call if ` ` wait ` ` flag is on ; dictionary of " status " if none of the two flags are on .
 : rtype : dict
 : raises tower _ cli . exceptions . BadRequest : When the inventory source cannot be updated .
 = = = = = API DOCS = = = = ="""

implementation:
# Establish that we are able to update this inventory source
# at all .
debug . log ( 'Asking whether the inventory source can be updated.' , header = 'details' )
r = client . get ( '%s%d/update/' % ( self . endpoint , inventory_source ) )
if not r . json ( ) [ 'can_update' ] :
    raise exc . BadRequest ( 'Tower says it cannot run an update against this inventory source.' )
# Run the update .
debug . log ( 'Updating the inventory source.' , header = 'details' )
r = client . post ( '%s%d/update/' % ( self . endpoint , inventory_source ) , data = { } )
inventory_update_id = r . json ( ) [ 'inventory_update' ]
# If we were told to monitor the project update ' s status , do so .
if monitor or wait :
    if monitor :
        result = self . monitor ( inventory_update_id , parent_pk = inventory_source , timeout = timeout )
    elif wait :
        result = self . wait ( inventory_update_id , parent_pk = inventory_source , timeout = timeout )
    inventory = client . get ( '/inventory_sources/%d/' % result [ 'inventory_source' ] ) . json ( ) [ 'inventory' ]
    result [ 'inventory' ] = int ( inventory )
    return result
# Done .
return { 'id' : inventory_update_id , 'status' : 'ok' }


signature:
def list ( self , group = None , host_filter = None , ** kwargs ) :
    """Return a list of hosts .
 = = = = = API DOCS = = = = =
 Retrieve a list of hosts .
 : param group : Primary key or name of the group whose hosts will be listed .
 : type group : str
 : param all _ pages : Flag that if set , collect all pages of content from the API when returning results .
 : type all _ pages : bool
 : param page : The page to show . Ignored if all _ pages is set .
 : type page : int
 : param query : Contains 2 - tuples used as query parameters to filter resulting resource objects .
 : type query : list
 : param ` * * kwargs ` : Keyword arguments list of available fields used for searching resource objects .
 : returns : A JSON object containing details of all resource objects returned by Tower backend .
 : rtype : dict
 = = = = = API DOCS = = = = ="""

implementation:
if group :
    kwargs [ 'query' ] = kwargs . get ( 'query' , ( ) ) + ( ( 'groups__in' , group ) , )
if host_filter :
    kwargs [ 'query' ] = kwargs . get ( 'query' , ( ) ) + ( ( 'host_filter' , host_filter ) , )
return super ( Resource , self ) . list ( ** kwargs )


signature:
def list_facts ( self , pk = None , ** kwargs ) :
    """Return a JSON object of all available facts of the given host .
 Note global option - - format is not available here , as the output would always be JSON - formatted .
 = = = = = API DOCS = = = = =
 List all available facts of the given host .
 : param pk : Primary key of the target host .
 : type pk : int
 : param ` * * kwargs ` : Keyword arguments list of available fields used for searching resource objects .
 : returns : A JSON object of all available facts of the given host .
 : rtype : dict
 = = = = = API DOCS = = = = ="""

implementation:
res = self . get ( pk = pk , ** kwargs )
url = self . endpoint + '%d/%s/' % ( res [ 'id' ] , 'ansible_facts' )
return client . get ( url , params = { } ) . json ( )


signature:
def format_command_subsection ( self , ctx , formatter , commands , header ) :
    """Writes help text for a sub - section of commands ,
 specifically to be reused for resource commands
 and system / configuration commands ."""

implementation:
rows = [ ]
for subcommand in commands :
    cmd = self . get_command ( ctx , subcommand )
    # What is this , the tool lied about a command . Ignore it
    if cmd is None :
        continue
    help = cmd . short_help or ''
    rows . append ( ( subcommand , help ) )
if rows :
    with formatter . section ( header ) :
        formatter . write_dl ( rows )


signature:
def format_commands ( self , ctx , formatter ) :
    """Extra format methods for multi methods that adds all the commands
 after the options ."""

implementation:
self . format_command_subsection ( ctx , formatter , self . list_misc_commands ( ) , 'Commands' )
self . format_command_subsection ( ctx , formatter , self . list_resource_commands ( ) , 'Resources' )


signature:
def list_commands ( self , ctx ) :
    """Return a list of commands present in the commands and resources
 folders , but not subcommands ."""

implementation:
commands = set ( self . list_resource_commands ( ) )
commands . union ( set ( self . list_misc_commands ( ) ) )
return sorted ( commands )


signature:
def list_resource_commands ( self ) :
    """Returns a list of multi - commands for each resource type ."""

implementation:
resource_path = os . path . abspath ( os . path . join ( os . path . dirname ( __file__ ) , os . pardir , 'resources' ) )
answer = set ( [ ] )
for _ , name , _ in pkgutil . iter_modules ( [ resource_path ] ) :
    res = tower_cli . get_resource ( name )
    if not getattr ( res , 'internal' , False ) :
        answer . add ( name )
return sorted ( answer )


signature:
def list_misc_commands ( self ) :
    """Returns a list of global commands , realted to CLI
 configuration or system management in general ."""

implementation:
answer = set ( [ ] )
for cmd_name in misc . __all__ :
    answer . add ( cmd_name )
return sorted ( answer )


signature:
def get_command ( self , ctx , name ) :
    """Given a command identified by its name , import the appropriate
 module and return the decorated command .
 Resources are automatically commands , but if both a resource and
 a command are defined , the command takes precedence ."""

implementation:
# First , attempt to get a basic command from ` tower _ cli . api . misc ` .
if name in misc . __all__ :
    return getattr ( misc , name )
# No command was found ; try to get a resource .
try :
    resource = tower_cli . get_resource ( name )
    return ResSubcommand ( resource )
except ImportError :
    pass
# Okay , we weren ' t able to find a command .
secho ( 'No such command: %s.' % name , fg = 'red' , bold = True )
sys . exit ( 2 )


signature:
def command ( method = None , ** kwargs ) :
    """Mark this method as a CLI command .
 This will only have any meaningful effect in methods that are members of a
 Resource subclass ."""

implementation:
# Create the actual decorator to be applied .
# This is done in such a way to make ` @ resources . command ` ,
# ` @ resources . command ( ) ` , and ` @ resources . command ( foo = ' bar ' ) ` all work .
def actual_decorator ( method ) :
    method . _cli_command = True
    method . _cli_command_attrs = kwargs
    return method
# If we got the method straight - up , apply the decorator and return
# the decorated method ; otherwise , return the actual decorator for
# the Python interpreter to apply .
if method and isinstance ( method , types . FunctionType ) :
    return actual_decorator ( method )
else :
    return actual_decorator


signature:
def unified_job_template_options ( method ) :
    """Adds the decorators for all types of unified job templates ,
 and if the non - unified type is specified , converts it into the
 unified _ job _ template kwarg ."""

implementation:
jt_dec = click . option ( '--job-template' , type = types . Related ( 'job_template' ) , help = 'Use this job template as unified_job_template field' )
prj_dec = click . option ( '--project' , type = types . Related ( 'project' ) , help = 'Use this project as unified_job_template field' )
inv_src_dec = click . option ( '--inventory-source' , type = types . Related ( 'inventory_source' ) , help = 'Use this inventory source as unified_job_template field' )
def ujt_translation ( _method ) :
    def _ujt_translation ( * args , ** kwargs ) :
        for fd in [ 'job_template' , 'project' , 'inventory_source' ] :
            if fd in kwargs and kwargs [ fd ] is not None :
                kwargs [ 'unified_job_template' ] = kwargs . pop ( fd )
        return _method ( * args , ** kwargs )
    return functools . wraps ( _method ) ( _ujt_translation )
return ujt_translation ( inv_src_dec ( prj_dec ( jt_dec ( method ) ) ) )


signature:
def new_state ( self ) :
    """Generates a state string to be used in authorizations ."""

implementation:
try :
    self . _state = self . state ( )
    log . debug ( "Generated new state %s." , self . _state )
except TypeError :
    self . _state = self . state
    log . debug ( "Re-using previously supplied state %s." , self . _state )
return self . _state


signature:
def authorization_url ( self , url , state = None , ** kwargs ) :
    """Form an authorization URL .
 : param url : Authorization endpoint url , must be HTTPS .
 : param state : An optional state string for CSRF protection . If not
 given it will be generated for you .
 : param kwargs : Extra parameters to include .
 : return : authorization _ url , state"""

implementation:
state = state or self . new_state ( )
return ( self . _client . prepare_request_uri ( url , redirect_uri = self . redirect_uri , scope = self . scope , state = state , ** kwargs ) , state , )


signature:
def fetch_token ( self , token_url , code = None , authorization_response = None , body = "" , auth = None , username = None , password = None , method = "POST" , force_querystring = False , timeout = None , headers = None , verify = True , proxies = None , include_client_id = None , client_secret = None , ** kwargs ) :
    """Generic method for fetching an access token from the token endpoint .
 If you are using the MobileApplicationClient you will want to use
 ` token _ from _ fragment ` instead of ` fetch _ token ` .
 The current implementation enforces the RFC guidelines .
 : param token _ url : Token endpoint URL , must use HTTPS .
 : param code : Authorization code ( used by WebApplicationClients ) .
 : param authorization _ response : Authorization response URL , the callback
 URL of the request back to you . Used by
 WebApplicationClients instead of code .
 : param body : Optional application / x - www - form - urlencoded body to add the
 include in the token request . Prefer kwargs over body .
 : param auth : An auth tuple or method as accepted by ` requests ` .
 : param username : Username required by LegacyApplicationClients to appear
 in the request body .
 : param password : Password required by LegacyApplicationClients to appear
 in the request body .
 : param method : The HTTP method used to make the request . Defaults
 to POST , but may also be GET . Other methods should
 be added as needed .
 : param force _ querystring : If True , force the request body to be sent
 in the querystring instead .
 : param timeout : Timeout of the request in seconds .
 : param headers : Dict to default request headers with .
 : param verify : Verify SSL certificate .
 : param proxies : The ` proxies ` argument is passed onto ` requests ` .
 : param include _ client _ id : Should the request body include the
 ` client _ id ` parameter . Default is ` None ` ,
 which will attempt to autodetect . This can be
 forced to always include ( True ) or never
 include ( False ) .
 : param client _ secret : The ` client _ secret ` paired to the ` client _ id ` .
 This is generally required unless provided in the
 ` auth ` tuple . If the value is ` None ` , it will be
 omitted from the request , however if the value is
 an empty string , an empty string will be sent .
 : param kwargs : Extra parameters to include in the token request .
 : return : A token dict"""

implementation:
if not is_secure_transport ( token_url ) :
    raise InsecureTransportError ( )
if not code and authorization_response :
    self . _client . parse_request_uri_response ( authorization_response , state = self . _state )
    code = self . _client . code
elif not code and isinstance ( self . _client , WebApplicationClient ) :
    code = self . _client . code
    if not code :
        raise ValueError ( "Please supply either code or " "authorization_response parameters." )
# Earlier versions of this library build an HTTPBasicAuth header out of
# ` username ` and ` password ` . The RFC states , however these attributes
# must be in the request body and not the header .
# If an upstream server is not spec compliant and requires them to
# appear as an Authorization header , supply an explicit ` auth ` header
# to this function .
# This check will allow for empty strings , but not ` None ` .
# Refernences
# 4.3.2 - Resource Owner Password Credentials Grant
# https : / / tools . ietf . org / html / rfc6749 # section - 4.3.2
if isinstance ( self . _client , LegacyApplicationClient ) :
    if username is None :
        raise ValueError ( "`LegacyApplicationClient` requires both the " "`username` and `password` parameters." )
    if password is None :
        raise ValueError ( "The required paramter `username` was supplied, " "but `password` was not." )
# merge username and password into kwargs for ` prepare _ request _ body `
if username is not None :
    kwargs [ "username" ] = username
if password is not None :
    kwargs [ "password" ] = password
# is an auth explicitly supplied ?
if auth is not None : # if we ' re dealing with the default of ` include _ client _ id ` ( None ) :
# we will assume the ` auth ` argument is for an RFC compliant server
# and we should not send the ` client _ id ` in the body .
# This approach allows us to still force the client _ id by submitting
# ` include _ client _ id = True ` along with an ` auth ` object .
    if include_client_id is None :
        include_client_id = False
# otherwise we may need to create an auth header
else : # since we don ' t have an auth header , we MAY need to create one
# it is possible that we want to send the ` client _ id ` in the body
# if so , ` include _ client _ id ` should be set to True
# otherwise , we will generate an auth header
    if include_client_id is not True :
        client_id = self . client_id
        if client_id :
            log . debug ( 'Encoding `client_id` "%s" with `client_secret` ' "as Basic auth credentials." , client_id , )
            client_secret = client_secret if client_secret is not None else ""
            auth = requests . auth . HTTPBasicAuth ( client_id , client_secret )
if include_client_id : # this was pulled out of the params
# it needs to be passed into prepare _ request _ body
    if client_secret is not None :
        kwargs [ "client_secret" ] = client_secret
body = self . _client . prepare_request_body ( code = code , body = body , redirect_uri = self . redirect_uri , include_client_id = include_client_id , ** kwargs )
headers = headers or { "Accept" : "application/json" , "Content-Type" : "application/x-www-form-urlencoded;charset=UTF-8" , }
self . token = { }
request_kwargs = { }
if method . upper ( ) == "POST" :
    request_kwargs [ "params" if force_querystring else "data" ] = dict ( urldecode ( body ) )
elif method . upper ( ) == "GET" :
    request_kwargs [ "params" ] = dict ( urldecode ( body ) )
else :
    raise ValueError ( "The method kwarg must be POST or GET." )
r = self . request ( method = method , url = token_url , timeout = timeout , headers = headers , auth = auth , verify = verify , proxies = proxies , ** request_kwargs )
log . debug ( "Request to fetch token completed with status %s." , r . status_code )
log . debug ( "Request url was %s" , r . request . url )
log . debug ( "Request headers were %s" , r . request . headers )
log . debug ( "Request body was %s" , r . request . body )
log . debug ( "Response headers were %s and content %s." , r . headers , r . text )
log . debug ( "Invoking %d token response hooks." , len ( self . compliance_hook [ "access_token_response" ] ) , )
for hook in self . compliance_hook [ "access_token_response" ] :
    log . debug ( "Invoking hook %s." , hook )
    r = hook ( r )
self . _client . parse_request_body_response ( r . text , scope = self . scope )
self . token = self . _client . token
log . debug ( "Obtained token %s." , self . token )
return self . token


signature:
def token_from_fragment ( self , authorization_response ) :
    """Parse token from the URI fragment , used by MobileApplicationClients .
 : param authorization _ response : The full URL of the redirect back to you
 : return : A token dict"""

implementation:
self . _client . parse_request_uri_response ( authorization_response , state = self . _state )
self . token = self . _client . token
return self . token


signature:
def refresh_token ( self , token_url , refresh_token = None , body = "" , auth = None , timeout = None , headers = None , verify = True , proxies = None , ** kwargs ) :
    """Fetch a new access token using a refresh token .
 : param token _ url : The token endpoint , must be HTTPS .
 : param refresh _ token : The refresh _ token to use .
 : param body : Optional application / x - www - form - urlencoded body to add the
 include in the token request . Prefer kwargs over body .
 : param auth : An auth tuple or method as accepted by ` requests ` .
 : param timeout : Timeout of the request in seconds .
 : param headers : A dict of headers to be used by ` requests ` .
 : param verify : Verify SSL certificate .
 : param proxies : The ` proxies ` argument will be passed to ` requests ` .
 : param kwargs : Extra parameters to include in the token request .
 : return : A token dict"""

implementation:
if not token_url :
    raise ValueError ( "No token endpoint set for auto_refresh." )
if not is_secure_transport ( token_url ) :
    raise InsecureTransportError ( )
refresh_token = refresh_token or self . token . get ( "refresh_token" )
log . debug ( "Adding auto refresh key word arguments %s." , self . auto_refresh_kwargs )
kwargs . update ( self . auto_refresh_kwargs )
body = self . _client . prepare_refresh_body ( body = body , refresh_token = refresh_token , scope = self . scope , ** kwargs )
log . debug ( "Prepared refresh token request body %s" , body )
if headers is None :
    headers = { "Accept" : "application/json" , "Content-Type" : ( "application/x-www-form-urlencoded;charset=UTF-8" ) , }
r = self . post ( token_url , data = dict ( urldecode ( body ) ) , auth = auth , timeout = timeout , headers = headers , verify = verify , withhold_token = True , proxies = proxies , )
log . debug ( "Request to refresh token completed with status %s." , r . status_code )
log . debug ( "Response headers were %s and content %s." , r . headers , r . text )
log . debug ( "Invoking %d token response hooks." , len ( self . compliance_hook [ "refresh_token_response" ] ) , )
for hook in self . compliance_hook [ "refresh_token_response" ] :
    log . debug ( "Invoking hook %s." , hook )
    r = hook ( r )
self . token = self . _client . parse_request_body_response ( r . text , scope = self . scope )
if not "refresh_token" in self . token :
    log . debug ( "No new refresh token given. Re-using old." )
    self . token [ "refresh_token" ] = refresh_token
return self . token


signature:
def request ( self , method , url , data = None , headers = None , withhold_token = False , client_id = None , client_secret = None , ** kwargs ) :
    """Intercept all requests and add the OAuth 2 token if present ."""

implementation:
if not is_secure_transport ( url ) :
    raise InsecureTransportError ( )
if self . token and not withhold_token :
    log . debug ( "Invoking %d protected resource request hooks." , len ( self . compliance_hook [ "protected_request" ] ) , )
    for hook in self . compliance_hook [ "protected_request" ] :
        log . debug ( "Invoking hook %s." , hook )
        url , headers , data = hook ( url , headers , data )
    log . debug ( "Adding token %s to request." , self . token )
    try :
        url , headers , data = self . _client . add_token ( url , http_method = method , body = data , headers = headers )
    # Attempt to retrieve and save new access token if expired
    except TokenExpiredError :
        if self . auto_refresh_url :
            log . debug ( "Auto refresh is set, attempting to refresh at %s." , self . auto_refresh_url , )
            # We mustn ' t pass auth twice .
            auth = kwargs . pop ( "auth" , None )
            if client_id and client_secret and ( auth is None ) :
                log . debug ( 'Encoding client_id "%s" with client_secret as Basic auth credentials.' , client_id , )
                auth = requests . auth . HTTPBasicAuth ( client_id , client_secret )
            token = self . refresh_token ( self . auto_refresh_url , auth = auth , ** kwargs )
            if self . token_updater :
                log . debug ( "Updating token to %s using %s." , token , self . token_updater )
                self . token_updater ( token )
                url , headers , data = self . _client . add_token ( url , http_method = method , body = data , headers = headers )
            else :
                raise TokenUpdated ( token )
        else :
            raise
log . debug ( "Requesting url %s using method %s." , url , method )
log . debug ( "Supplying headers %s and data %s" , headers , data )
log . debug ( "Passing through key word arguments %s." , kwargs )
return super ( OAuth2Session , self ) . request ( method , url , headers = headers , data = data , ** kwargs )


signature:
def register_compliance_hook ( self , hook_type , hook ) :
    """Register a hook for request / response tweaking .
 Available hooks are :
 access _ token _ response invoked before token parsing .
 refresh _ token _ response invoked before refresh token parsing .
 protected _ request invoked before making a request .
 If you find a new hook is needed please send a GitHub PR request
 or open an issue ."""

implementation:
if hook_type not in self . compliance_hook :
    raise ValueError ( "Hook type %s is not in %s." , hook_type , self . compliance_hook )
self . compliance_hook [ hook_type ] . add ( hook )


signature:
def authorized ( self ) :
    """Boolean that indicates whether this session has an OAuth token
 or not . If ` self . authorized ` is True , you can reasonably expect
 OAuth - protected requests to the resource to succeed . If
 ` self . authorized ` is False , you need the user to go through the OAuth
 authentication dance before OAuth - protected requests to the resource
 will succeed ."""

implementation:
if self . _client . client . signature_method == SIGNATURE_RSA : # RSA only uses resource _ owner _ key
    return bool ( self . _client . client . resource_owner_key )
else : # other methods of authentication use all three pieces
    return ( bool ( self . _client . client . client_secret ) and bool ( self . _client . client . resource_owner_key ) and bool ( self . _client . client . resource_owner_secret ) )


signature:
def authorization_url ( self , url , request_token = None , ** kwargs ) :
    """Create an authorization URL by appending request _ token and optional
 kwargs to url .
 This is the second step in the OAuth 1 workflow . The user should be
 redirected to this authorization URL , grant access to you , and then
 be redirected back to you . The redirection back can either be specified
 during client registration or by supplying a callback URI per request .
 : param url : The authorization endpoint URL .
 : param request _ token : The previously obtained request token .
 : param kwargs : Optional parameters to append to the URL .
 : returns : The authorization URL with new parameters embedded .
 An example using a registered default callback URI .
 > > > request _ token _ url = ' https : / / api . twitter . com / oauth / request _ token '
 > > > authorization _ url = ' https : / / api . twitter . com / oauth / authorize '
 > > > oauth _ session = OAuth1Session ( ' client - key ' , client _ secret = ' secret ' )
 > > > oauth _ session . fetch _ request _ token ( request _ token _ url )
 ' oauth _ token ' : ' sdf0o9823sjdfsdf ' ,
 ' oauth _ token _ secret ' : ' 2kjshdfp92i34asdasd ' ,
 > > > oauth _ session . authorization _ url ( authorization _ url )
 ' https : / / api . twitter . com / oauth / authorize ? oauth _ token = sdf0o9823sjdfsdf '
 > > > oauth _ session . authorization _ url ( authorization _ url , foo = ' bar ' )
 ' https : / / api . twitter . com / oauth / authorize ? oauth _ token = sdf0o9823sjdfsdf & foo = bar '
 An example using an explicit callback URI .
 > > > request _ token _ url = ' https : / / api . twitter . com / oauth / request _ token '
 > > > authorization _ url = ' https : / / api . twitter . com / oauth / authorize '
 > > > oauth _ session = OAuth1Session ( ' client - key ' , client _ secret = ' secret ' , callback _ uri = ' https : / / 127.0.0.1 / callback ' )
 > > > oauth _ session . fetch _ request _ token ( request _ token _ url )
 ' oauth _ token ' : ' sdf0o9823sjdfsdf ' ,
 ' oauth _ token _ secret ' : ' 2kjshdfp92i34asdasd ' ,
 > > > oauth _ session . authorization _ url ( authorization _ url )
 ' https : / / api . twitter . com / oauth / authorize ? oauth _ token = sdf0o9823sjdfsdf & oauth _ callback = https % 3A % 2F % 2F127.0.0.1%2Fcallback '"""

implementation:
kwargs [ "oauth_token" ] = request_token or self . _client . client . resource_owner_key
log . debug ( "Adding parameters %s to url %s" , kwargs , url )
return add_params_to_uri ( url , kwargs . items ( ) )


signature:
def fetch_request_token ( self , url , realm = None , ** request_kwargs ) :
    r"""Fetch a request token .
 This is the first step in the OAuth 1 workflow . A request token is
 obtained by making a signed post request to url . The token is then
 parsed from the application / x - www - form - urlencoded response and ready
 to be used to construct an authorization url .
 : param url : The request token endpoint URL .
 : param realm : A list of realms to request access to .
 : param \ * \ * request _ kwargs : Optional arguments passed to ' ' post ' '
 function in ' ' requests . Session ' '
 : returns : The response in dict format .
 Note that a previously set callback _ uri will be reset for your
 convenience , or else signature creation will be incorrect on
 consecutive requests .
 > > > request _ token _ url = ' https : / / api . twitter . com / oauth / request _ token '
 > > > oauth _ session = OAuth1Session ( ' client - key ' , client _ secret = ' secret ' )
 > > > oauth _ session . fetch _ request _ token ( request _ token _ url )
 ' oauth _ token ' : ' sdf0o9823sjdfsdf ' ,
 ' oauth _ token _ secret ' : ' 2kjshdfp92i34asdasd ' ,"""

implementation:
self . _client . client . realm = " " . join ( realm ) if realm else None
token = self . _fetch_token ( url , ** request_kwargs )
log . debug ( "Resetting callback_uri and realm (not needed in next phase)." )
self . _client . client . callback_uri = None
self . _client . client . realm = None
return token


signature:
def fetch_access_token ( self , url , verifier = None , ** request_kwargs ) :
    """Fetch an access token .
 This is the final step in the OAuth 1 workflow . An access token is
 obtained using all previously obtained credentials , including the
 verifier from the authorization step .
 Note that a previously set verifier will be reset for your
 convenience , or else signature creation will be incorrect on
 consecutive requests .
 > > > access _ token _ url = ' https : / / api . twitter . com / oauth / access _ token '
 > > > redirect _ response = ' https : / / 127.0.0.1 / callback ? oauth _ token = kjerht2309uf & oauth _ token _ secret = lsdajfh923874 & oauth _ verifier = w34o8967345'
 > > > oauth _ session = OAuth1Session ( ' client - key ' , client _ secret = ' secret ' )
 > > > oauth _ session . parse _ authorization _ response ( redirect _ response )
 ' oauth _ token : ' kjerht2309u ' ,
 ' oauth _ token _ secret : ' lsdajfh923874 ' ,
 ' oauth _ verifier : ' w34o8967345 ' ,
 > > > oauth _ session . fetch _ access _ token ( access _ token _ url )
 ' oauth _ token ' : ' sdf0o9823sjdfsdf ' ,
 ' oauth _ token _ secret ' : ' 2kjshdfp92i34asdasd ' ,"""

implementation:
if verifier :
    self . _client . client . verifier = verifier
if not getattr ( self . _client . client , "verifier" , None ) :
    raise VerifierMissing ( "No client verifier has been set." )
token = self . _fetch_token ( url , ** request_kwargs )
log . debug ( "Resetting verifier attribute, should not be used anymore." )
self . _client . client . verifier = None
return token


signature:
def parse_authorization_response ( self , url ) :
    """Extract parameters from the post authorization redirect response URL .
 : param url : The full URL that resulted from the user being redirected
 back from the OAuth provider to you , the client .
 : returns : A dict of parameters extracted from the URL .
 > > > redirect _ response = ' https : / / 127.0.0.1 / callback ? oauth _ token = kjerht2309uf & oauth _ token _ secret = lsdajfh923874 & oauth _ verifier = w34o8967345'
 > > > oauth _ session = OAuth1Session ( ' client - key ' , client _ secret = ' secret ' )
 > > > oauth _ session . parse _ authorization _ response ( redirect _ response )
 ' oauth _ token : ' kjerht2309u ' ,
 ' oauth _ token _ secret : ' lsdajfh923874 ' ,
 ' oauth _ verifier : ' w34o8967345 ' ,"""

implementation:
log . debug ( "Parsing token from query part of url %s" , url )
token = dict ( urldecode ( urlparse ( url ) . query ) )
log . debug ( "Updating internal client token attribute." )
self . _populate_attributes ( token )
self . token = token
return token


signature:
def rebuild_auth ( self , prepared_request , response ) :
    """When being redirected we should always strip Authorization
 header , since nonce may not be reused as per OAuth spec ."""

implementation:
if "Authorization" in prepared_request . headers : # If we get redirected to a new host , we should strip out
# any authentication headers .
    prepared_request . headers . pop ( "Authorization" , True )
    prepared_request . prepare_auth ( self . auth )
return


signature:
def get_fpath ( self , cfgstr = None ) :
    """Reports the filepath that the cacher will use .
 It will attempt to use ' { fname } _ { cfgstr } { ext } ' unless that is too long .
 Then cfgstr will be hashed .
 Example :
 > > > from ubelt . util _ cache import Cacher
 > > > import pytest
 > > > with pytest . warns ( UserWarning ) :
 > > > cacher = Cacher ( ' test _ cacher1 ' )
 > > > cacher . get _ fpath ( )
 > > > self = Cacher ( ' test _ cacher2 ' , cfgstr = ' cfg1 ' )
 > > > self . get _ fpath ( )
 > > > self = Cacher ( ' test _ cacher3 ' , cfgstr = ' cfg1 ' * 32)
 > > > self . get _ fpath ( )"""

implementation:
condensed = self . _condense_cfgstr ( cfgstr )
fname_cfgstr = '{}_{}{}' . format ( self . fname , condensed , self . ext )
fpath = join ( self . dpath , fname_cfgstr )
fpath = normpath ( fpath )
return fpath


signature:
def existing_versions ( self ) :
    """Returns data with different cfgstr values that were previously computed
 with this cacher .
 Example :
 > > > from ubelt . util _ cache import Cacher
 > > > # Ensure that some data exists
 > > > known _ fnames = set ( )
 > > > cacher = Cacher ( ' versioned _ data ' , cfgstr = ' 1 ' )
 > > > cacher . ensure ( lambda : ' data1 ' )
 > > > known _ fnames . add ( cacher . get _ fpath ( ) )
 > > > cacher = Cacher ( ' versioned _ data ' , cfgstr = ' 2 ' )
 > > > cacher . ensure ( lambda : ' data2 ' )
 > > > known _ fnames . add ( cacher . get _ fpath ( ) )
 > > > # List previously computed configs for this type
 > > > from os . path import basename
 > > > cacher = Cacher ( ' versioned _ data ' , cfgstr = ' 2 ' )
 > > > exist _ fpaths = set ( cacher . existing _ versions ( ) )
 > > > exist _ fnames = list ( map ( basename , exist _ fpaths ) )
 > > > print ( exist _ fnames )
 > > > assert exist _ fpaths = = known _ fnames
 [ ' versioned _ data _ 1 . pkl ' , ' versioned _ data _ 2 . pkl ' ]"""

implementation:
import glob
pattern = join ( self . dpath , self . fname + '_*' + self . ext )
for fname in glob . iglob ( pattern ) :
    data_fpath = join ( self . dpath , fname )
    yield data_fpath


signature:
def clear ( self , cfgstr = None ) :
    """Removes the saved cache and metadata from disk"""

implementation:
data_fpath = self . get_fpath ( cfgstr )
if self . verbose > 0 :
    self . log ( '[cacher] clear cache' )
if exists ( data_fpath ) :
    if self . verbose > 0 :
        self . log ( '[cacher] removing {}' . format ( data_fpath ) )
    os . remove ( data_fpath )
    # Remove the metadata if it exists
    meta_fpath = data_fpath + '.meta'
    if exists ( meta_fpath ) :
        os . remove ( meta_fpath )
else :
    if self . verbose > 0 :
        self . log ( '[cacher] ... nothing to clear' )


signature:
def tryload ( self , cfgstr = None , on_error = 'raise' ) :
    """Like load , but returns None if the load fails due to a cache miss .
 Args :
 on _ error ( str ) : How to handle non - io errors errors . Either raise ,
 which re - raises the exception , or clear which deletes the cache
 and returns None ."""

implementation:
cfgstr = self . _rectify_cfgstr ( cfgstr )
if self . enabled :
    try :
        if self . verbose > 1 :
            self . log ( '[cacher] tryload fname={}' . format ( self . fname ) )
        return self . load ( cfgstr )
    except IOError :
        if self . verbose > 0 :
            self . log ( '[cacher] ... {} cache miss' . format ( self . fname ) )
    except Exception :
        if self . verbose > 0 :
            self . log ( '[cacher] ... failed to load' )
        if on_error == 'raise' :
            raise
        elif on_error == 'clear' :
            self . clear ( cfgstr )
            return None
        else :
            raise KeyError ( 'Unknown method on_error={}' . format ( on_error ) )
else :
    if self . verbose > 1 :
        self . log ( '[cacher] ... cache disabled: fname={}' . format ( self . fname ) )
return None


signature:
def load ( self , cfgstr = None ) :
    """Loads the data
 Raises :
 IOError - if the data is unable to be loaded . This could be due to
 a cache miss or because the cache is disabled .
 Example :
 > > > from ubelt . util _ cache import * # NOQA
 > > > # Setting the cacher as enabled = False turns it off
 > > > cacher = Cacher ( ' test _ disabled _ load ' , ' ' , enabled = True )
 > > > cacher . save ( ' data ' )
 > > > assert cacher . load ( ) = = ' data '
 > > > cacher . enabled = False
 > > > assert cacher . tryload ( ) is None"""

implementation:
from six . moves import cPickle as pickle
cfgstr = self . _rectify_cfgstr ( cfgstr )
dpath = self . dpath
fname = self . fname
verbose = self . verbose
if not self . enabled :
    if verbose > 1 :
        self . log ( '[cacher] ... cache disabled: fname={}' . format ( self . fname ) )
    raise IOError ( 3 , 'Cache Loading Is Disabled' )
fpath = self . get_fpath ( cfgstr = cfgstr )
if not exists ( fpath ) :
    if verbose > 2 :
        self . log ( '[cacher] ... cache does not exist: ' 'dpath={} fname={} cfgstr={}' . format ( basename ( dpath ) , fname , cfgstr ) )
    raise IOError ( 2 , 'No such file or directory: %r' % ( fpath , ) )
else :
    if verbose > 3 :
        self . log ( '[cacher] ... cache exists: ' 'dpath={} fname={} cfgstr={}' . format ( basename ( dpath ) , fname , cfgstr ) )
try :
    with open ( fpath , 'rb' ) as file_ :
        data = pickle . load ( file_ )
except Exception as ex :
    if verbose > 0 :
        self . log ( 'CORRUPTED? fpath = %s' % ( fpath , ) )
    if verbose > 1 :
        self . log ( '[cacher] ... CORRUPTED? dpath={} cfgstr={}' . format ( basename ( dpath ) , cfgstr ) )
    if isinstance ( ex , ( EOFError , IOError , ImportError ) ) :
        raise IOError ( str ( ex ) )
    else :
        if verbose > 1 :
            self . log ( '[cacher] ... unknown reason for exception' )
        raise
else :
    if self . verbose > 2 :
        self . log ( '[cacher] ... {} cache hit' . format ( self . fname ) )
    elif verbose > 1 :
        self . log ( '[cacher] ... cache hit' )
return data


signature:
def save ( self , data , cfgstr = None ) :
    """Writes data to path specified by ` self . fpath ( cfgstr ) ` .
 Metadata containing information about the cache will also be appended
 to an adjacent file with the ` . meta ` suffix .
 Example :
 > > > from ubelt . util _ cache import * # NOQA
 > > > # Normal functioning
 > > > cfgstr = ' long - cfg ' * 32
 > > > cacher = Cacher ( ' test _ enabled _ save ' , cfgstr )
 > > > cacher . save ( ' data ' )
 > > > assert exists ( cacher . get _ fpath ( ) ) , ' should be enabeled '
 > > > assert exists ( cacher . get _ fpath ( ) + ' . meta ' ) , ' missing metadata '
 > > > # Setting the cacher as enabled = False turns it off
 > > > cacher2 = Cacher ( ' test _ disabled _ save ' , ' params ' , enabled = False )
 > > > cacher2 . save ( ' data ' )
 > > > assert not exists ( cacher2 . get _ fpath ( ) ) , ' should be disabled '"""

implementation:
from six . moves import cPickle as pickle
if not self . enabled :
    return
if self . verbose > 0 :
    self . log ( '[cacher] ... {} cache save' . format ( self . fname ) )
cfgstr = self . _rectify_cfgstr ( cfgstr )
condensed = self . _condense_cfgstr ( cfgstr )
# Make sure the cache directory exists
util_path . ensuredir ( self . dpath )
data_fpath = self . get_fpath ( cfgstr = cfgstr )
meta_fpath = data_fpath + '.meta'
# Also save metadata file to reconstruct hashing
with open ( meta_fpath , 'a' ) as file_ : # TODO : maybe append this in json or YML format ?
    file_ . write ( '\n\nsaving {}\n' . format ( util_time . timestamp ( ) ) )
    file_ . write ( self . fname + '\n' )
    file_ . write ( condensed + '\n' )
    file_ . write ( cfgstr + '\n' )
    file_ . write ( str ( self . meta ) + '\n' )
with open ( data_fpath , 'wb' ) as file_ : # Use protocol 2 to support python2 and 3
    pickle . dump ( data , file_ , protocol = self . protocol )


signature:
def ensure ( self , func , * args , ** kwargs ) :
    r"""Wraps around a function . A cfgstr must be stored in the base cacher .
 Args :
 func ( callable ) : function that will compute data on cache miss
 * args : passed to func
 * * kwargs : passed to func
 Example :
 > > > from ubelt . util _ cache import * # NOQA
 > > > def func ( ) :
 > > > return ' expensive result '
 > > > fname = ' test _ cacher _ ensure '
 > > > cfgstr = ' func params '
 > > > cacher = Cacher ( fname , cfgstr )
 > > > cacher . clear ( )
 > > > data1 = cacher . ensure ( func )
 > > > data2 = cacher . ensure ( func )
 > > > assert data1 = = ' expensive result '
 > > > assert data1 = = data2
 > > > cacher . clear ( )"""

implementation:
data = self . tryload ( )
if data is None :
    data = func ( * args , ** kwargs )
    self . save ( data )
return data


signature:
def _get_certificate ( self , cfgstr = None ) :
    """Returns the stamp certificate if it exists"""

implementation:
certificate = self . cacher . tryload ( cfgstr = cfgstr )
return certificate


signature:
def _rectify_products ( self , product = None ) :
    """puts products in a normalized format"""

implementation:
products = self . product if product is None else product
if products is None :
    return None
if not isinstance ( products , ( list , tuple ) ) :
    products = [ products ]
return products


signature:
def _product_file_hash ( self , product = None ) :
    """Get the hash of the each product file"""

implementation:
if self . hasher is None :
    return None
else :
    products = self . _rectify_products ( product )
    product_file_hash = [ util_hash . hash_file ( p , hasher = self . hasher , base = 'hex' ) for p in products ]
    return product_file_hash


signature:
def expired ( self , cfgstr = None , product = None ) :
    """Check to see if a previously existing stamp is still valid and if the
 expected result of that computation still exists .
 Args :
 cfgstr ( str , optional ) : override the default cfgstr if specified
 product ( PathLike or Sequence [ PathLike ] , optional ) : override the
 default product if specified"""

implementation:
products = self . _rectify_products ( product )
certificate = self . _get_certificate ( cfgstr = cfgstr )
if certificate is None : # We dont have a certificate , so we are expired
    is_expired = True
elif products is None : # We dont have a product to check , so assume not expired
    is_expired = False
elif not all ( map ( os . path . exists , products ) ) : # We are expired if the expected product does not exist
    is_expired = True
else : # We are expired if the hash of the existing product data
# does not match the expected hash in the certificate
    product_file_hash = self . _product_file_hash ( products )
    certificate_hash = certificate . get ( 'product_file_hash' , None )
    is_expired = product_file_hash != certificate_hash
return is_expired


signature:
def renew ( self , cfgstr = None , product = None ) :
    """Recertify that the product has been recomputed by writing a new
 certificate to disk ."""

implementation:
products = self . _rectify_products ( product )
certificate = { 'timestamp' : util_time . timestamp ( ) , 'product' : products , }
if products is not None :
    if not all ( map ( os . path . exists , products ) ) :
        raise IOError ( 'The stamped product must exist: {}' . format ( products ) )
    certificate [ 'product_file_hash' ] = self . _product_file_hash ( products )
self . cacher . save ( certificate , cfgstr = cfgstr )
return certificate


signature:
def isatty ( self ) : # nocover
    """Returns true of the redirect is a terminal .
 Notes :
 Needed for IPython . embed to work properly when this class is used
 to override stdout / stderr ."""

implementation: return ( self . redirect is not None and hasattr ( self . redirect , 'isatty' ) and self . redirect . isatty ( ) )


signature:
def encoding ( self ) :
    """Gets the encoding of the ` redirect ` IO object
 Doctest :
 > > > redirect = io . StringIO ( )
 > > > assert TeeStringIO ( redirect ) . encoding is None
 > > > assert TeeStringIO ( None ) . encoding is None
 > > > assert TeeStringIO ( sys . stdout ) . encoding is sys . stdout . encoding
 > > > redirect = io . TextIOWrapper ( io . StringIO ( ) )
 > > > assert TeeStringIO ( redirect ) . encoding is redirect . encoding"""

implementation:
if self . redirect is not None :
    return self . redirect . encoding
else :
    return super ( TeeStringIO , self ) . encoding


signature:
def write ( self , msg ) :
    """Write to this and the redirected stream"""

implementation:
if self . redirect is not None :
    self . redirect . write ( msg )
if six . PY2 :
    from xdoctest . utils . util_str import ensure_unicode
    msg = ensure_unicode ( msg )
super ( TeeStringIO , self ) . write ( msg )


signature:
def flush ( self ) : # nocover
    """Flush to this and the redirected stream"""

implementation:
if self . redirect is not None :
    self . redirect . flush ( )
super ( TeeStringIO , self ) . flush ( )


signature:
def log_part ( self ) :
    """Log what has been captured so far"""

implementation:
self . cap_stdout . seek ( self . _pos )
text = self . cap_stdout . read ( )
self . _pos = self . cap_stdout . tell ( )
self . parts . append ( text )
self . text = text


signature:
def stop ( self ) :
    """Doctest :
 > > > CaptureStdout ( enabled = False ) . stop ( )
 > > > CaptureStdout ( enabled = True ) . stop ( )"""

implementation:
if self . enabled :
    self . started = False
    sys . stdout = self . orig_stdout


signature:
def platform_data_dir ( ) :
    """Returns path for user - specific data files
 Returns :
 PathLike : path to the data dir used by the current operating system"""

implementation:
if LINUX : # nocover
    dpath_ = os . environ . get ( 'XDG_DATA_HOME' , '~/.local/share' )
elif DARWIN : # nocover
    dpath_ = '~/Library/Application Support'
elif WIN32 : # nocover
    dpath_ = os . environ . get ( 'APPDATA' , '~/AppData/Roaming' )
else : # nocover
    raise '~/AppData/Local'
dpath = normpath ( expanduser ( dpath_ ) )
return dpath


signature:
def platform_config_dir ( ) :
    """Returns a directory which should be writable for any application
 This should be used for persistent configuration files .
 Returns :
 PathLike : path to the cahce dir used by the current operating system"""

implementation:
if LINUX : # nocover
    dpath_ = os . environ . get ( 'XDG_CONFIG_HOME' , '~/.config' )
elif DARWIN : # nocover
    dpath_ = '~/Library/Application Support'
elif WIN32 : # nocover
    dpath_ = os . environ . get ( 'APPDATA' , '~/AppData/Roaming' )
else : # nocover
    raise NotImplementedError ( 'Unknown Platform  %r' % ( sys . platform , ) )
dpath = normpath ( expanduser ( dpath_ ) )
return dpath


signature:
def ensure_app_data_dir ( appname , * args ) :
    """Calls ` get _ app _ data _ dir ` but ensures the directory exists .
 Args :
 appname ( str ) : the name of the application
 * args : any other subdirectories may be specified
 SeeAlso :
 get _ app _ data _ dir
 Example :
 > > > import ubelt as ub
 > > > dpath = ub . ensure _ app _ data _ dir ( ' ubelt ' )
 > > > assert exists ( dpath )"""

implementation:
from ubelt import util_path
dpath = get_app_data_dir ( appname , * args )
util_path . ensuredir ( dpath )
return dpath


signature:
def ensure_app_config_dir ( appname , * args ) :
    """Calls ` get _ app _ config _ dir ` but ensures the directory exists .
 Args :
 appname ( str ) : the name of the application
 * args : any other subdirectories may be specified
 SeeAlso :
 get _ app _ config _ dir
 Example :
 > > > import ubelt as ub
 > > > dpath = ub . ensure _ app _ config _ dir ( ' ubelt ' )
 > > > assert exists ( dpath )"""

implementation:
from ubelt import util_path
dpath = get_app_config_dir ( appname , * args )
util_path . ensuredir ( dpath )
return dpath


signature:
def ensure_app_cache_dir ( appname , * args ) :
    """Calls ` get _ app _ cache _ dir ` but ensures the directory exists .
 Args :
 appname ( str ) : the name of the application
 * args : any other subdirectories may be specified
 SeeAlso :
 get _ app _ cache _ dir
 Example :
 > > > import ubelt as ub
 > > > dpath = ub . ensure _ app _ cache _ dir ( ' ubelt ' )
 > > > assert exists ( dpath )"""

implementation:
from ubelt import util_path
dpath = get_app_cache_dir ( appname , * args )
util_path . ensuredir ( dpath )
return dpath


signature:
def startfile ( fpath , verbose = True ) : # nocover
    """Uses default program defined by the system to open a file .
 This is done via ` os . startfile ` on windows , ` open ` on mac , and ` xdg - open `
 on linux .
 Args :
 fpath ( PathLike ) : a file to open using the program associated with the
 files extension type .
 verbose ( int ) : verbosity
 References :
 http : / / stackoverflow . com / questions / 2692873 / quote - posix
 DisableExample :
 > > > # This test interacts with a GUI frontend , not sure how to test .
 > > > import ubelt as ub
 > > > base = ub . ensure _ app _ cache _ dir ( ' ubelt ' )
 > > > fpath1 = join ( base , ' test _ open . txt ' )
 > > > ub . touch ( fpath1)
 > > > proc = ub . startfile ( fpath1)"""

implementation:
from ubelt import util_cmd
if verbose :
    print ( '[ubelt] startfile("{}")' . format ( fpath ) )
fpath = normpath ( fpath )
if not exists ( fpath ) :
    raise Exception ( 'Cannot start nonexistant file: %r' % fpath )
if not WIN32 :
    import pipes
    fpath = pipes . quote ( fpath )
if LINUX :
    info = util_cmd . cmd ( ( 'xdg-open' , fpath ) , detach = True , verbose = verbose )
elif DARWIN :
    info = util_cmd . cmd ( ( 'open' , fpath ) , detach = True , verbose = verbose )
elif WIN32 :
    os . startfile ( fpath )
    info = None
else :
    raise RuntimeError ( 'Unknown Platform' )
if info is not None :
    if not info [ 'proc' ] :
        raise Exception ( 'startfile failed' )


signature:
def find_exe ( name , multi = False , path = None ) :
    """Locate a command .
 Search your local filesystem for an executable and return the first
 matching file with executable permission .
 Args :
 name ( str ) : globstr of matching filename
 multi ( bool ) : if True return all matches instead of just the first .
 Defaults to False .
 path ( str or Iterable [ PathLike ] ) : overrides the system PATH variable .
 Returns :
 PathLike or List [ PathLike ] or None : returns matching executable ( s ) .
 SeeAlso :
 shutil . which - which is available in Python 3.3 + .
 Notes :
 This is essentially the ` which ` UNIX command
 References :
 https : / / stackoverflow . com / questions / 377017 / test - if - executable - exists - in - python / 377028#377028
 https : / / docs . python . org / dev / library / shutil . html # shutil . which
 Example :
 > > > find _ exe ( ' ls ' )
 > > > find _ exe ( ' ping ' )
 > > > assert find _ exe ( ' which ' ) = = find _ exe ( find _ exe ( ' which ' ) )
 > > > find _ exe ( ' which ' , multi = True )
 > > > find _ exe ( ' ping ' , multi = True )
 > > > find _ exe ( ' cmake ' , multi = True )
 > > > find _ exe ( ' nvcc ' , multi = True )
 > > > find _ exe ( ' noexist ' , multi = True )
 Example :
 > > > assert not find _ exe ( ' noexist ' , multi = False )
 > > > assert find _ exe ( ' ping ' , multi = False )
 > > > assert not find _ exe ( ' noexist ' , multi = True )
 > > > assert find _ exe ( ' ping ' , multi = True )
 Benchmark :
 > > > # xdoctest : + IGNORE _ WANT
 > > > import ubelt as ub
 > > > import shutil
 > > > for timer in ub . Timerit ( 100 , bestof = 10 , label = ' ub . find _ exe ' ) :
 > > > ub . find _ exe ( ' which ' )
 > > > for timer in ub . Timerit ( 100 , bestof = 10 , label = ' shutil . which ' ) :
 > > > shutil . which ( ' which ' )
 Timed best = 58.71 μs , mean = 59.64 ± 0.96 μs for ub . find _ exe
 Timed best = 72.75 μs , mean = 73.07 ± 0.22 μs for shutil . which"""

implementation:
candidates = find_path ( name , path = path , exact = True )
mode = os . X_OK | os . F_OK
results = ( fpath for fpath in candidates if os . access ( fpath , mode ) and not isdir ( fpath ) )
if not multi :
    for fpath in results :
        return fpath
else :
    return list ( results )


signature:
def find_path ( name , path = None , exact = False ) :
    """Search for a file or directory on your local filesystem by name
 ( file must be in a directory specified in a PATH environment variable )
 Args :
 fname ( PathLike or str ) : file name to match .
 If exact is False this may be a glob pattern
 path ( str or Iterable [ PathLike ] ) : list of directories to search either
 specified as an os . pathsep separated string or a list of
 directories . Defaults to environment PATH .
 exact ( bool ) : if True , only returns exact matches . Default False .
 Notes :
 For recursive behavior set ` path = ( d for d , _ , _ in os . walk ( ' . ' ) ) ` ,
 where ' . ' might be replaced by the root directory of interest .
 Example :
 > > > list ( find _ path ( ' ping ' , exact = True ) )
 > > > list ( find _ path ( ' bin ' ) )
 > > > list ( find _ path ( ' bin ' ) )
 > > > list ( find _ path ( ' * cc * ' ) )
 > > > list ( find _ path ( ' cmake * ' ) )
 Example :
 > > > import ubelt as ub
 > > > from os . path import dirname
 > > > path = dirname ( dirname ( ub . util _ platform . _ _ file _ _ ) )
 > > > res = sorted ( find _ path ( ' ubelt / util _ * . py ' , path = path ) )
 > > > assert len ( res ) > = 10
 > > > res = sorted ( find _ path ( ' ubelt / util _ platform . py ' , path = path , exact = True ) )
 > > > print ( res )
 > > > assert len ( res ) = = 1"""

implementation:
path = os . environ . get ( 'PATH' , os . defpath ) if path is None else path
dpaths = path . split ( os . pathsep ) if isinstance ( path , six . string_types ) else path
candidates = ( join ( dpath , name ) for dpath in dpaths )
if exact :
    if WIN32 : # nocover
        pathext = [ '' ] + os . environ . get ( 'PATHEXT' , '' ) . split ( os . pathsep )
        candidates = ( p + ext for p in candidates for ext in pathext )
    candidates = filter ( exists , candidates )
else :
    import glob
    candidates = it . chain . from_iterable ( glob . glob ( pattern ) for pattern in candidates )
return candidates


signature:
def editfile ( fpath , verbose = True ) : # nocover
    """DEPRICATED : This has been ported to xdev , please use that version .
 Opens a file or code corresponding to a live python object in your
 preferred visual editor . This function is mainly useful in an interactive
 IPython session .
 The visual editor is determined by the ` VISUAL ` environment variable . If
 this is not specified it defaults to gvim .
 Args :
 fpath ( PathLike ) : a file path or python module / function
 verbose ( int ) : verbosity
 DisableExample :
 > > > # This test interacts with a GUI frontend , not sure how to test .
 > > > import ubelt as ub
 > > > ub . editfile ( ub . util _ platform . _ _ file _ _ )
 > > > ub . editfile ( ub )
 > > > ub . editfile ( ub . editfile )"""

implementation:
from six import types
from ubelt import util_cmd
import warnings
warnings . warn ( 'Please use xdev.editfile instead' , DeprecationWarning )
if not isinstance ( fpath , six . string_types ) :
    if isinstance ( fpath , types . ModuleType ) :
        fpath = fpath . __file__
    else :
        fpath = sys . modules [ fpath . __module__ ] . __file__
    fpath_py = fpath . replace ( '.pyc' , '.py' )
    if exists ( fpath_py ) :
        fpath = fpath_py
if verbose :
    print ( '[ubelt] editfile("{}")' . format ( fpath ) )
editor = os . environ . get ( 'VISUAL' , 'gvim' )
if not exists ( fpath ) :
    raise IOError ( 'Cannot start nonexistant file: %r' % fpath )
util_cmd . cmd ( [ editor , fpath ] , fpath , detach = True )


signature:
def augpath ( path , suffix = '' , prefix = '' , ext = None , base = None , multidot = False ) :
    """Augments a path with a new basename , extension , prefix and / or suffix .
 A prefix is inserted before the basename . A suffix is inserted
 between the basename and the extension . The basename and extension can be
 replaced with a new one .
 Args :
 path ( PathLike ) : string representation of a path
 suffix ( str ) : placed between the basename and extension
 prefix ( str ) : placed in front of the basename
 ext ( str ) : if specified , replaces the extension
 base ( str ) : if specified , replaces the basename ( without extension )
 multidot ( bool ) : if False , everything after the last dot in the
 basename is the extension . If True , everything after the first dot
 in the basename is the extension ( Defaults to False ) .
 Returns :
 PathLike : augmented path
 CommandLine :
 python - m ubelt . util _ path augpath
 Example :
 > > > import ubelt as ub
 > > > path = ' foo . bar '
 > > > suffix = ' _ suff '
 > > > prefix = ' pref _ '
 > > > ext = ' . baz '
 > > > newpath = ub . augpath ( path , suffix , prefix , ext = ext , base = ' bar ' )
 > > > print ( ' newpath = % s ' % ( newpath , ) )
 newpath = pref _ bar _ suff . baz
 Example :
 > > > augpath ( ' foo . bar ' )
 ' foo . bar '
 > > > augpath ( ' foo . bar ' , ext = ' . BAZ ' )
 ' foo . BAZ '
 > > > augpath ( ' foo . bar ' , suffix = ' _ ' )
 ' foo _ . bar '
 > > > augpath ( ' foo . bar ' , prefix = ' _ ' )
 ' _ foo . bar '
 > > > augpath ( ' foo . bar ' , base = ' baz ' )
 ' baz . bar '
 > > > augpath ( ' foo . tar . gz ' , ext = ' . zip ' , multidot = True )
 foo . zip
 > > > augpath ( ' foo . tar . gz ' , ext = ' . zip ' , multidot = False )
 foo . tar . zip"""

implementation:
# Breakup path
dpath , fname = split ( path )
if multidot :
    parts = fname . split ( '.' )
    fname_noext = '.' . join ( parts [ : 1 ] )
    orig_ext = '.' . join ( parts [ 1 : ] )
else :
    fname_noext , orig_ext = splitext ( fname )
ext = orig_ext if ext is None else ext
fname_noext = fname_noext if base is None else base
# Augment and recombine into new path
new_fname = '' . join ( ( prefix , fname_noext , suffix , ext ) )
newpath = join ( dpath , new_fname )
return newpath


signature:
def userhome ( username = None ) :
    """Returns the user ' s home directory .
 If ` username ` is None , this is the directory for the current user .
 Args :
 username ( str ) : name of a user on the system
 Returns :
 PathLike : userhome _ dpath : path to the home directory
 Example :
 > > > import getpass
 > > > username = getpass . getuser ( )
 > > > assert userhome ( ) = = expanduser ( ' ~ ' )
 > > > assert userhome ( username ) = = expanduser ( ' ~ ' )"""

implementation:
if username is None : # get home directory for the current user
    if 'HOME' in os . environ :
        userhome_dpath = os . environ [ 'HOME' ]
    else : # nocover
        if sys . platform . startswith ( 'win32' ) : # win32 fallback when HOME is not defined
            if 'USERPROFILE' in os . environ :
                userhome_dpath = os . environ [ 'USERPROFILE' ]
            elif 'HOMEPATH' in os . environ :
                drive = os . environ . get ( 'HOMEDRIVE' , '' )
                userhome_dpath = join ( drive , os . environ [ 'HOMEPATH' ] )
            else :
                raise OSError ( "Cannot determine the user's home directory" )
        else : # posix fallback when HOME is not defined
            import pwd
            userhome_dpath = pwd . getpwuid ( os . getuid ( ) ) . pw_dir
else : # A specific user directory was requested
    if sys . platform . startswith ( 'win32' ) : # nocover
    # get the directory name for the current user
        c_users = dirname ( userhome ( ) )
        userhome_dpath = join ( c_users , username )
        if not exists ( userhome_dpath ) :
            raise KeyError ( 'Unknown user: {}' . format ( username ) )
    else :
        import pwd
        try :
            pwent = pwd . getpwnam ( username )
        except KeyError : # nocover
            raise KeyError ( 'Unknown user: {}' . format ( username ) )
        userhome_dpath = pwent . pw_dir
return userhome_dpath


signature:
def compressuser ( path , home = '~' ) :
    """Inverse of ` os . path . expanduser `
 Args :
 path ( PathLike ) : path in system file structure
 home ( str ) : symbol used to replace the home path . Defaults to ' ~ ' , but
 you might want to use ' $ HOME ' or ' % USERPROFILE % ' instead .
 Returns :
 PathLike : path : shortened path replacing the home directory with a tilde
 CommandLine :
 xdoctest - m ubelt . util _ path compressuser
 Example :
 > > > path = expanduser ( ' ~ ' )
 > > > assert path ! = ' ~ '
 > > > assert compressuser ( path ) = = ' ~ '
 > > > assert compressuser ( path + ' 1 ' ) = = path + ' 1'
 > > > assert compressuser ( path + ' / 1 ' ) = = join ( ' ~ ' , ' 1 ' )
 > > > assert compressuser ( path + ' / 1 ' , ' $ HOME ' ) = = join ( ' $ HOME ' , ' 1 ' )"""

implementation:
path = normpath ( path )
userhome_dpath = userhome ( )
if path . startswith ( userhome_dpath ) :
    if len ( path ) == len ( userhome_dpath ) :
        path = home
    elif path [ len ( userhome_dpath ) ] == os . path . sep :
        path = home + path [ len ( userhome_dpath ) : ]
return path


signature:
def truepath ( path , real = False ) :
    """Normalizes a string representation of a path and does shell - like expansion .
 Args :
 path ( PathLike ) : string representation of a path
 real ( bool ) : if True , all symbolic links are followed . ( default : False )
 Returns :
 PathLike : normalized path
 Note :
 This function is similar to the composition of expanduser , expandvars ,
 normpath , and ( realpath if ` real ` else abspath ) . However , on windows
 backslashes are then replaced with forward slashes to offer a
 consistent unix - like experience across platforms .
 On windows expanduser will expand environment variables formatted as
 % name % , whereas on unix , this will not occur .
 CommandLine :
 python - m ubelt . util _ path truepath
 Example :
 > > > import ubelt as ub
 > > > assert ub . truepath ( ' ~ / foo ' ) = = join ( ub . userhome ( ) , ' foo ' )
 > > > assert ub . truepath ( ' ~ / foo ' ) = = ub . truepath ( ' ~ / foo / bar / . . ' )
 > > > assert ub . truepath ( ' ~ / foo ' , real = True ) = = ub . truepath ( ' ~ / foo ' )"""

implementation:
path = expanduser ( path )
path = expandvars ( path )
if real :
    path = realpath ( path )
else :
    path = abspath ( path )
path = normpath ( path )
return path


signature:
def ensuredir ( dpath , mode = 0o1777 , verbose = None ) :
    r"""Ensures that directory will exist . Creates new dir with sticky bits by
 default
 Args :
 dpath ( PathLike ) : dir to ensure . Can also be a tuple to send to join
 mode ( int ) : octal mode of directory ( default 0o1777)
 verbose ( int ) : verbosity ( default 0)
 Returns :
 PathLike : path : the ensured directory
 Notes :
 This function is not thread - safe in Python2
 Example :
 > > > from ubelt . util _ platform import * # NOQA
 > > > import ubelt as ub
 > > > cache _ dpath = ub . ensure _ app _ cache _ dir ( ' ubelt ' )
 > > > dpath = join ( cache _ dpath , ' ensuredir ' )
 > > > if exists ( dpath ) :
 . . . os . rmdir ( dpath )
 > > > assert not exists ( dpath )
 > > > ub . ensuredir ( dpath )
 > > > assert exists ( dpath )
 > > > os . rmdir ( dpath )"""

implementation:
if verbose is None : # nocover
    verbose = 0
if isinstance ( dpath , ( list , tuple ) ) : # nocover
    dpath = join ( * dpath )
if not exists ( dpath ) :
    if verbose : # nocover
        print ( 'Ensuring new directory (%r)' % dpath )
    if sys . version_info . major == 2 : # nocover
        os . makedirs ( normpath ( dpath ) , mode = mode )
    else :
        os . makedirs ( normpath ( dpath ) , mode = mode , exist_ok = True )
else :
    if verbose : # nocover
        print ( 'Ensuring existing directory (%r)' % dpath )
return dpath


signature:
def parse_version ( package ) :
    """Statically parse the version number from _ _ init _ _ . py
 CommandLine :
 python - c " import setup ; print ( setup . parse _ version ( ' ubelt ' ) ) " """

implementation:
from os . path import dirname , join , exists
import ast
# Check if the package is a single - file or multi - file package
_candiates = [ join ( dirname ( __file__ ) , package + '.py' ) , join ( dirname ( __file__ ) , package , '__init__.py' ) , ]
_found = [ init_fpath for init_fpath in _candiates if exists ( init_fpath ) ]
if len ( _found ) > 0 :
    init_fpath = _found [ 0 ]
elif len ( _found ) > 1 :
    raise Exception ( 'parse_version found multiple init files' )
elif len ( _found ) == 0 :
    raise Exception ( 'Cannot find package init file' )
with open ( init_fpath , 'r' ) as file_ :
    sourcecode = file_ . read ( )
pt = ast . parse ( sourcecode )
class VersionVisitor ( ast . NodeVisitor ) :
    def visit_Assign ( self , node ) :
        for target in node . targets :
            if getattr ( target , 'id' , None ) == '__version__' :
                self . version = node . value . s
visitor = VersionVisitor ( )
visitor . visit ( pt )
return visitor . version