signature:
def build_recursive_delocalize_command ( source , outputs , file_provider ) :
    """Return a multi - line string with a shell script to copy recursively .
 Arguments :
 source : Folder with the data .
 For example / mnt / data
 outputs : a list of OutputFileParam .
 file _ provider : file provider string used to filter the output params ; the
 returned command will only apply outputs whose file provider
 matches this file filter .
 Returns :
 a multi - line string with a shell script that copies the inputs
 recursively to GCS ."""

implementation:
command = _LOCALIZE_COMMAND_MAP [ file_provider ]
filtered_outputs = [ var for var in outputs if var . recursive and var . file_provider == file_provider ]
return '\n' . join ( [ textwrap . dedent ( """
      for ((i = 0; i < 3; i++)); do
        if {command} {data_mount}/{docker_path} {destination_uri}; then
          break
        elif ((i == 2)); then
          2>&1 echo "Recursive de-localization failed."
          exit 1
        fi
      done
      """ ) . format ( command = command , data_mount = source . rstrip ( '/' ) , docker_path = var . docker_path , destination_uri = var . uri ) for var in filtered_outputs ] )


signature:
def build_mount_env ( source , mounts ) :
    """Return a multi - line string with export statements for the variables .
 Arguments :
 source : Folder with the data . For example / mnt / data
 mounts : a list of MountParam
 Returns :
 a multi - line string with a shell script that sets environment variables
 corresponding to the mounts ."""

implementation: return '\n' . join ( [ 'export {0}={1}/{2}' . format ( var . name , source . rstrip ( '/' ) , var . docker_path . rstrip ( '/' ) ) for var in mounts ] )


signature:
def get_job_and_task_param ( job_params , task_params , field ) :
    """Returns a dict combining the field for job and task params ."""

implementation: return job_params . get ( field , set ( ) ) | task_params . get ( field , set ( ) )


signature:
def _parse_arguments ( ) :
    """Parses command line arguments .
 Returns :
 A Namespace of parsed arguments ."""

implementation:
# Handle version flag and exit if it was passed .
param_util . handle_version_flag ( )
parser = provider_base . create_parser ( sys . argv [ 0 ] )
parser . add_argument ( '--version' , '-v' , default = False , help = 'Print the dsub version and exit.' )
parser . add_argument ( '--jobs' , '-j' , required = True , nargs = '*' , help = 'List of job-ids to delete. Use "*" to delete all running jobs.' )
parser . add_argument ( '--tasks' , '-t' , nargs = '*' , help = 'List of tasks in an array job to delete.' )
parser . add_argument ( '--users' , '-u' , nargs = '*' , default = [ ] , help = """Deletes only those jobs which were submitted by the list of users.
          Use "*" to delete jobs of any user.""" )
parser . add_argument ( '--age' , help = """Deletes only those jobs newer than the specified age. Ages can be
          listed using a number followed by a unit. Supported units are
          s (seconds), m (minutes), h (hours), d (days), w (weeks).
          For example: '7d' (7 days). Bare numbers are treated as UTC.""" )
parser . add_argument ( '--label' , nargs = '*' , action = param_util . ListParamAction , default = [ ] , help = 'User labels to match. Tasks returned must match all labels.' , metavar = 'KEY=VALUE' )
# Shared arguments between the " google " and " google - v2 " providers
google_common = parser . add_argument_group ( title = 'google-common' , description = 'Options common to the "google" and "google-v2" providers' )
google_common . add_argument ( '--project' , help = 'Cloud project ID in which to find and delete the job(s)' )
return provider_base . parse_args ( parser , { 'google' : [ 'project' ] , 'google-v2' : [ 'project' ] , 'test-fails' : [ ] , 'local' : [ ] , } , sys . argv [ 1 : ] )


signature:
def _emit_search_criteria ( user_ids , job_ids , task_ids , labels ) :
    """Print the filters used to delete tasks . Use raw flags as arguments ."""

implementation:
print ( 'Delete running jobs:' )
print ( '  user:' )
print ( '    %s\n' % user_ids )
print ( '  job-id:' )
print ( '    %s\n' % job_ids )
if task_ids :
    print ( '  task-id:' )
    print ( '    %s\n' % task_ids )
# Labels are in a LabelParam namedtuple and must be reformated for printing .
if labels :
    print ( '  labels:' )
    print ( '    %s\n' % repr ( labels ) )


signature:
def ddel_tasks ( provider , user_ids = None , job_ids = None , task_ids = None , labels = None , create_time_min = None , create_time_max = None ) :
    """Kill jobs or job tasks .
 This function separates ddel logic from flag parsing and user output . Users
 of ddel who intend to access the data programmatically should use this .
 Args :
 provider : an instantiated dsub provider .
 user _ ids : a set of user ids who " own " the job ( s ) to delete .
 job _ ids : a set of job ids to delete .
 task _ ids : a set of task ids to delete .
 labels : a set of LabelParam , each must match the job ( s ) to be cancelled .
 create _ time _ min : a timezone - aware datetime value for the earliest create
 time of a task , inclusive .
 create _ time _ max : a timezone - aware datetime value for the most recent create
 time of a task , inclusive .
 Returns :
 list of job ids which were deleted ."""

implementation:
# Delete the requested jobs
deleted_tasks , error_messages = provider . delete_jobs ( user_ids , job_ids , task_ids , labels , create_time_min , create_time_max )
# Emit any errors canceling jobs
for msg in error_messages :
    print ( msg )
return deleted_tasks


signature:
def get_action_by_id ( op , action_id ) :
    """Return the operation ' s array of actions ."""

implementation:
actions = get_actions ( op )
if actions and 1 <= action_id < len ( actions ) :
    return actions [ action_id - 1 ]


signature:
def _get_action_by_name ( op , name ) :
    """Return the value for the specified action ."""

implementation:
actions = get_actions ( op )
for action in actions :
    if action . get ( 'name' ) == name :
        return action


signature:
def get_action_environment ( op , name ) :
    """Return the environment for the operation ."""

implementation:
action = _get_action_by_name ( op , name )
if action :
    return action . get ( 'environment' )


signature:
def get_action_image ( op , name ) :
    """Return the image for the operation ."""

implementation:
action = _get_action_by_name ( op , name )
if action :
    return action . get ( 'imageUri' )


signature:
def get_failed_events ( op ) :
    """Return the events ( if any ) with a non - zero exitStatus ."""

implementation:
events = get_events ( op )
if events :
    return [ e for e in events if int ( e . get ( 'details' , { } ) . get ( 'exitStatus' , 0 ) ) != 0 ]
return None


signature:
def get_event_of_type ( op , event_type ) :
    """Return all events of a particular type ."""

implementation:
events = get_events ( op )
if not events :
    return None
return [ e for e in events if e . get ( 'details' , { } ) . get ( '@type' ) == event_type ]


signature:
def get_last_update ( op ) :
    """Return the most recent timestamp in the operation ."""

implementation:
last_update = get_end_time ( op )
if not last_update :
    last_event = get_last_event ( op )
    if last_event :
        last_update = last_event [ 'timestamp' ]
if not last_update :
    last_update = get_create_time ( op )
return last_update


signature:
def is_dsub_operation ( op ) :
    """Determine if a pipelines operation is a dsub request .
 We don ' t have a rigorous way to identify an operation as being submitted
 by dsub . Our best option is to check for certain fields that have always
 been part of dsub operations .
 - labels : job - id , job - name , and user - id have always existed . The dsub - version
 label has always existed for the google - v2 provider .
 Args :
 op : a pipelines operation .
 Returns :
 Boolean , true if the pipeline run was generated by dsub ."""

implementation:
if not is_pipeline ( op ) :
    return False
for name in [ 'dsub-version' , 'job-id' , 'job-name' , 'user-id' ] :
    if not get_label ( op , name ) :
        return False
return True


signature:
def _prepare_summary_table ( rows ) :
    """Create a new table that is a summary of the input rows .
 All with the same ( job - name or job - id , status ) go together .
 Args :
 rows : the input rows , a list of dictionaries .
 Returns :
 A new row set of summary information ."""

implementation:
if not rows :
    return [ ]
# We either group on the job - name ( if present ) or fall back to the job - id
key_field = 'job-name'
if key_field not in rows [ 0 ] :
    key_field = 'job-id'
# Group each of the rows based on ( job - name or job - id , status )
grouped = collections . defaultdict ( lambda : collections . defaultdict ( lambda : [ ] ) )
for row in rows :
    grouped [ row . get ( key_field , '' ) ] [ row . get ( 'status' , '' ) ] += [ row ]
# Now that we have the rows grouped , create a summary table .
# Use the original table as the driver in order to preserve the order .
new_rows = [ ]
for job_key in sorted ( grouped . keys ( ) ) :
    group = grouped . get ( job_key , None )
    canonical_status = [ 'RUNNING' , 'SUCCESS' , 'FAILURE' , 'CANCEL' ]
    # Written this way to ensure that if somehow a new status is introduced ,
    # it shows up in our output .
    for status in canonical_status + sorted ( group . keys ( ) ) :
        if status not in group :
            continue
        task_count = len ( group [ status ] )
        del group [ status ]
        if task_count :
            summary_row = collections . OrderedDict ( )
            summary_row [ key_field ] = job_key
            summary_row [ 'status' ] = status
            summary_row [ 'task-count' ] = task_count
            new_rows . append ( summary_row )
return new_rows


signature:
def _prepare_row ( task , full , summary ) :
    """return a dict with the task ' s info ( more if " full " is set ) ."""

implementation:
# Would like to include the Job ID in the default set of columns , but
# it is a long value and would leave little room for status and update time .
row_spec = collections . namedtuple ( 'row_spec' , [ 'key' , 'required' , 'default_value' ] )
# pyformat : disable
default_columns = [ row_spec ( 'job-name' , True , None ) , row_spec ( 'task-id' , False , None ) , row_spec ( 'last-update' , True , None ) , row_spec ( 'status-message' , True , None ) ]
full_columns = default_columns + [ row_spec ( 'job-id' , True , None ) , row_spec ( 'user-id' , True , None ) , row_spec ( 'status' , True , None ) , row_spec ( 'status-detail' , True , None ) , row_spec ( 'task-attempt' , False , None ) , row_spec ( 'create-time' , True , None ) , row_spec ( 'start-time' , True , None ) , row_spec ( 'end-time' , True , None ) , row_spec ( 'internal-id' , True , None ) , row_spec ( 'logging' , True , None ) , row_spec ( 'labels' , True , { } ) , row_spec ( 'envs' , True , { } ) , row_spec ( 'inputs' , True , { } ) , row_spec ( 'input-recursives' , False , { } ) , row_spec ( 'outputs' , True , { } ) , row_spec ( 'output-recursives' , False , { } ) , row_spec ( 'mounts' , True , { } ) , row_spec ( 'provider' , True , None ) , row_spec ( 'provider-attributes' , True , { } ) , row_spec ( 'events' , True , [ ] ) , row_spec ( 'user-project' , False , None ) , row_spec ( 'dsub-version' , False , None ) , row_spec ( 'script-name' , False , None ) , row_spec ( 'script' , False , None ) , ]
summary_columns = default_columns + [ row_spec ( 'job-id' , True , None ) , row_spec ( 'user-id' , True , None ) , row_spec ( 'status' , True , None ) , ]
# pyformat : enable
assert not ( full and summary ) , 'Full and summary cannot both be enabled'
if full :
    columns = full_columns
elif summary :
    columns = summary_columns
else :
    columns = default_columns
row = { }
for col in columns :
    key , required , default = col
    value = task . get_field ( key , default )
    if required or value is not None :
        row [ key ] = value
return row


implementation:
# Handle version flag and exit if it was passed .
param_util . handle_version_flag ( )
parser = provider_base . create_parser ( sys . argv [ 0 ] )
parser . add_argument ( '--version' , '-v' , default = False , help = 'Print the dsub version and exit.' )
parser . add_argument ( '--jobs' , '-j' , nargs = '*' , help = 'A list of jobs IDs on which to check status' )
parser . add_argument ( '--names' , '-n' , nargs = '*' , help = 'A list of job names on which to check status' )
parser . add_argument ( '--tasks' , '-t' , nargs = '*' , help = 'A list of task IDs on which to check status' )
parser . add_argument ( '--attempts' , nargs = '*' , help = 'A list of task attempts on which to check status' )
parser . add_argument ( '--users' , '-u' , nargs = '*' , default = [ ] , help = """Lists only those jobs which were submitted by the list of users.
          Use "*" to list jobs of any user.""" )
parser . add_argument ( '--status' , '-s' , nargs = '*' , default = [ 'RUNNING' ] , choices = [ 'RUNNING' , 'SUCCESS' , 'FAILURE' , 'CANCELED' , '*' ] , help = """Lists only those jobs which match the specified status(es).
          Choose from {'RUNNING', 'SUCCESS', 'FAILURE', 'CANCELED'}.
          Use "*" to list jobs of any status.""" , metavar = 'STATUS' )
parser . add_argument ( '--age' , help = """List only those jobs newer than the specified age. Ages can be
          listed using a number followed by a unit. Supported units are
          s (seconds), m (minutes), h (hours), d (days), w (weeks).
          For example: '7d' (7 days). Bare numbers are treated as UTC.""" )
parser . add_argument ( '--label' , nargs = '*' , action = param_util . ListParamAction , default = [ ] , help = 'User labels to match. Tasks returned must match all labels.' , metavar = 'KEY=VALUE' )
parser . add_argument ( '--poll-interval' , default = 10 , type = int , help = 'Polling interval (in seconds) for checking job status ' 'when --wait is set.' )
parser . add_argument ( '--wait' , action = 'store_true' , help = 'Wait until jobs have all completed.' )
parser . add_argument ( '--limit' , default = 0 , type = int , help = 'The maximum number of tasks to list. The default is unlimited.' )
parser . add_argument ( '--format' , choices = [ 'text' , 'json' , 'yaml' , 'provider-json' ] , help = 'Set the output format.' )
output_style = parser . add_mutually_exclusive_group ( )
output_style . add_argument ( '--full' , '-f' , action = 'store_true' , help = 'Display output with full task information' ' and input parameters.' )
output_style . add_argument ( '--summary' , action = 'store_true' , help = 'Display a summary of the results, grouped by (job, status).' )
# Shared arguments between the " google " and " google - v2 " providers
google_common = parser . add_argument_group ( title = 'google-common' , description = 'Options common to the "google" and "google-v2" providers' )
google_common . add_argument ( '--project' , help = 'Cloud project ID in which to find and delete the job(s)' )
return provider_base . parse_args ( parser , { 'google' : [ 'project' ] , 'google-v2' : [ 'project' ] , 'test-fails' : [ ] , 'local' : [ ] , } , sys . argv [ 1 : ] )


signature:
def dstat_job_producer ( provider , statuses , user_ids = None , job_ids = None , job_names = None , task_ids = None , task_attempts = None , labels = None , create_time_min = None , create_time_max = None , max_tasks = 0 , full_output = False , summary_output = False , poll_interval = 0 , raw_format = False ) :
    """Generate jobs as lists of task dicts ready for formatting / output .
 Args :
 provider : an instantiated dsub provider .
 statuses : a set of status strings that eligible jobs may match .
 user _ ids : a set of user strings that eligible jobs may match .
 job _ ids : a set of job - id strings eligible jobs may match .
 job _ names : a set of job - name strings eligible jobs may match .
 task _ ids : a set of task - id strings eligible tasks may match .
 task _ attempts : a set of task - attempt strings eligible tasks may match .
 labels : set of LabelParam that all tasks must match .
 create _ time _ min : a timezone - aware datetime value for the earliest create
 time of a task , inclusive .
 create _ time _ max : a timezone - aware datetime value for the most recent create
 time of a task , inclusive .
 max _ tasks : ( int ) maximum number of tasks to return per dstat job lookup .
 full _ output : ( bool ) return all dsub fields .
 summary _ output : ( bool ) return a summary of the job list .
 poll _ interval : ( int ) wait time between poll events , dstat will poll jobs
 until all jobs succeed or fail . Set to zero to disable
 polling and return after the first lookup .
 raw _ format : ( bool ) set True to prevent dsub from normalizing the task dict ,
 this defaults to False and should only be set True if a
 provider - specific view of tasks is absolutely required .
 ( NB : provider interfaces change over time , no transition path
 will be provided for users depending on this flag ) .
 Yields :
 lists of task dictionaries - each list representing a dstat poll event ."""

implementation:
some_job_running = True
while some_job_running : # Get a batch of jobs .
    tasks = provider . lookup_job_tasks ( statuses , user_ids = user_ids , job_ids = job_ids , job_names = job_names , task_ids = task_ids , task_attempts = task_attempts , labels = labels , create_time_min = create_time_min , create_time_max = create_time_max , max_tasks = max_tasks , page_size = max_tasks )
    some_job_running = False
    formatted_tasks = [ ]
    for task in tasks :
        if 0 < max_tasks <= len ( formatted_tasks ) :
            break
        # Format tasks as specified .
        if raw_format :
            formatted_tasks . append ( task . raw_task_data ( ) )
        else :
            formatted_tasks . append ( _prepare_row ( task , full_output , summary_output ) )
        # Determine if any of the jobs are running .
        if task . get_field ( 'task-status' ) == 'RUNNING' :
            some_job_running = True
    # Yield the tasks and determine if the loop should continue .
    yield formatted_tasks
    if poll_interval and some_job_running :
        time . sleep ( poll_interval )
    else :
        break


signature:
def lookup_job_tasks ( provider , statuses , user_ids = None , job_ids = None , job_names = None , task_ids = None , task_attempts = None , labels = None , create_time_min = None , create_time_max = None , max_tasks = 0 , page_size = 0 , summary_output = False ) :
    """Generate formatted jobs individually , in order of create - time .
 Args :
 provider : an instantiated dsub provider .
 statuses : a set of status strings that eligible jobs may match .
 user _ ids : a set of user strings that eligible jobs may match .
 job _ ids : a set of job - id strings eligible jobs may match .
 job _ names : a set of job - name strings eligible jobs may match .
 task _ ids : a set of task - id strings eligible tasks may match .
 task _ attempts : a set of task - attempt strings eligible tasks may match .
 labels : set of LabelParam that all tasks must match .
 create _ time _ min : a timezone - aware datetime value for the earliest create
 time of a task , inclusive .
 create _ time _ max : a timezone - aware datetime value for the most recent create
 time of a task , inclusive .
 max _ tasks : ( int ) maximum number of tasks to return per dstat job lookup .
 page _ size : the page size to use for each query to the backend . May be
 ignored by some provider implementations .
 summary _ output : ( bool ) summarize the job list .
 Yields :
 Individual task dictionaries with associated metadata"""

implementation:
tasks_generator = provider . lookup_job_tasks ( statuses , user_ids = user_ids , job_ids = job_ids , job_names = job_names , task_ids = task_ids , task_attempts = task_attempts , labels = labels , create_time_min = create_time_min , create_time_max = create_time_max , max_tasks = max_tasks , page_size = page_size )
# Yield formatted tasks .
for task in tasks_generator :
    yield _prepare_row ( task , True , summary_output )


signature:
def prepare_output ( self , row ) :
    """Convert types of task fields ."""

implementation:
date_fields = [ 'last-update' , 'create-time' , 'start-time' , 'end-time' ]
int_fields = [ 'task-attempt' ]
for col in date_fields :
    if col in row :
        row [ col ] = self . default_format_date ( row [ col ] )
for col in int_fields :
    if col in row and row [ col ] is not None :
        row [ col ] = int ( row [ col ] )
return row


signature:
def trim_display_field ( self , value , max_length ) :
    """Return a value for display ; if longer than max length , use ellipsis ."""

implementation:
if not value :
    return ''
if len ( value ) > max_length :
    return value [ : max_length - 3 ] + '...'
return value


signature:
def format_pairs ( self , values ) :
    """Returns a string of comma - delimited key = value pairs ."""

implementation: return ', ' . join ( '%s=%s' % ( key , value ) for key , value in sorted ( values . items ( ) ) )


signature:
def string_presenter ( self , dumper , data ) :
    """Presenter to force yaml . dump to use multi - line string style ."""

implementation:
if '\n' in data :
    return dumper . represent_scalar ( 'tag:yaml.org,2002:str' , data , style = '|' )
else :
    return dumper . represent_scalar ( 'tag:yaml.org,2002:str' , data )


signature:
def get_zones ( input_list ) :
    """Returns a list of zones based on any wildcard input .
 This function is intended to provide an easy method for producing a list
 of desired zones for a pipeline to run in .
 The Pipelines API default zone list is " any zone " . The problem with
 " any zone " is that it can lead to incurring Cloud Storage egress charges
 if the GCE zone selected is in a different region than the GCS bucket .
 See https : / / cloud . google . com / storage / pricing # network - egress .
 A user with a multi - region US bucket would want to pipelines to run in
 a " us - * " zone .
 A user with a regional bucket in US would want to restrict pipelines to
 run in a zone in that region .
 Rarely does the specific zone matter for a pipeline .
 This function allows for a simple short - hand such as :
 [ " us - * " ]
 [ " us - central1 - * " ]
 These examples will expand out to the full list of US and us - central1 zones
 respectively .
 Args :
 input _ list : list of zone names / patterns
 Returns :
 A list of zones , with any wildcard zone specifications expanded ."""

implementation:
if not input_list :
    return [ ]
output_list = [ ]
for zone in input_list :
    if zone . endswith ( '*' ) :
        prefix = zone [ : - 1 ]
        output_list . extend ( [ z for z in _ZONES if z . startswith ( prefix ) ] )
    else :
        output_list . append ( zone )
return output_list


signature:
def build_pipeline_labels ( job_metadata , task_metadata , task_id_pattern = None ) :
    """Build a set ( ) of standard job and task labels .
 Args :
 job _ metadata : Job metadata , such as job - id , job - name , and user - id .
 task _ metadata : Task metadata , such as the task - id .
 task _ id _ pattern : A pattern for the task - id value , such as " task - % d " ; the
 original google label values could not be strictly numeric , so " task - "
 was prepended .
 Returns :
 A set of standard dsub Label ( ) objects to attach to a pipeline ."""

implementation:
labels = { Label ( name , job_metadata [ name ] ) for name in [ 'job-name' , 'job-id' , 'user-id' , 'dsub-version' ] }
task_id = task_metadata . get ( 'task-id' )
if task_id is not None : # Check for None ( as 0 is conceivably valid )
    if task_id_pattern :
        task_id = task_id_pattern % task_id
    labels . add ( Label ( 'task-id' , str ( task_id ) ) )
task_attempt = task_metadata . get ( 'task-attempt' )
if task_attempt is not None :
    labels . add ( Label ( 'task-attempt' , str ( task_attempt ) ) )
return labels


signature:
def prepare_job_metadata ( script , job_name , user_id , create_time ) :
    """Returns a dictionary of metadata fields for the job ."""

implementation:
# The name of the pipeline gets set into the ephemeralPipeline . name as - is .
# The default name of the pipeline is the script name
# The name of the job is derived from the job _ name and gets set as a
# ' job - name ' label ( and so the value must be normalized ) .
if job_name :
    pipeline_name = job_name
    job_name_value = job_model . convert_to_label_chars ( job_name )
else :
    pipeline_name = os . path . basename ( script )
    job_name_value = job_model . convert_to_label_chars ( pipeline_name . split ( '.' , 1 ) [ 0 ] )
# The user - id will get set as a label
user_id = job_model . convert_to_label_chars ( user_id )
# Now build the job - id . We want the job - id to be expressive while also
# having a low - likelihood of collisions .
# For expressiveness , we :
# * use the job name ( truncated at 10 characters ) .
# * insert the user - id
# * add a datetime value
# To have a high likelihood of uniqueness , the datetime value is out to
# hundredths of a second .
# The full job - id is :
# < job - name > - - < user - id > - - < timestamp >
job_id = '%s--%s--%s' % ( job_name_value [ : 10 ] , user_id , create_time . strftime ( '%y%m%d-%H%M%S-%f' ) [ : 16 ] )
# Standard version is MAJOR . MINOR ( . PATCH ) . This will convert the version
# string to " vMAJOR - MINOR ( - PATCH ) " . Example ; " 0.1.0 " - > " v0-1-0 " .
version = job_model . convert_to_label_chars ( 'v%s' % DSUB_VERSION )
return { 'pipeline-name' : pipeline_name , 'job-name' : job_name_value , 'job-id' : job_id , 'user-id' : user_id , 'dsub-version' : version , }


signature:
def parse_rfc3339_utc_string ( rfc3339_utc_string ) :
    """Converts a datestamp from RFC3339 UTC to a datetime .
 Args :
 rfc3339 _ utc _ string : a datetime string in RFC3339 UTC " Zulu " format
 Returns :
 A datetime ."""

implementation:
# The timestamp from the Google Operations are all in RFC3339 format , but
# they are sometimes formatted to millisconds , microseconds , sometimes
# nanoseconds , and sometimes only seconds :
# * 2016-11-14T23:05:56Z
# * 2016-11-14T23:05:56.010Z
# * 2016-11-14T23:05:56.010429Z
# * 2016-11-14T23:05:56.010429380Z
m = re . match ( r'(\d{4})-(\d{2})-(\d{2})T(\d{2}):(\d{2}):(\d{2}).?(\d*)Z' , rfc3339_utc_string )
# It would be unexpected to get a different date format back from Google .
# If we raise an exception here , we can break people completely .
# Instead , let ' s just return None and people can report that some dates
# are not showing up .
# We might reconsider this approach in the future ; it was originally
# established when dates were only used for display .
if not m :
    return None
groups = m . groups ( )
if len ( groups [ 6 ] ) not in ( 0 , 3 , 6 , 9 ) :
    return None
# Create a UTC datestamp from parsed components
# 1 - Turn components 0-5 from strings to integers
# 2 - If the last component does not exist , set it to 0.
# If it does exist , make sure to interpret it as milliseconds .
g = [ int ( val ) for val in groups [ : 6 ] ]
fraction = groups [ 6 ]
if not fraction :
    micros = 0
elif len ( fraction ) == 3 :
    micros = int ( fraction ) * 1000
elif len ( fraction ) == 6 :
    micros = int ( fraction )
elif len ( fraction ) == 9 : # When nanoseconds are provided , we round
    micros = int ( round ( int ( fraction ) / 1000 ) )
else :
    assert False , 'Fraction length not 0, 6, or 9: {}' . len ( fraction )
try :
    return datetime ( g [ 0 ] , g [ 1 ] , g [ 2 ] , g [ 3 ] , g [ 4 ] , g [ 5 ] , micros , tzinfo = pytz . utc )
except ValueError as e :
    assert False , 'Could not parse RFC3339 datestring: {} exception: {}' . format ( rfc3339_utc_string , e )


signature:
def get_operation_full_job_id ( op ) :
    """Returns the job - id or job - id . task - id for the operation ."""

implementation:
job_id = op . get_field ( 'job-id' )
task_id = op . get_field ( 'task-id' )
if task_id :
    return '%s.%s' % ( job_id , task_id )
else :
    return job_id


signature:
def cancel ( batch_fn , cancel_fn , ops ) :
    """Cancel operations .
 Args :
 batch _ fn : API - specific batch function .
 cancel _ fn : API - specific cancel function .
 ops : A list of operations to cancel .
 Returns :
 A list of operations canceled and a list of error messages ."""

implementation:
# Canceling many operations one - by - one can be slow .
# The Pipelines API doesn ' t directly support a list of operations to cancel ,
# but the requests can be performed in batch .
canceled_ops = [ ]
error_messages = [ ]
max_batch = 256
total_ops = len ( ops )
for first_op in range ( 0 , total_ops , max_batch ) :
    batch_canceled , batch_messages = _cancel_batch ( batch_fn , cancel_fn , ops [ first_op : first_op + max_batch ] )
    canceled_ops . extend ( batch_canceled )
    error_messages . extend ( batch_messages )
return canceled_ops , error_messages


signature:
def retry_api_check ( exception ) :
    """Return True if we should retry . False otherwise .
 Args :
 exception : An exception to test for transience .
 Returns :
 True if we should retry . False otherwise ."""

implementation:
if isinstance ( exception , apiclient . errors . HttpError ) :
    if exception . resp . status in TRANSIENT_HTTP_ERROR_CODES :
        _print_error ( 'Retrying...' )
        return True
if isinstance ( exception , socket . error ) :
    if exception . errno in TRANSIENT_SOCKET_ERROR_CODES :
        _print_error ( 'Retrying...' )
        return True
if isinstance ( exception , oauth2client . client . AccessTokenRefreshError ) :
    _print_error ( 'Retrying...' )
    return True
# For a given installation , this could be a permanent error , but has only
# been observed as transient .
if isinstance ( exception , SSLError ) :
    _print_error ( 'Retrying...' )
    return True
# This has been observed as a transient error :
# ServerNotFoundError : Unable to find the server at genomics . googleapis . com
if isinstance ( exception , ServerNotFoundError ) :
    _print_error ( 'Retrying...' )
    return True
return False


signature:
def retry_auth_check ( exception ) :
    """Specific check for auth error codes .
 Return True if we should retry .
 False otherwise .
 Args :
 exception : An exception to test for transience .
 Returns :
 True if we should retry . False otherwise ."""

implementation:
if isinstance ( exception , apiclient . errors . HttpError ) :
    if exception . resp . status in HTTP_AUTH_ERROR_CODES :
        _print_error ( 'Retrying...' )
        return True
return False


signature:
def setup_service ( api_name , api_version , credentials = None ) :
    """Configures genomics API client .
 Args :
 api _ name : Name of the Google API ( for example : " genomics " )
 api _ version : Version of the API ( for example : " v2alpha1 " )
 credentials : Credentials to be used for the gcloud API calls .
 Returns :
 A configured Google Genomics API client with appropriate credentials ."""

implementation:
if not credentials :
    credentials = oauth2client . client . GoogleCredentials . get_application_default ( )
return apiclient . discovery . build ( api_name , api_version , credentials = credentials )


signature:
def execute ( api ) :
    """Executes operation .
 Args :
 api : The base API object
 Returns :
 A response body object"""

implementation:
try :
    return api . execute ( )
except Exception as exception :
    now = datetime . now ( ) . strftime ( '%Y-%m-%d %H:%M:%S.%f' )
    _print_error ( '%s: Exception %s: %s' % ( now , type ( exception ) . __name__ , str ( exception ) ) )
    # Re - raise exception to be handled by retry logic
    raise exception


signature:
def _eval_arg_type ( arg_type , T = Any , arg = None , sig = None ) :
    """Returns a type from a snippit of python source . Should normally be
 something just like ' str ' or ' Object ' .
 arg _ type the source to be evaluated
 T the default type
 arg context of where this type was extracted
 sig context from where the arg was extracted
 Returns a type or a Type"""

implementation:
try :
    T = eval ( arg_type )
except Exception as e :
    raise ValueError ( 'The type of {0} could not be evaluated in {1} for {2}: {3}' . format ( arg_type , arg , sig , text_type ( e ) ) )
else :
    if type ( T ) not in ( type , Type ) :
        raise TypeError ( '{0} is not a valid type in {1} for {2}' . format ( repr ( T ) , arg , sig ) )
    return T


signature:
def jsonify_status_code ( status_code , * args , ** kw ) :
    """Returns a jsonified response with the specified HTTP status code .
 The positional and keyword arguments are passed directly to the
 : func : ` flask . jsonify ` function which creates the response ."""

implementation:
is_batch = kw . pop ( 'is_batch' , False )
if is_batch :
    response = flask_make_response ( json . dumps ( * args , ** kw ) )
    response . mimetype = 'application/json'
    response . status_code = status_code
    return response
response = jsonify ( * args , ** kw )
response . status_code = status_code
return response


signature:
def register ( self , app , options , first_registration = False ) :
    """Called by : meth : ` Flask . register _ blueprint ` to register a blueprint
 on the application . This can be overridden to customize the register
 behavior . Keyword arguments from
 : func : ` ~ flask . Flask . register _ blueprint ` are directly forwarded to this
 method in the ` options ` dictionary ."""

implementation:
self . jsonrpc_site = options . get ( 'jsonrpc_site' )
self . _got_registered_once = True
state = self . make_setup_state ( app , options , first_registration )
if self . has_static_folder and not self . name + '.static' in state . app . view_functions . keys ( ) :
    state . add_url_rule ( self . static_url_path + '/<path:filename>' , view_func = self . send_static_file , endpoint = 'static' )
for deferred in self . deferred_functions :
    deferred ( state )


signature:
def send_payload ( self , params ) :
    """Performs the actual sending action and returns the result"""

implementation:
data = json . dumps ( { 'jsonrpc' : self . version , 'method' : self . service_name , 'params' : params , 'id' : text_type ( uuid . uuid4 ( ) ) } )
data_binary = data . encode ( 'utf-8' )
url_request = Request ( self . service_url , data_binary , headers = self . headers )
return urlopen ( url_request ) . read ( )


signature:
def make_response ( self , rv ) :
    """Converts the return value from a view function to a real
 response object that is an instance of : attr : ` response _ class ` ."""

implementation:
status_or_headers = headers = None
if isinstance ( rv , tuple ) :
    rv , status_or_headers , headers = rv + ( None , ) * ( 3 - len ( rv ) )
if rv is None :
    raise ValueError ( 'View function did not return a response' )
if isinstance ( status_or_headers , ( dict , list ) ) :
    headers , status_or_headers = status_or_headers , None
D = json . loads ( extract_raw_data_request ( request ) )
if type ( D ) is list :
    raise InvalidRequestError ( 'JSON-RPC batch with decorator (make_response) not is supported' )
else :
    response_obj = self . empty_response ( version = D [ 'jsonrpc' ] )
    response_obj [ 'id' ] = D [ 'id' ]
    response_obj [ 'result' ] = rv
    response_obj . pop ( 'error' , None )
    rv = jsonify ( response_obj )
if status_or_headers is not None :
    if isinstance ( status_or_headers , string_types ) :
        rv . status = status_or_headers
    else :
        rv . status_code = status_or_headers
if headers :
    rv . headers . extend ( headers )
return rv


signature:
def json_rpc_format ( self ) :
    """Return the Exception data in a format for JSON - RPC"""

implementation:
error = { 'name' : text_type ( self . __class__ . __name__ ) , 'code' : self . code , 'message' : '{0}' . format ( text_type ( self . message ) ) , 'data' : self . data }
if current_app . config [ 'DEBUG' ] :
    import sys , traceback
    error [ 'stack' ] = traceback . format_exc ( )
    error [ 'executable' ] = sys . executable
return error


signature:
def from_file ( cls , file ) :
    """Try loading given config file .
 : param str file : full path to the config file to load"""

implementation:
if not os . path . exists ( file ) :
    raise ValueError ( "Config file not found." )
try :
    config_parser = configparser . ConfigParser ( )
    config_parser . read ( file )
    configuration = cls ( file , config_parser )
    if not configuration . check_config_sanity ( ) :
        raise ValueError ( "Error in config file." )
    else :
        return configuration
except configparser . Error :
    raise ValueError ( "Config file is invalid." )


signature:
def discover ( cls ) :
    """Make a guess about the config file location an try loading it ."""

implementation:
file = os . path . join ( Config . config_dir , Config . config_name )
return cls . from_file ( file )


signature:
def create_config ( cls , cfgfile , nick , twtfile , twturl , disclose_identity , add_news ) :
    """Create a new config file at the default location .
 : param str cfgfile : path to the config file
 : param str nick : nickname to use for own tweets
 : param str twtfile : path to the local twtxt file
 : param str twturl : URL to the remote twtxt file
 : param bool disclose _ identity : if true the users id will be disclosed
 : param bool add _ news : if true follow twtxt news feed"""

implementation:
cfgfile_dir = os . path . dirname ( cfgfile )
if not os . path . exists ( cfgfile_dir ) :
    os . makedirs ( cfgfile_dir )
cfg = configparser . ConfigParser ( )
cfg . add_section ( "twtxt" )
cfg . set ( "twtxt" , "nick" , nick )
cfg . set ( "twtxt" , "twtfile" , twtfile )
cfg . set ( "twtxt" , "twturl" , twturl )
cfg . set ( "twtxt" , "disclose_identity" , str ( disclose_identity ) )
cfg . set ( "twtxt" , "character_limit" , "140" )
cfg . set ( "twtxt" , "character_warning" , "140" )
cfg . add_section ( "following" )
if add_news :
    cfg . set ( "following" , "twtxt" , "https://buckket.org/twtxt_news.txt" )
conf = cls ( cfgfile , cfg )
conf . write_config ( )
return conf


signature:
def write_config ( self ) :
    """Writes ` self . cfg ` to ` self . config _ file ` ."""

implementation:
with open ( self . config_file , "w" ) as config_file :
    self . cfg . write ( config_file )


signature:
def following ( self ) :
    """A : class : ` list ` of all : class : ` Source ` objects ."""

implementation:
following = [ ]
try :
    for ( nick , url ) in self . cfg . items ( "following" ) :
        source = Source ( nick , url )
        following . append ( source )
except configparser . NoSectionError as e :
    logger . debug ( e )
return following


signature:
def options ( self ) :
    """A : class : ` dict ` of all config options ."""

implementation:
try :
    return dict ( self . cfg . items ( "twtxt" ) )
except configparser . NoSectionError as e :
    logger . debug ( e )
    return { }


signature:
def add_source ( self , source ) :
    """Adds a new : class : ` Source ` to the config ’ s following section ."""

implementation:
if not self . cfg . has_section ( "following" ) :
    self . cfg . add_section ( "following" )
self . cfg . set ( "following" , source . nick , source . url )
self . write_config ( )


signature:
def get_source_by_nick ( self , nick ) :
    """Returns the : class : ` Source ` of the given nick .
 : param str nick : nickname for which will be searched in the config"""

implementation:
url = self . cfg . get ( "following" , nick , fallback = None )
return Source ( nick , url ) if url else None


signature:
def remove_source_by_nick ( self , nick ) :
    """Removes a : class : ` Source ` form the config ’ s following section .
 : param str nick : nickname for which will be searched in the config"""

implementation:
if not self . cfg . has_section ( "following" ) :
    return False
ret_val = self . cfg . remove_option ( "following" , nick )
self . write_config ( )
return ret_val


signature:
def build_default_map ( self ) :
    """Maps config options to the default values used by click , returns : class : ` dict ` ."""

implementation:
default_map = { "following" : { "check" : self . check_following , "timeout" : self . timeout , "porcelain" : self . porcelain , } , "tweet" : { "twtfile" : self . twtfile , } , "timeline" : { "pager" : self . use_pager , "cache" : self . use_cache , "limit" : self . limit_timeline , "timeout" : self . timeout , "sorting" : self . sorting , "porcelain" : self . porcelain , "twtfile" : self . twtfile , "update_interval" : self . timeline_update_interval , } , "view" : { "pager" : self . use_pager , "cache" : self . use_cache , "limit" : self . limit_timeline , "timeout" : self . timeout , "sorting" : self . sorting , "porcelain" : self . porcelain , "update_interval" : self . timeline_update_interval , } }
return default_map


signature:
def check_config_sanity ( self ) :
    """Checks if the given values in the config file are sane ."""

implementation:
is_sane = True
# This extracts some properties which cannot be checked like " nick " ,
# but it is definitely better than writing the property names as a
# string literal .
properties = [ property_name for property_name , obj in self . __class__ . __dict__ . items ( ) if isinstance ( obj , property ) ]
for property_name in properties :
    try :
        getattr ( self , property_name )
    except ValueError as e :
        click . echo ( "✗ Config error on {0} - {1}" . format ( property_name , e ) )
        is_sane = False
return is_sane


signature:
def validate_config_key ( ctx , param , value ) :
    """Validate a configuration key according to ` section . item ` ."""

implementation:
if not value :
    return value
try :
    section , item = value . split ( "." , 1 )
except ValueError :
    raise click . BadArgumentUsage ( "Given key does not contain a section name." )
else :
    return section , item


signature:
def expand_mentions ( text , embed_names = True ) :
    """Searches the given text for mentions and expands them .
 For example :
 " @ source . nick " will be expanded to " @ < source . nick source . url > " ."""

implementation:
if embed_names :
    mention_format = "@<{name} {url}>"
else :
    mention_format = "@<{url}>"
def handle_mention ( match ) :
    source = get_source_by_name ( match . group ( 1 ) )
    if source is None :
        return "@{0}" . format ( match . group ( 1 ) )
    return mention_format . format ( name = source . nick , url = source . url )
return short_mention_re . sub ( handle_mention , text )


signature:
def format_mentions ( text , format_callback = format_mention ) :
    """Searches the given text for mentions generated by ` expand _ mention ( ) ` and returns a human - readable form .
 For example :
 " @ < bob http : / / example . org / twtxt . txt > " will result in " @ bob "
 If you follow a source : source . nick will be bold
 If you are the mentioned source : source . nick will be bold and coloured
 If nothing from the above is true : nick will be unstyled
 If nothing from the above is true and nick is not given : url will be used"""

implementation:
def handle_mention ( match ) :
    name , url = match . groups ( )
    return format_callback ( name , url )
return mention_re . sub ( handle_mention , text )


signature:
def make_aware ( dt ) :
    """Appends tzinfo and assumes UTC , if datetime object has no tzinfo already ."""

implementation: return dt if dt . tzinfo else dt . replace ( tzinfo = timezone . utc )


signature:
def parse_tweets ( raw_tweets , source , now = None ) :
    """Parses a list of raw tweet lines from a twtxt file
 and returns a list of : class : ` Tweet ` objects .
 : param list raw _ tweets : list of raw tweet lines
 : param Source source : the source of the given tweets
 : param Datetime now : the current datetime
 : returns : a list of parsed tweets : class : ` Tweet ` objects
 : rtype : list"""

implementation:
if now is None :
    now = datetime . now ( timezone . utc )
tweets = [ ]
for line in raw_tweets :
    try :
        tweet = parse_tweet ( line , source , now )
    except ( ValueError , OverflowError ) as e :
        logger . debug ( "{0} - {1}" . format ( source . url , e ) )
    else :
        tweets . append ( tweet )
return tweets


signature:
def parse_tweet ( raw_tweet , source , now = None ) :
    """Parses a single raw tweet line from a twtxt file
 and returns a : class : ` Tweet ` object .
 : param str raw _ tweet : a single raw tweet line
 : param Source source : the source of the given tweet
 : param Datetime now : the current datetime
 : returns : the parsed tweet
 : rtype : Tweet"""

implementation:
if now is None :
    now = datetime . now ( timezone . utc )
raw_created_at , text = raw_tweet . split ( "\t" , 1 )
created_at = parse_iso8601 ( raw_created_at )
if created_at > now :
    raise ValueError ( "Tweet is from the future" )
return Tweet ( click . unstyle ( text . strip ( ) ) , created_at , source )


signature:
def from_file ( cls , file , * args , ** kwargs ) :
    """Try loading given cache file ."""

implementation:
try :
    cache = shelve . open ( file )
    return cls ( file , cache , * args , ** kwargs )
except OSError as e :
    logger . debug ( "Loading {0} failed" . format ( file ) )
    raise e


signature:
def discover ( cls , * args , ** kwargs ) :
    """Make a guess about the cache file location an try loading it ."""

implementation:
file = os . path . join ( Cache . cache_dir , Cache . cache_name )
return cls . from_file ( file , * args , ** kwargs )


signature:
def is_cached ( self , url ) :
    """Checks if specified URL is cached ."""

implementation:
try :
    return True if url in self . cache else False
except TypeError :
    return False


signature:
def add_tweets ( self , url , last_modified , tweets ) :
    """Adds new tweets to the cache ."""

implementation:
try :
    self . cache [ url ] = { "last_modified" : last_modified , "tweets" : tweets }
    self . mark_updated ( )
    return True
except TypeError :
    return False


signature:
def get_tweets ( self , url , limit = None ) :
    """Retrieves tweets from the cache ."""

implementation:
try :
    tweets = self . cache [ url ] [ "tweets" ]
    self . mark_updated ( )
    return sorted ( tweets , reverse = True ) [ : limit ]
except KeyError :
    return [ ]


signature:
def remove_tweets ( self , url ) :
    """Tries to remove cached tweets ."""

implementation:
try :
    del self . cache [ url ]
    self . mark_updated ( )
    return True
except KeyError :
    return False


signature:
def cli ( ctx , config , verbose ) :
    """Decentralised , minimalist microblogging service for hackers ."""

implementation:
init_logging ( debug = verbose )
if ctx . invoked_subcommand == "quickstart" :
    return
# Skip initializing config file
try :
    if config :
        conf = Config . from_file ( config )
    else :
        conf = Config . discover ( )
except ValueError as e :
    if "Error in config file." in str ( e ) :
        click . echo ( "✗ Please correct the errors mentioned above an run twtxt again." )
    else :
        click . echo ( "✗ Config file not found or not readable. You may want to run twtxt quickstart." )
    sys . exit ( )
ctx . default_map = conf . build_default_map ( )
ctx . obj = { 'conf' : conf }


signature:
def tweet ( ctx , created_at , twtfile , text ) :
    """Append a new tweet to your twtxt file ."""

implementation:
text = expand_mentions ( text )
tweet = Tweet ( text , created_at ) if created_at else Tweet ( text )
pre_tweet_hook = ctx . obj [ "conf" ] . pre_tweet_hook
if pre_tweet_hook :
    run_pre_tweet_hook ( pre_tweet_hook , ctx . obj [ "conf" ] . options )
if not add_local_tweet ( tweet , twtfile ) :
    click . echo ( "✗ Couldn’t write to file." )
else :
    post_tweet_hook = ctx . obj [ "conf" ] . post_tweet_hook
    if post_tweet_hook :
        run_post_tweet_hook ( post_tweet_hook , ctx . obj [ "conf" ] . options )


signature:
def timeline ( ctx , pager , limit , twtfile , sorting , timeout , porcelain , source , cache , force_update ) :
    """Retrieve your personal timeline ."""

implementation:
if source :
    source_obj = ctx . obj [ "conf" ] . get_source_by_nick ( source )
    if not source_obj :
        logger . debug ( "Not following {0}, trying as URL" . format ( source ) )
        source_obj = Source ( source , source )
    sources = [ source_obj ]
else :
    sources = ctx . obj [ "conf" ] . following
tweets = [ ]
if cache :
    try :
        with Cache . discover ( update_interval = ctx . obj [ "conf" ] . timeline_update_interval ) as cache :
            force_update = force_update or not cache . is_valid
            if force_update :
                tweets = get_remote_tweets ( sources , limit , timeout , cache )
            else :
                logger . debug ( "Multiple calls to 'timeline' within {0} seconds. Skipping update" . format ( cache . update_interval ) )
                # Behold , almighty list comprehensions ! ( I might have gone overboard here . . . )
                tweets = list ( chain . from_iterable ( [ cache . get_tweets ( source . url ) for source in sources ] ) )
    except OSError as e :
        logger . debug ( e )
        tweets = get_remote_tweets ( sources , limit , timeout )
else :
    tweets = get_remote_tweets ( sources , limit , timeout )
if twtfile and not source :
    source = Source ( ctx . obj [ "conf" ] . nick , ctx . obj [ "conf" ] . twturl , file = twtfile )
    tweets . extend ( get_local_tweets ( source , limit ) )
if not tweets :
    return
tweets = sort_and_truncate_tweets ( tweets , sorting , limit )
if pager :
    click . echo_via_pager ( style_timeline ( tweets , porcelain ) )
else :
    click . echo ( style_timeline ( tweets , porcelain ) )


signature:
def following ( ctx , check , timeout , porcelain ) :
    """Return the list of sources you ’ re following ."""

implementation:
sources = ctx . obj [ 'conf' ] . following
if check :
    sources = get_remote_status ( sources , timeout )
    for ( source , status ) in sources :
        click . echo ( style_source_with_status ( source , status , porcelain ) )
else :
    sources = sorted ( sources , key = lambda source : source . nick )
    for source in sources :
        click . echo ( style_source ( source , porcelain ) )


signature:
def follow ( ctx , nick , url , force ) :
    """Add a new source to your followings ."""

implementation:
source = Source ( nick , url )
sources = ctx . obj [ 'conf' ] . following
if not force :
    if source . nick in ( source . nick for source in sources ) :
        click . confirm ( "➤ You’re already following {0}. Overwrite?" . format ( click . style ( source . nick , bold = True ) ) , default = False , abort = True )
    _ , status = get_remote_status ( [ source ] ) [ 0 ]
    if not status or status . status_code != 200 :
        click . confirm ( "➤ The feed of {0} at {1} is not available. Follow anyway?" . format ( click . style ( source . nick , bold = True ) , click . style ( source . url , bold = True ) ) , default = False , abort = True )
ctx . obj [ 'conf' ] . add_source ( source )
click . echo ( "✓ You’re now following {0}." . format ( click . style ( source . nick , bold = True ) ) )


signature:
def unfollow ( ctx , nick ) :
    """Remove an existing source from your followings ."""

implementation:
source = ctx . obj [ 'conf' ] . get_source_by_nick ( nick )
try :
    with Cache . discover ( ) as cache :
        cache . remove_tweets ( source . url )
except OSError as e :
    logger . debug ( e )
ret_val = ctx . obj [ 'conf' ] . remove_source_by_nick ( nick )
if ret_val :
    click . echo ( "✓ You’ve unfollowed {0}." . format ( click . style ( source . nick , bold = True ) ) )
else :
    click . echo ( "✗ You’re not following {0}." . format ( click . style ( nick , bold = True ) ) )


signature:
def quickstart ( ) :
    """Quickstart wizard for setting up twtxt ."""

implementation:
width = click . get_terminal_size ( ) [ 0 ]
width = width if width <= 79 else 79
click . secho ( "twtxt - quickstart" , fg = "cyan" )
click . secho ( "==================" , fg = "cyan" )
click . echo ( )
help_text = "This wizard will generate a basic configuration file for twtxt with all mandatory options set. " "You can change all of these later with either twtxt itself or by editing the config file manually. " "Have a look at the docs to get information about the other available options and their meaning."
click . echo ( textwrap . fill ( help_text , width ) )
click . echo ( )
nick = click . prompt ( "➤ Please enter your desired nick" , default = os . environ . get ( "USER" , "" ) )
def overwrite_check ( path ) :
    if os . path . isfile ( path ) :
        click . confirm ( "➤ '{0}' already exists. Overwrite?" . format ( path ) , abort = True )
cfgfile = click . prompt ( "➤ Please enter the desired location for your config file" , os . path . join ( Config . config_dir , Config . config_name ) , type = click . Path ( readable = True , writable = True , file_okay = True ) )
cfgfile = os . path . expanduser ( cfgfile )
overwrite_check ( cfgfile )
twtfile = click . prompt ( "➤ Please enter the desired location for your twtxt file" , os . path . expanduser ( "~/twtxt.txt" ) , type = click . Path ( readable = True , writable = True , file_okay = True ) )
twtfile = os . path . expanduser ( twtfile )
overwrite_check ( twtfile )
twturl = click . prompt ( "➤ Please enter the URL your twtxt file will be accessible from" , default = "https://example.org/twtxt.txt" )
disclose_identity = click . confirm ( "➤ Do you want to disclose your identity? Your nick and URL will be shared when " "making HTTP requests" , default = False )
click . echo ( )
add_news = click . confirm ( "➤ Do you want to follow the twtxt news feed?" , default = True )
conf = Config . create_config ( cfgfile , nick , twtfile , twturl , disclose_identity , add_news )
twtfile_dir = os . path . dirname ( twtfile )
if not os . path . exists ( twtfile_dir ) :
    os . makedirs ( twtfile_dir )
open ( twtfile , "a" ) . close ( )
click . echo ( )
click . echo ( "✓ Created config file at '{0}'." . format ( click . format_filename ( conf . config_file ) ) )
click . echo ( "✓ Created twtxt file at '{0}'." . format ( click . format_filename ( twtfile ) ) )


signature:
def config ( ctx , key , value , remove , edit ) :
    """Get or set config item ."""

implementation:
conf = ctx . obj [ "conf" ]
if not edit and not key :
    raise click . BadArgumentUsage ( "You have to specify either a key or use --edit." )
if edit :
    return click . edit ( filename = conf . config_file )
if remove :
    try :
        conf . cfg . remove_option ( key [ 0 ] , key [ 1 ] )
    except Exception as e :
        logger . debug ( e )
    else :
        conf . write_config ( )
    return
if not value :
    try :
        click . echo ( conf . cfg . get ( key [ 0 ] , key [ 1 ] ) )
    except Exception as e :
        logger . debug ( e )
    return
if not conf . cfg . has_section ( key [ 0 ] ) :
    conf . cfg . add_section ( key [ 0 ] )
conf . cfg . set ( key [ 0 ] , key [ 1 ] , value )
conf . write_config ( )


signature:
def relative_datetime ( self ) :
    """Return human - readable relative time string ."""

implementation:
now = datetime . now ( timezone . utc )
tense = "from now" if self . created_at > now else "ago"
return "{0} {1}" . format ( humanize . naturaldelta ( now - self . created_at ) , tense )


signature:
def save ( url , * args , ** kwargs ) :
    """Parse the options , set defaults and then fire up PhantomJS ."""

implementation:
device = heimdallDevice ( kwargs . get ( 'device' , None ) )
kwargs [ 'width' ] = kwargs . get ( 'width' , None ) or device . width
kwargs [ 'height' ] = kwargs . get ( 'height' , None ) or device . height
kwargs [ 'user_agent' ] = kwargs . get ( 'user_agent' , None ) or device . user_agent
screenshot_image = screenshot ( url , ** kwargs )
if kwargs . get ( 'optimize' ) :
    image = Image . open ( screenshot_image . path )
    image . save ( screenshot_image . path , optimize = True )
return screenshot_image


signature:
def screenshot ( url , * args , ** kwargs ) :
    """Call PhantomJS with the specified flags and options ."""

implementation:
phantomscript = os . path . join ( os . path . dirname ( __file__ ) , 'take_screenshot.js' )
directory = kwargs . get ( 'save_dir' , '/tmp' )
image_name = kwargs . get ( 'image_name' , None ) or _image_name_from_url ( url )
ext = kwargs . get ( 'format' , 'png' ) . lower ( )
save_path = os . path . join ( directory , image_name ) + '.' + ext
crop_to_visible = kwargs . get ( 'crop_to_visible' , False )
cmd_args = [ 'phantomjs' , '--ssl-protocol=any' , phantomscript , url , '--width' , str ( kwargs [ 'width' ] ) , '--height' , str ( kwargs [ 'height' ] ) , '--useragent' , str ( kwargs [ 'user_agent' ] ) , '--dir' , directory , '--ext' , ext , '--name' , str ( image_name ) , ]
if crop_to_visible :
    cmd_args . append ( '--croptovisible' )
# TODO :
# - quality
# - renderafter
# - maxexecutiontime
# - resourcetimeout
output = subprocess . Popen ( cmd_args , stdout = subprocess . PIPE ) . communicate ( ) [ 0 ]
return Screenshot ( save_path , directory , image_name + '.' + ext , ext )


signature:
def _image_name_from_url ( url ) :
    """Create a nice image name from the url ."""

implementation:
find = r'https?://|[^\w]'
replace = '_'
return re . sub ( find , replace , url ) . strip ( '_' )


signature:
def worker ( f ) :
    """Decorator . Abortable worker . If wrapped task will be cancelled by
 dispatcher , decorator will send ftp codes of successful interrupt .
 > > > @ worker
 . . . async def worker ( self , connection , rest ) :"""

implementation:
@ functools . wraps ( f )
async def wrapper ( cls , connection , rest ) :
    try :
        await f ( cls , connection , rest )
    except asyncio . CancelledError :
        connection . response ( "426" , "transfer aborted" )
        connection . response ( "226" , "abort successful" )
return wrapper


signature:
def get_permissions ( self , path ) :
    """Return nearest parent permission for ` path ` .
 : param path : path which permission you want to know
 : type path : : py : class : ` str ` or : py : class : ` pathlib . PurePosixPath `
 : rtype : : py : class : ` aioftp . Permission `"""

implementation:
path = pathlib . PurePosixPath ( path )
parents = filter ( lambda p : p . is_parent ( path ) , self . permissions )
perm = min ( parents , key = lambda p : len ( path . relative_to ( p . path ) . parts ) , default = Permission ( ) , )
return perm


signature:
def release ( self ) :
    """Release , incrementing the internal counter by one ."""

implementation:
if self . value is not None :
    self . value += 1
    if self . value > self . maximum_value :
        raise ValueError ( "Too many releases" )


signature:
async def start ( self , host = None , port = 0 , ** kwargs ) :
    """: py : func : ` asyncio . coroutine `
 Start server .
 : param host : ip address to bind for listening .
 : type host : : py : class : ` str `
 : param port : port number to bind for listening .
 : type port : : py : class : ` int `
 : param kwargs : keyword arguments , they passed to
 : py : func : ` asyncio . start _ server `"""

implementation:
self . _start_server_extra_arguments = kwargs
self . connections = { }
self . server_host = host
self . server_port = port
self . server = await asyncio . start_server ( self . dispatcher , host , port , ssl = self . ssl , ** self . _start_server_extra_arguments , )
for sock in self . server . sockets :
    if sock . family in ( socket . AF_INET , socket . AF_INET6 ) :
        host , port , * _ = sock . getsockname ( )
        if not self . server_port :
            self . server_port = port
        if not self . server_host :
            self . server_host = host
        logger . info ( "serving on %s:%s" , host , port )


signature:
async def close ( self ) :
    """: py : func : ` asyncio . coroutine `
 Shutdown the server and close all connections ."""

implementation:
self . server . close ( )
tasks = [ self . server . wait_closed ( ) ]
for connection in self . connections . values ( ) :
    connection . _dispatcher . cancel ( )
    tasks . append ( connection . _dispatcher )
logger . info ( "waiting for %d tasks" , len ( tasks ) )
await asyncio . wait ( tasks )


signature:
async def write_response ( self , stream , code , lines = "" , list = False ) :
    """: py : func : ` asyncio . coroutine `
 Complex method for sending response .
 : param stream : command connection stream
 : type stream : : py : class : ` aioftp . StreamIO `
 : param code : server response code
 : type code : : py : class : ` str `
 : param lines : line or lines , which are response information
 : type lines : : py : class : ` str ` or : py : class : ` collections . Iterable `
 : param list : if true , then lines will be sended without code prefix .
 This is useful for * * LIST * * FTP command and some others .
 : type list : : py : class : ` bool `"""

implementation:
lines = wrap_with_container ( lines )
write = functools . partial ( self . write_line , stream )
if list :
    head , * body , tail = lines
    await write ( code + "-" + head )
    for line in body :
        await write ( " " + line )
    await write ( code + " " + tail )
else :
    * body , tail = lines
    for line in body :
        await write ( code + "-" + line )
    await write ( code + " " + tail )


signature:
async def parse_command ( self , stream ) :
    """: py : func : ` asyncio . coroutine `
 Complex method for getting command .
 : param stream : connection steram
 : type stream : : py : class : ` asyncio . StreamIO `
 : return : ( code , rest )
 : rtype : ( : py : class : ` str ` , : py : class : ` str ` )"""

implementation:
line = await stream . readline ( )
if not line :
    raise ConnectionResetError
s = line . decode ( encoding = self . encoding ) . rstrip ( )
logger . info ( s )
cmd , _ , rest = s . partition ( " " )
return cmd . lower ( ) , rest


signature:
async def response_writer ( self , stream , response_queue ) :
    """: py : func : ` asyncio . coroutine `
 Worker for write _ response with current connection . Get data to response
 from queue , this is for right order of responses . Exits if received
 : py : class : ` None ` .
 : param stream : command connection stream
 : type connection : : py : class : ` aioftp . StreamIO `
 : param response _ queue :
 : type response _ queue : : py : class : ` asyncio . Queue `"""

implementation:
while True :
    args = await response_queue . get ( )
    try :
        await self . write_response ( stream , * args )
    finally :
        response_queue . task_done ( )


signature:
def get_paths ( self , connection , path ) :
    """Return * real * and * virtual * paths , resolves " . . " with " up " action .
 * Real * path is path for path _ io , when * virtual * deals with
 " user - view " and user requests
 : param connection : internal options for current connected user
 : type connection : : py : class : ` dict `
 : param path : received path from user
 : type path : : py : class : ` str ` or : py : class : ` pathlib . PurePosixPath `
 : return : ( real _ path , virtual _ path )
 : rtype : ( : py : class : ` pathlib . Path ` , : py : class : ` pathlib . PurePosixPath ` )"""

implementation:
virtual_path = pathlib . PurePosixPath ( path )
if not virtual_path . is_absolute ( ) :
    virtual_path = connection . current_directory / virtual_path
resolved_virtual_path = pathlib . PurePosixPath ( "/" )
for part in virtual_path . parts [ 1 : ] :
    if part == ".." :
        resolved_virtual_path = resolved_virtual_path . parent
    else :
        resolved_virtual_path /= part
base_path = connection . user . base_path
real_path = base_path / resolved_virtual_path . relative_to ( "/" )
return real_path , resolved_virtual_path


signature:
def bytes2human ( n , format = "%(value).1f%(symbol)s" ) :
    """> > > bytes2human ( 10000)
 '9K '
 > > > bytes2human ( 100001221)
 '95M '"""

implementation:
symbols = ( 'B' , 'K' , 'M' , 'G' , 'T' , 'P' , 'E' , 'Z' , 'Y' )
prefix = { }
for i , s in enumerate ( symbols [ 1 : ] ) :
    prefix [ s ] = 1 << ( i + 1 ) * 10
for symbol in reversed ( symbols [ 1 : ] ) :
    if n >= prefix [ symbol ] :
        value = float ( n ) / prefix [ symbol ]
        return format % locals ( )
return format % dict ( symbol = symbols [ 0 ] , value = n )


implementation:
symbols = ( 'B' , 'K' , 'M' , 'G' , 'T' , 'P' , 'E' , 'Z' , 'Y' )
letter = s [ - 1 : ] . strip ( ) . upper ( )
num = s [ : - 1 ]
assert num . isdigit ( ) and letter in symbols , s
num = float ( num )
prefix = { symbols [ 0 ] : 1 }
for i , s in enumerate ( symbols [ 1 : ] ) :
    prefix [ s ] = 1 << ( i + 1 ) * 10
return int ( num * prefix [ letter ] )


signature:
def register_memory ( ) :
    """Register an approximation of memory used by FTP server process
 and all of its children ."""

implementation:
# XXX How to get a reliable representation of memory being used is
# not clear . ( rss - shared ) seems kind of ok but we might also use
# the private working set via get _ memory _ maps ( ) . private * .
def get_mem ( proc ) :
    if os . name == 'posix' :
        mem = proc . memory_info_ex ( )
        counter = mem . rss
        if 'shared' in mem . _fields :
            counter -= mem . shared
        return counter
    else : # TODO figure out what to do on Windows
        return proc . get_memory_info ( ) . rss
if SERVER_PROC is not None :
    mem = get_mem ( SERVER_PROC )
    for child in SERVER_PROC . children ( ) :
        mem += get_mem ( child )
    server_memory . append ( bytes2human ( mem ) )


signature:
def timethis ( what ) :
    """" Utility function for making simple benchmarks ( calculates time calls ) .
 It can be used either as a context manager or as a decorator ."""

implementation:
@ contextlib . contextmanager
def benchmark ( ) :
    timer = time . clock if sys . platform == "win32" else time . time
    start = timer ( )
    yield
    stop = timer ( )
    res = ( stop - start )
    print_bench ( what , res , "secs" )
if hasattr ( what , "__call__" ) :
    def timed ( * args , ** kwargs ) :
        with benchmark ( ) :
            return what ( * args , ** kwargs )
    return timed
else :
    return benchmark ( )


signature:
def connect ( ) :
    """Connect to FTP server , login and return an ftplib . FTP instance ."""

implementation:
ftp_class = ftplib . FTP if not SSL else ftplib . FTP_TLS
ftp = ftp_class ( timeout = TIMEOUT )
ftp . connect ( HOST , PORT )
ftp . login ( USER , PASSWORD )
if SSL :
    ftp . prot_p ( )
    # secure data connection
return ftp


signature:
def retr ( ftp ) :
    """Same as ftplib ' s retrbinary ( ) but discard the received data ."""

implementation:
ftp . voidcmd ( 'TYPE I' )
with contextlib . closing ( ftp . transfercmd ( "RETR " + TESTFN ) ) as conn :
    recv_bytes = 0
    while True :
        data = conn . recv ( BUFFER_LEN )
        if not data :
            break
        recv_bytes += len ( data )
ftp . voidresp ( )


signature:
def stor ( ftp = None ) :
    """Same as ftplib ' s storbinary ( ) but just sends dummy data
 instead of reading it from a real file ."""

implementation:
if ftp is None :
    ftp = connect ( )
    quit = True
else :
    quit = False
ftp . voidcmd ( 'TYPE I' )
with contextlib . closing ( ftp . transfercmd ( "STOR " + TESTFN ) ) as conn :
    chunk = b'x' * BUFFER_LEN
    total_sent = 0
    while True :
        sent = conn . send ( chunk )
        total_sent += sent
        if total_sent >= FILE_SIZE :
            break
ftp . voidresp ( )
if quit :
    ftp . quit ( )
return ftp


signature:
def bytes_per_second ( ftp , retr = True ) :
    """Return the number of bytes transmitted in 1 second ."""

implementation:
tot_bytes = 0
if retr :
    def request_file ( ) :
        ftp . voidcmd ( 'TYPE I' )
        conn = ftp . transfercmd ( "retr " + TESTFN )
        return conn
    with contextlib . closing ( request_file ( ) ) as conn :
        register_memory ( )
        stop_at = time . time ( ) + 1.0
        while stop_at > time . time ( ) :
            chunk = conn . recv ( BUFFER_LEN )
            if not chunk :
                a = time . time ( )
                ftp . voidresp ( )
                conn . close ( )
                conn = request_file ( )
                stop_at += time . time ( ) - a
            tot_bytes += len ( chunk )
    try :
        while chunk :
            chunk = conn . recv ( BUFFER_LEN )
        ftp . voidresp ( )
        conn . close ( )
    except ( ftplib . error_temp , ftplib . error_perm ) :
        pass
else :
    ftp . voidcmd ( 'TYPE I' )
    with contextlib . closing ( ftp . transfercmd ( "STOR " + TESTFN ) ) as conn :
        register_memory ( )
        chunk = b'x' * BUFFER_LEN
        stop_at = time . time ( ) + 1
        while stop_at > time . time ( ) :
            tot_bytes += conn . send ( chunk )
    ftp . voidresp ( )
return tot_bytes


signature:
def universal_exception ( coro ) :
    """Decorator . Reraising any exception ( except ` CancelledError ` and
 ` NotImplementedError ` ) with universal exception
 : py : class : ` aioftp . PathIOError `"""

implementation:
@ functools . wraps ( coro )
async def wrapper ( * args , ** kwargs ) :
    try :
        return await coro ( * args , ** kwargs )
    except ( asyncio . CancelledError , NotImplementedError , StopAsyncIteration ) :
        raise
    except Exception :
        raise errors . PathIOError ( reason = sys . exc_info ( ) )
return wrapper


signature:
def defend_file_methods ( coro ) :
    """Decorator . Raises exception when file methods called with wrapped by
 : py : class : ` aioftp . AsyncPathIOContext ` file object ."""

implementation:
@ functools . wraps ( coro )
async def wrapper ( self , file , * args , ** kwargs ) :
    if isinstance ( file , AsyncPathIOContext ) :
        raise ValueError ( "Native path io file methods can not be used " "with wrapped file object" )
    return await coro ( self , file , * args , ** kwargs )
return wrapper


signature:
def async_enterable ( f ) :
    """Decorator . Bring coroutine result up , so it can be used as async context
 > > > async def foo ( ) :
 . . . return AsyncContextInstance ( . . . )
 . . . ctx = await foo ( )
 . . . async with ctx :
 . . . # do
 > > > @ async _ enterable
 . . . async def foo ( ) :
 . . . return AsyncContextInstance ( . . . )
 . . . async with foo ( ) as ctx :
 . . . # do
 . . . ctx = await foo ( )
 . . . async with ctx :
 . . . # do"""

implementation:
@ functools . wraps ( f )
def wrapper ( * args , ** kwargs ) :
    class AsyncEnterableInstance :
        async def __aenter__ ( self ) :
            self . context = await f ( * args , ** kwargs )
            return await self . context . __aenter__ ( )
        async def __aexit__ ( self , * args , ** kwargs ) :
            await self . context . __aexit__ ( * args , ** kwargs )
        def __await__ ( self ) :
            return f ( * args , ** kwargs ) . __await__ ( )
    return AsyncEnterableInstance ( )
return wrapper


signature:
def setlocale ( name ) :
    """Context manager with threading lock for set locale on enter , and set it
 back to original state on exit .
 > > > with setlocale ( " C " ) :"""

implementation:
with LOCALE_LOCK :
    old_locale = locale . setlocale ( locale . LC_ALL )
    try :
        yield locale . setlocale ( locale . LC_ALL , name )
    finally :
        locale . setlocale ( locale . LC_ALL , old_locale )


signature:
async def wait ( self ) :
    """: py : func : ` asyncio . coroutine `
 Wait until can do IO"""

implementation:
if self . _limit is not None and self . _limit > 0 and self . _start is not None :
    now = _now ( )
    end = self . _start + self . _sum / self . _limit
    await asyncio . sleep ( max ( 0 , end - now ) )


signature:
def append ( self , data , start ) :
    """Count ` data ` for throttle
 : param data : bytes of data for count
 : type data : : py : class : ` bytes `
 : param start : start of read / write time from
 : py : meth : ` asyncio . BaseEventLoop . time `
 : type start : : py : class : ` float `"""

implementation:
if self . _limit is not None and self . _limit > 0 :
    if self . _start is None :
        self . _start = start
    if start - self . _start > self . reset_rate :
        self . _sum -= round ( ( start - self . _start ) * self . _limit )
        self . _start = start
    self . _sum += len ( data )


signature:
def limit ( self , value ) :
    """Set throttle limit
 : param value : bytes per second
 : type value : : py : class : ` int ` or : py : class : ` None `"""

implementation:
self . _limit = value
self . _start = None
self . _sum = 0


signature:
def clone ( self ) :
    """Clone throttles without memory"""

implementation: return StreamThrottle ( read = self . read . clone ( ) , write = self . write . clone ( ) )


signature:
def from_limits ( cls , read_speed_limit = None , write_speed_limit = None ) :
    """Simple wrapper for creation : py : class : ` aioftp . StreamThrottle `
 : param read _ speed _ limit : stream read speed limit in bytes or
 : py : class : ` None ` for unlimited
 : type read _ speed _ limit : : py : class : ` int ` or : py : class : ` None `
 : param write _ speed _ limit : stream write speed limit in bytes or
 : py : class : ` None ` for unlimited
 : type write _ speed _ limit : : py : class : ` int ` or : py : class : ` None `"""

implementation: return cls ( read = Throttle ( limit = read_speed_limit ) , write = Throttle ( limit = write_speed_limit ) )


signature:
async def wait ( self , name ) :
    """: py : func : ` asyncio . coroutine `
 Wait for all throttles
 : param name : name of throttle to acquire ( " read " or " write " )
 : type name : : py : class : ` str `"""

implementation:
waiters = [ ]
for throttle in self . throttles . values ( ) :
    curr_throttle = getattr ( throttle , name )
    if curr_throttle . limit :
        waiters . append ( curr_throttle . wait ( ) )
if waiters :
    await asyncio . wait ( waiters )


signature:
def append ( self , name , data , start ) :
    """Update timeout for all throttles
 : param name : name of throttle to append to ( " read " or " write " )
 : type name : : py : class : ` str `
 : param data : bytes of data for count
 : type data : : py : class : ` bytes `
 : param start : start of read / write time from
 : py : meth : ` asyncio . BaseEventLoop . time `
 : type start : : py : class : ` float `"""

implementation:
for throttle in self . throttles . values ( ) :
    getattr ( throttle , name ) . append ( data , start )


signature:
async def read ( self , count = - 1 ) :
    """: py : func : ` asyncio . coroutine `
 : py : meth : ` aioftp . StreamIO . read ` proxy"""

implementation:
await self . wait ( "read" )
start = _now ( )
data = await super ( ) . read ( count )
self . append ( "read" , data , start )
return data


signature:
async def readline ( self ) :
    """: py : func : ` asyncio . coroutine `
 : py : meth : ` aioftp . StreamIO . readline ` proxy"""

implementation:
await self . wait ( "read" )
start = _now ( )
data = await super ( ) . readline ( )
self . append ( "read" , data , start )
return data


signature:
async def write ( self , data ) :
    """: py : func : ` asyncio . coroutine `
 : py : meth : ` aioftp . StreamIO . write ` proxy"""

implementation:
await self . wait ( "write" )
start = _now ( )
await super ( ) . write ( data )
self . append ( "write" , data , start )


signature:
def matches ( self , mask ) :
    """: param mask : Template for comparision . If mask symbol is not digit
 then it passes .
 : type mask : : py : class : ` str `
 > > > Code ( " 123 " ) . matches ( " 1 " )
 True
 > > > Code ( " 123 " ) . matches ( " 1x3 " )
 True"""

implementation: return all ( map ( lambda m , c : not m . isdigit ( ) or m == c , mask , self ) )


signature:
async def finish ( self , expected_codes = "2xx" , wait_codes = "1xx" ) :
    """: py : func : ` asyncio . coroutine `
 Close connection and wait for ` expected _ codes ` response from server
 passing ` wait _ codes ` .
 : param expected _ codes : tuple of expected codes or expected code
 : type expected _ codes : : py : class : ` tuple ` of : py : class : ` str ` or
 : py : class : ` str `
 : param wait _ codes : tuple of wait codes or wait code
 : type wait _ codes : : py : class : ` tuple ` of : py : class : ` str ` or
 : py : class : ` str `"""

implementation:
self . close ( )
await self . client . command ( None , expected_codes , wait_codes )


signature:
async def parse_line ( self ) :
    """: py : func : ` asyncio . coroutine `
 Parsing server response line .
 : return : ( code , line )
 : rtype : ( : py : class : ` aioftp . Code ` , : py : class : ` str ` )
 : raises ConnectionResetError : if received data is empty ( this
 means , that connection is closed )
 : raises asyncio . TimeoutError : if there where no data for ` timeout `
 period"""

implementation:
line = await self . stream . readline ( )
if not line :
    self . stream . close ( )
    raise ConnectionResetError
s = line . decode ( encoding = self . encoding ) . rstrip ( )
logger . info ( s )
return Code ( s [ : 3 ] ) , s [ 3 : ]


signature:
async def parse_response ( self ) :
    """: py : func : ` asyncio . coroutine `
 Parsing full server response ( all lines ) .
 : return : ( code , lines )
 : rtype : ( : py : class : ` aioftp . Code ` , : py : class : ` list ` of : py : class : ` str ` )
 : raises aioftp . StatusCodeError : if received code does not matches all
 already received codes"""

implementation:
code , rest = await self . parse_line ( )
info = [ rest ]
curr_code = code
while rest . startswith ( "-" ) or not curr_code . isdigit ( ) :
    curr_code , rest = await self . parse_line ( )
    if curr_code . isdigit ( ) :
        info . append ( rest )
        if curr_code != code :
            raise errors . StatusCodeError ( code , curr_code , info )
    else :
        info . append ( curr_code + rest )
return code , info


signature:
def check_codes ( self , expected_codes , received_code , info ) :
    """Checks if any of expected matches received .
 : param expected _ codes : tuple of expected codes
 : type expected _ codes : : py : class : ` tuple `
 : param received _ code : received code for matching
 : type received _ code : : py : class : ` aioftp . Code `
 : param info : list of response lines from server
 : type info : : py : class : ` list `
 : raises aioftp . StatusCodeError : if received code does not matches any
 expected code"""

implementation:
if not any ( map ( received_code . matches , expected_codes ) ) :
    raise errors . StatusCodeError ( expected_codes , received_code , info )


signature:
async def command ( self , command = None , expected_codes = ( ) , wait_codes = ( ) ) :
    """: py : func : ` asyncio . coroutine `
 Basic command logic .
 1 . Send command if not omitted .
 2 . Yield response until no wait code matches .
 3 . Check code for expected .
 : param command : command line
 : type command : : py : class : ` str `
 : param expected _ codes : tuple of expected codes or expected code
 : type expected _ codes : : py : class : ` tuple ` of : py : class : ` str ` or
 : py : class : ` str `
 : param wait _ codes : tuple of wait codes or wait code
 : type wait _ codes : : py : class : ` tuple ` of : py : class : ` str ` or
 : py : class : ` str `"""

implementation:
expected_codes = wrap_with_container ( expected_codes )
wait_codes = wrap_with_container ( wait_codes )
if command :
    logger . info ( command )
    message = command + END_OF_LINE
    await self . stream . write ( message . encode ( encoding = self . encoding ) )
if expected_codes or wait_codes :
    code , info = await self . parse_response ( )
    while any ( map ( code . matches , wait_codes ) ) :
        code , info = await self . parse_response ( )
    if expected_codes :
        self . check_codes ( expected_codes , code , info )
    return code , info


signature:
def parse_epsv_response ( s ) :
    """Parsing ` EPSV ` ( ` message ( | | | port | ) ` ) response .
 : param s : response line
 : type s : : py : class : ` str `
 : return : ( ip , port )
 : rtype : ( : py : class : ` None ` , : py : class : ` int ` )"""

implementation:
matches = tuple ( re . finditer ( r"\((.)\1\1\d+\1\)" , s ) )
s = matches [ - 1 ] . group ( )
port = int ( s [ 4 : - 2 ] )
return None , port


signature:
def parse_pasv_response ( s ) :
    """Parsing ` PASV ` server response .
 : param s : response line
 : type s : : py : class : ` str `
 : return : ( ip , port )
 : rtype : ( : py : class : ` str ` , : py : class : ` int ` )"""

implementation:
sub , * _ = re . findall ( r"[^(]*\(([^)]*)" , s )
nums = tuple ( map ( int , sub . split ( "," ) ) )
ip = "." . join ( map ( str , nums [ : 4 ] ) )
port = ( nums [ 4 ] << 8 ) | nums [ 5 ]
return ip , port


signature:
def parse_directory_response ( s ) :
    """Parsing directory server response .
 : param s : response line
 : type s : : py : class : ` str `
 : rtype : : py : class : ` pathlib . PurePosixPath `"""

implementation:
seq_quotes = 0
start = False
directory = ""
for ch in s :
    if not start :
        if ch == "\"" :
            start = True
    else :
        if ch == "\"" :
            seq_quotes += 1
        else :
            if seq_quotes == 1 :
                break
            elif seq_quotes == 2 :
                seq_quotes = 0
                directory += '"'
            directory += ch
return pathlib . PurePosixPath ( directory )


signature:
def parse_unix_mode ( s ) :
    """Parsing unix mode strings ( " rwxr - x - - t " ) into hexacimal notation .
 : param s : mode string
 : type s : : py : class : ` str `
 : return mode :
 : rtype : : py : class : ` int `"""


signature:
def parse_ls_date ( self , s , * , now = None ) :
    """Parsing dates from the ls unix utility . For example ,
 " Nov 18 1958 " and " Nov 18 12:29 " .
 : param s : ls date
 : type s : : py : class : ` str `
 : rtype : : py : class : ` str `"""

implementation:
with setlocale ( "C" ) :
    try :
        if now is None :
            now = datetime . datetime . now ( )
        d = datetime . datetime . strptime ( s , "%b %d %H:%M" )
        d = d . replace ( year = now . year )
        diff = ( now - d ) . total_seconds ( )
        if diff > HALF_OF_YEAR_IN_SECONDS :
            d = d . replace ( year = now . year + 1 )
        elif diff < - HALF_OF_YEAR_IN_SECONDS :
            d = d . replace ( year = now . year - 1 )
    except ValueError :
        d = datetime . datetime . strptime ( s , "%b %d  %Y" )
return self . format_date_time ( d )


signature:
def parse_list_line_unix ( self , b ) :
    """Attempt to parse a LIST line ( similar to unix ls utility ) .
 : param b : response line
 : type b : : py : class : ` bytes ` or : py : class : ` str `
 : return : ( path , info )
 : rtype : ( : py : class : ` pathlib . PurePosixPath ` , : py : class : ` dict ` )"""

implementation:
s = b . decode ( encoding = self . encoding ) . rstrip ( )
info = { }
if s [ 0 ] == "-" :
    info [ "type" ] = "file"
elif s [ 0 ] == "d" :
    info [ "type" ] = "dir"
elif s [ 0 ] == "l" :
    info [ "type" ] = "link"
else :
    info [ "type" ] = "unknown"
# TODO : handle symlinks ( beware the symlink loop )
info [ "unix.mode" ] = self . parse_unix_mode ( s [ 1 : 10 ] )
s = s [ 10 : ] . lstrip ( )
i = s . index ( " " )
info [ "unix.links" ] = s [ : i ]
if not info [ "unix.links" ] . isdigit ( ) :
    raise ValueError
s = s [ i : ] . lstrip ( )
i = s . index ( " " )
info [ "unix.owner" ] = s [ : i ]
s = s [ i : ] . lstrip ( )
i = s . index ( " " )
info [ "unix.group" ] = s [ : i ]
s = s [ i : ] . lstrip ( )
i = s . index ( " " )
info [ "size" ] = s [ : i ]
if not info [ "size" ] . isdigit ( ) :
    raise ValueError
s = s [ i : ] . lstrip ( )
info [ "modify" ] = self . parse_ls_date ( s [ : 12 ] )
s = s [ 12 : ] . strip ( )
if info [ "type" ] == "link" :
    i = s . rindex ( " -> " )
    link_dst = s [ i + 4 : ]
    link_src = s [ : i ]
    i = - 2 if link_dst [ - 1 ] == "\'" or link_dst [ - 1 ] == "\"" else - 1
    info [ "type" ] = "dir" if link_dst [ i ] == "/" else "file"
    s = link_src
return pathlib . PurePosixPath ( s ) , info


signature:
def parse_list_line_windows ( self , b ) :
    """Parsing Microsoft Windows ` dir ` output
 : param b : response line
 : type b : : py : class : ` bytes ` or : py : class : ` str `
 : return : ( path , info )
 : rtype : ( : py : class : ` pathlib . PurePosixPath ` , : py : class : ` dict ` )"""

implementation:
line = b . decode ( encoding = self . encoding ) . rstrip ( "\r\n" )
date_time_end = line . index ( "M" )
date_time_str = line [ : date_time_end + 1 ] . strip ( ) . split ( " " )
date_time_str = " " . join ( [ x for x in date_time_str if len ( x ) > 0 ] )
line = line [ date_time_end + 1 : ] . lstrip ( )
with setlocale ( "C" ) :
    strptime = datetime . datetime . strptime
    date_time = strptime ( date_time_str , "%m/%d/%Y %I:%M %p" )
info = { }
info [ "modify" ] = self . format_date_time ( date_time )
next_space = line . index ( " " )
if line . startswith ( "<DIR>" ) :
    info [ "type" ] = "dir"
else :
    info [ "type" ] = "file"
    info [ "size" ] = line [ : next_space ] . replace ( "," , "" )
    if not info [ "size" ] . isdigit ( ) :
        raise ValueError
# This here could cause a problem if a filename started with
# whitespace , but if we were to try to detect such a condition
# we would have to make strong assumptions about the input format
filename = line [ next_space : ] . lstrip ( )
if filename == "." or filename == ".." :
    raise ValueError
return pathlib . PurePosixPath ( filename ) , info


signature:
def parse_list_line ( self , b ) :
    """Parse LIST response with both Microsoft Windows ® parser and
 UNIX parser
 : param b : response line
 : type b : : py : class : ` bytes ` or : py : class : ` str `
 : return : ( path , info )
 : rtype : ( : py : class : ` pathlib . PurePosixPath ` , : py : class : ` dict ` )"""

implementation:
ex = [ ]
parsers = ( self . parse_list_line_unix , self . parse_list_line_windows )
for parser in parsers :
    try :
        return parser ( b )
    except ( ValueError , KeyError , IndexError ) as e :
        ex . append ( e )
raise ValueError ( "All parsers failed to parse" , b , ex )


signature:
def parse_mlsx_line ( self , b ) :
    """Parsing MLS ( T | D ) response .
 : param b : response line
 : type b : : py : class : ` bytes ` or : py : class : ` str `
 : return : ( path , info )
 : rtype : ( : py : class : ` pathlib . PurePosixPath ` , : py : class : ` dict ` )"""

implementation:
if isinstance ( b , bytes ) :
    s = b . decode ( encoding = self . encoding )
else :
    s = b
line = s . rstrip ( )
facts_found , _ , name = line . partition ( " " )
entry = { }
for fact in facts_found [ : - 1 ] . split ( ";" ) :
    key , _ , value = fact . partition ( "=" )
    entry [ key . lower ( ) ] = value
return pathlib . PurePosixPath ( name ) , entry


signature:
async def connect ( self , host , port = DEFAULT_PORT ) :
    """: py : func : ` asyncio . coroutine `
 Connect to server .
 : param host : host name for connection
 : type host : : py : class : ` str `
 : param port : port number for connection
 : type port : : py : class : ` int `"""

implementation:
await super ( ) . connect ( host , port )
code , info = await self . command ( None , "220" , "120" )
return info


signature:
async def login ( self , user = DEFAULT_USER , password = DEFAULT_PASSWORD , account = DEFAULT_ACCOUNT ) :
    """: py : func : ` asyncio . coroutine `
 Server authentication .
 : param user : username
 : type user : : py : class : ` str `
 : param password : password
 : type password : : py : class : ` str `
 : param account : account ( almost always blank )
 : type account : : py : class : ` str `
 : raises aioftp . StatusCodeError : if unknown code received"""

implementation:
code , info = await self . command ( "USER " + user , ( "230" , "33x" ) )
while code . matches ( "33x" ) :
    if code == "331" :
        cmd = "PASS " + password
    elif code == "332" :
        cmd = "ACCT " + account
    else :
        raise errors . StatusCodeError ( "33x" , code , info )
    code , info = await self . command ( cmd , ( "230" , "33x" ) )


signature:
async def get_current_directory ( self ) :
    """: py : func : ` asyncio . coroutine `
 Getting current working directory .
 : rtype : : py : class : ` pathlib . PurePosixPath `"""

implementation:
code , info = await self . command ( "PWD" , "257" )
directory = self . parse_directory_response ( info [ - 1 ] )
return directory


signature:
async def change_directory ( self , path = ".." ) :
    """: py : func : ` asyncio . coroutine `
 Change current directory . Goes « up » if no parameters passed .
 : param path : new directory , goes « up » if omitted
 : type path : : py : class : ` str ` or : py : class : ` pathlib . PurePosixPath `"""

implementation:
path = pathlib . PurePosixPath ( path )
if path == pathlib . PurePosixPath ( ".." ) :
    cmd = "CDUP"
else :
    cmd = "CWD " + str ( path )
await self . command ( cmd , "2xx" )


signature:
async def make_directory ( self , path , * , parents = True ) :
    """: py : func : ` asyncio . coroutine `
 Make directory .
 : param path : path to directory to create
 : type path : : py : class : ` str ` or : py : class : ` pathlib . PurePosixPath `
 : param parents : create parents if does not exists
 : type parents : : py : class : ` bool `"""

implementation:
path = pathlib . PurePosixPath ( path )
need_create = [ ]
while path . name and not await self . exists ( path ) :
    need_create . append ( path )
    path = path . parent
    if not parents :
        break
need_create . reverse ( )
for path in need_create :
    await self . command ( "MKD " + str ( path ) , "257" )


signature:
def list ( self , path = "" , * , recursive = False , raw_command = None ) :
    """: py : func : ` asyncio . coroutine `
 List all files and directories in " path " .
 : param path : directory or file path
 : type path : : py : class : ` str ` or : py : class : ` pathlib . PurePosixPath `
 : param recursive : list recursively
 : type recursive : : py : class : ` bool `
 : param raw _ command : optional ftp command to use in place of
 fallback logic ( must be one of " MLSD " , " LIST " )
 : type raw _ command : : py : class : ` str `
 : rtype : : py : class : ` list ` or ` async for ` context
 > > > # lazy list
 > > > async for path , info in client . list ( ) :
 . . . # no interaction with client should be here ( ! )
 > > > # eager list
 > > > for path , info in ( await client . list ( ) ) :
 . . . # interaction with client allowed , since all paths are
 . . . # collected already
 > > > stats = await client . list ( )"""

implementation:
class AsyncLister ( AsyncListerMixin ) :
    stream = None
    async def _new_stream ( cls , local_path ) :
        cls . path = local_path
        cls . parse_line = self . parse_mlsx_line
        if raw_command not in [ None , "MLSD" , "LIST" ] :
            raise ValueError ( "raw_command must be one of MLSD or " f"LIST, but got {raw_command}" )
        if raw_command in [ None , "MLSD" ] :
            try :
                command = ( "MLSD " + str ( cls . path ) ) . strip ( )
                return await self . get_stream ( command , "1xx" )
            except errors . StatusCodeError as e :
                code = e . received_codes [ - 1 ]
                if not code . matches ( "50x" ) or raw_command is not None :
                    raise
        if raw_command in [ None , "LIST" ] :
            cls . parse_line = self . parse_list_line
            command = ( "LIST " + str ( cls . path ) ) . strip ( )
            return await self . get_stream ( command , "1xx" )
    def __aiter__ ( cls ) :
        cls . directories = collections . deque ( )
        return cls
    async def __anext__ ( cls ) :
        if cls . stream is None :
            cls . stream = await cls . _new_stream ( path )
        while True :
            line = await cls . stream . readline ( )
            while not line :
                await cls . stream . finish ( )
                if cls . directories :
                    current_path , info = cls . directories . popleft ( )
                    cls . stream = await cls . _new_stream ( current_path )
                    line = await cls . stream . readline ( )
                else :
                    raise StopAsyncIteration
            try :
                name , info = cls . parse_line ( line )
            except Exception :
                continue
            stat = cls . path / name , info
            if info [ "type" ] == "dir" and recursive :
                cls . directories . append ( stat )
            return stat
return AsyncLister ( )


signature:
async def stat ( self , path ) :
    """: py : func : ` asyncio . coroutine `
 Getting path stats .
 : param path : path for getting info
 : type path : : py : class : ` str ` or : py : class : ` pathlib . PurePosixPath `
 : return : path info
 : rtype : : py : class : ` dict `"""

implementation:
path = pathlib . PurePosixPath ( path )
try :
    code , info = await self . command ( "MLST " + str ( path ) , "2xx" )
    name , info = self . parse_mlsx_line ( info [ 1 ] . lstrip ( ) )
    return info
except errors . StatusCodeError as e :
    if not e . received_codes [ - 1 ] . matches ( "50x" ) :
        raise
for p , info in await self . list ( path . parent ) :
    if p . name == path . name :
        return info
else :
    raise errors . StatusCodeError ( Code ( "2xx" ) , Code ( "550" ) , "path does not exists" , )


signature:
async def exists ( self , path ) :
    """: py : func : ` asyncio . coroutine `
 Check path for existence .
 : param path : path to check
 : type path : : py : class : ` str ` or : py : class : ` pathlib . PurePosixPath `
 : rtype : : py : class : ` bool `"""

implementation:
try :
    await self . stat ( path )
    return True
except errors . StatusCodeError as e :
    if e . received_codes [ - 1 ] . matches ( "550" ) :
        return False
    raise


signature:
async def rename ( self , source , destination ) :
    """: py : func : ` asyncio . coroutine `
 Rename ( move ) file or directory .
 : param source : path to rename
 : type source : : py : class : ` str ` or : py : class : ` pathlib . PurePosixPath `
 : param destination : path new name
 : type destination : : py : class : ` str ` or : py : class : ` pathlib . PurePosixPath `"""

implementation:
await self . command ( "RNFR " + str ( source ) , "350" )
await self . command ( "RNTO " + str ( destination ) , "2xx" )


signature:
async def remove ( self , path ) :
    """: py : func : ` asyncio . coroutine `
 High level remove method for removing path recursively ( file or
 directory ) .
 : param path : path to remove
 : type path : : py : class : ` str ` or : py : class : ` pathlib . PurePosixPath `"""

implementation:
if await self . exists ( path ) :
    info = await self . stat ( path )
    if info [ "type" ] == "file" :
        await self . remove_file ( path )
    elif info [ "type" ] == "dir" :
        for name , info in ( await self . list ( path ) ) :
            if info [ "type" ] in ( "dir" , "file" ) :
                await self . remove ( name )
        await self . remove_directory ( path )


signature:
def upload_stream ( self , destination , * , offset = 0 ) :
    """Create stream for write data to ` destination ` file .
 : param destination : destination path of file on server side
 : type destination : : py : class : ` str ` or : py : class : ` pathlib . PurePosixPath `
 : param offset : byte offset for stream start position
 : type offset : : py : class : ` int `
 : rtype : : py : class : ` aioftp . DataConnectionThrottleStreamIO `"""

implementation: return self . get_stream ( "STOR " + str ( destination ) , "1xx" , offset = offset , )


signature:
def append_stream ( self , destination , * , offset = 0 ) :
    """Create stream for append ( write ) data to ` destination ` file .
 : param destination : destination path of file on server side
 : type destination : : py : class : ` str ` or : py : class : ` pathlib . PurePosixPath `
 : param offset : byte offset for stream start position
 : type offset : : py : class : ` int `
 : rtype : : py : class : ` aioftp . DataConnectionThrottleStreamIO `"""

implementation: return self . get_stream ( "APPE " + str ( destination ) , "1xx" , offset = offset , )


signature:
async def upload ( self , source , destination = "" , * , write_into = False , block_size = DEFAULT_BLOCK_SIZE ) :
    """: py : func : ` asyncio . coroutine `
 High level upload method for uploading files and directories
 recursively from file system .
 : param source : source path of file or directory on client side
 : type source : : py : class : ` str ` or : py : class : ` pathlib . Path `
 : param destination : destination path of file or directory on server
 side
 : type destination : : py : class : ` str ` or : py : class : ` pathlib . PurePosixPath `
 : param write _ into : write source into destination ( if you want upload
 file and change it name , as well with directories )
 : type write _ into : : py : class : ` bool `
 : param block _ size : block size for transaction
 : type block _ size : : py : class : ` int `"""

implementation:
source = pathlib . Path ( source )
destination = pathlib . PurePosixPath ( destination )
if not write_into :
    destination = destination / source . name
if await self . path_io . is_file ( source ) :
    await self . make_directory ( destination . parent )
    async with self . path_io . open ( source , mode = "rb" ) as file_in , self . upload_stream ( destination ) as stream :
        async for block in file_in . iter_by_block ( block_size ) :
            await stream . write ( block )
elif await self . path_io . is_dir ( source ) :
    await self . make_directory ( destination )
    sources = collections . deque ( [ source ] )
    while sources :
        src = sources . popleft ( )
        async for path in self . path_io . list ( src ) :
            if write_into :
                relative = destination . name / path . relative_to ( source )
            else :
                relative = path . relative_to ( source . parent )
            if await self . path_io . is_dir ( path ) :
                await self . make_directory ( relative )
                sources . append ( path )
            else :
                await self . upload ( path , relative , write_into = True , block_size = block_size )


signature:
def download_stream ( self , source , * , offset = 0 ) :
    """: py : func : ` asyncio . coroutine `
 Create stream for read data from ` source ` file .
 : param source : source path of file on server side
 : type source : : py : class : ` str ` or : py : class : ` pathlib . PurePosixPath `
 : param offset : byte offset for stream start position
 : type offset : : py : class : ` int `
 : rtype : : py : class : ` aioftp . DataConnectionThrottleStreamIO `"""

implementation: return self . get_stream ( "RETR " + str ( source ) , "1xx" , offset = offset )