signature:
def get_dusty_images ( ) :
    """Returns all images listed in dusty specs ( apps + bundles ) , in the form repository : tag . Tag will be set to latest
 if no tag is specified in the specs"""

implementation:
specs = get_specs ( )
dusty_image_names = [ spec [ 'image' ] for spec in specs [ 'apps' ] . values ( ) + specs [ 'services' ] . values ( ) if 'image' in spec ]
dusty_images = set ( [ name if ':' in name else "{}:latest" . format ( name ) for name in dusty_image_names ] )
return dusty_images


signature:
def get_docker_client ( ) :
    """Ripped off and slightly modified based on docker - py ' s
 kwargs _ from _ env utility function ."""

implementation:
env = get_docker_env ( )
host , cert_path , tls_verify = env [ 'DOCKER_HOST' ] , env [ 'DOCKER_CERT_PATH' ] , env [ 'DOCKER_TLS_VERIFY' ]
params = { 'base_url' : host . replace ( 'tcp://' , 'https://' ) , 'timeout' : None , 'version' : 'auto' }
if tls_verify and cert_path :
    params [ 'tls' ] = docker . tls . TLSConfig ( client_cert = ( os . path . join ( cert_path , 'cert.pem' ) , os . path . join ( cert_path , 'key.pem' ) ) , ca_cert = os . path . join ( cert_path , 'ca.pem' ) , verify = True , ssl_version = None , assert_hostname = False )
return docker . Client ( ** params )


signature:
def get_dusty_containers ( services , include_exited = False ) :
    """Get a list of containers associated with the list
 of services . If no services are provided , attempts to
 return all containers associated with Dusty ."""

implementation:
client = get_docker_client ( )
if services :
    containers = [ get_container_for_app_or_service ( service , include_exited = include_exited ) for service in services ]
    return [ container for container in containers if container ]
else :
    return [ container for container in client . containers ( all = include_exited ) if any ( name . startswith ( '/dusty' ) for name in container . get ( 'Names' , [ ] ) ) ]


signature:
def configure_nfs_server ( ) :
    """This function is used with ` dusty up ` . It will check all active repos to see if
 they are exported . If any are missing , it will replace current dusty exports with
 exports that are needed for currently active repos , and restart
 the nfs server"""

implementation:
repos_for_export = get_all_repos ( active_only = True , include_specs_repo = False )
current_exports = _get_current_exports ( )
needed_exports = _get_exports_for_repos ( repos_for_export )
_ensure_managed_repos_dir_exists ( )
if not needed_exports . difference ( current_exports ) :
    if not _server_is_running ( ) :
        _restart_server ( )
    return
_write_exports_config ( needed_exports )
_restart_server ( )


signature:
def add_exports_for_repos ( repos ) :
    """This function will add needed entries to / etc / exports . It will not remove any
 entries from the file . It will then restart the server if necessary"""

implementation:
current_exports = _get_current_exports ( )
needed_exports = _get_exports_for_repos ( repos )
if not needed_exports . difference ( current_exports ) :
    if not _server_is_running ( ) :
        _restart_server ( )
    return
_write_exports_config ( current_exports . union ( needed_exports ) )
_restart_server ( )


signature:
def _ensure_managed_repos_dir_exists ( ) :
    """Our exports file will be invalid if this folder doesn ' t exist , and the NFS server
 will not run correctly ."""

implementation:
if not os . path . exists ( constants . REPOS_DIR ) :
    os . makedirs ( constants . REPOS_DIR )


signature:
def register_consumer ( ) :
    """Given a hostname and port attempting to be accessed ,
 return a unique consumer ID for accessing logs from
 the referenced container ."""

implementation:
global _consumers
hostname , port = request . form [ 'hostname' ] , request . form [ 'port' ]
app_name = _app_name_from_forwarding_info ( hostname , port )
containers = get_dusty_containers ( [ app_name ] , include_exited = True )
if not containers :
    raise ValueError ( 'No container exists for app {}' . format ( app_name ) )
container = containers [ 0 ]
new_id = uuid1 ( )
new_consumer = Consumer ( container [ 'Id' ] , datetime . utcnow ( ) )
_consumers [ str ( new_id ) ] = new_consumer
response = jsonify ( { 'app_name' : app_name , 'consumer_id' : new_id } )
response . headers [ 'Access-Control-Allow-Origin' ] = '*'
response . headers [ 'Access-Control-Allow-Methods' ] = 'GET, POST'
return response


signature:
def consume ( consumer_id ) :
    """Given an existing consumer ID , return any new lines from the
 log since the last time the consumer was consumed ."""

implementation:
global _consumers
consumer = _consumers [ consumer_id ]
client = get_docker_client ( )
try :
    status = client . inspect_container ( consumer . container_id ) [ 'State' ] [ 'Status' ]
except Exception as e :
    status = 'unknown'
new_logs = client . logs ( consumer . container_id , stdout = True , stderr = True , stream = False , timestamps = False , since = calendar . timegm ( consumer . offset . timetuple ( ) ) )
updated_consumer = Consumer ( consumer . container_id , datetime . utcnow ( ) )
_consumers [ str ( consumer_id ) ] = updated_consumer
response = jsonify ( { 'logs' : new_logs , 'status' : status } )
response . headers [ 'Access-Control-Allow-Origin' ] = '*'
response . headers [ 'Access-Control-Allow-Methods' ] = 'GET, POST'
return response


signature:
def get_app_volume_mounts ( app_name , assembled_specs , test = False ) :
    """This returns a list of formatted volume specs for an app . These mounts declared in the apps ' spec
 and mounts declared in all lib specs the app depends on"""

implementation:
app_spec = assembled_specs [ 'apps' ] [ app_name ]
volumes = [ get_command_files_volume_mount ( app_name , test = test ) ]
volumes . append ( get_asset_volume_mount ( app_name ) )
repo_mount = _get_app_repo_volume_mount ( app_spec )
if repo_mount :
    volumes . append ( repo_mount )
volumes += _get_app_libs_volume_mounts ( app_name , assembled_specs )
return volumes


signature:
def get_lib_volume_mounts ( base_lib_name , assembled_specs ) :
    """Returns a list of the formatted volume specs for a lib"""

implementation:
volumes = [ _get_lib_repo_volume_mount ( assembled_specs [ 'libs' ] [ base_lib_name ] ) ]
volumes . append ( get_command_files_volume_mount ( base_lib_name , test = True ) )
for lib_name in assembled_specs [ 'libs' ] [ base_lib_name ] [ 'depends' ] [ 'libs' ] :
    lib_spec = assembled_specs [ 'libs' ] [ lib_name ]
    volumes . append ( _get_lib_repo_volume_mount ( lib_spec ) )
return volumes


signature:
def _get_app_libs_volume_mounts ( app_name , assembled_specs ) :
    """Returns a list of the formatted volume mounts for all libs that an app uses"""

implementation:
volumes = [ ]
for lib_name in assembled_specs [ 'apps' ] [ app_name ] [ 'depends' ] [ 'libs' ] :
    lib_spec = assembled_specs [ 'libs' ] [ lib_name ]
    volumes . append ( "{}:{}" . format ( Repo ( lib_spec [ 'repo' ] ) . vm_path , container_code_path ( lib_spec ) ) )
return volumes


signature:
def _dusty_vm_exists ( ) :
    """We use VBox directly instead of Docker Machine because it
 shaves about 0.5 seconds off the runtime of this check ."""

implementation:
existing_vms = check_output_demoted ( [ 'VBoxManage' , 'list' , 'vms' ] )
for line in existing_vms . splitlines ( ) :
    if '"{}"' . format ( constants . VM_MACHINE_NAME ) in line :
        return True
return False


signature:
def _init_docker_vm ( ) :
    """Initialize the Dusty VM if it does not already exist ."""

implementation:
if not _dusty_vm_exists ( ) :
    log_to_client ( 'Initializing new Dusty VM with Docker Machine' )
    machine_options = [ '--driver' , 'virtualbox' , '--virtualbox-cpu-count' , '-1' , '--virtualbox-boot2docker-url' , constants . CONFIG_BOOT2DOCKER_URL , '--virtualbox-memory' , str ( get_config_value ( constants . CONFIG_VM_MEM_SIZE ) ) , '--virtualbox-hostonly-nictype' , constants . VM_NIC_TYPE ]
    check_call_demoted ( [ 'docker-machine' , 'create' ] + machine_options + [ constants . VM_MACHINE_NAME ] , redirect_stderr = True )


signature:
def _start_docker_vm ( ) :
    """Start the Dusty VM if it is not already running ."""

implementation:
is_running = docker_vm_is_running ( )
if not is_running :
    log_to_client ( 'Starting docker-machine VM {}' . format ( constants . VM_MACHINE_NAME ) )
    _apply_nat_dns_host_resolver ( )
    _apply_nat_net_less_greedy_subnet ( )
    check_and_log_output_and_error_demoted ( [ 'docker-machine' , 'start' , constants . VM_MACHINE_NAME ] , quiet_on_success = True )
return is_running


signature:
def docker_vm_is_running ( ) :
    """Using VBoxManage is 0.5 seconds or so faster than Machine ."""

implementation:
running_vms = check_output_demoted ( [ 'VBoxManage' , 'list' , 'runningvms' ] )
for line in running_vms . splitlines ( ) :
    if '"{}"' . format ( constants . VM_MACHINE_NAME ) in line :
        return True
return False


signature:
def _get_localhost_ssh_port ( ) :
    """Something in the VM chain , either VirtualBox or Machine , helpfully
 sets up localhost - to - VM forwarding on port 22 . We can inspect this
 rule to determine the port on localhost which gets forwarded to
 22 in the VM ."""

implementation:
for line in _get_vm_config ( ) :
    if line . startswith ( 'Forwarding' ) :
        spec = line . split ( '=' ) [ 1 ] . strip ( '"' )
        name , protocol , host , host_port , target , target_port = spec . split ( ',' )
        if name == 'ssh' and protocol == 'tcp' and target_port == '22' :
            return host_port
raise ValueError ( 'Could not determine localhost port for SSH forwarding' )


signature:
def _get_host_only_mac_address ( ) :
    """Returns the MAC address assigned to the host - only adapter ,
 using output from VBoxManage . Returned MAC address has no colons
 and is lower - cased ."""

implementation:
# Get the number of the host - only adapter
vm_config = _get_vm_config ( )
for line in vm_config :
    if line . startswith ( 'hostonlyadapter' ) :
        adapter_number = int ( line [ 15 : 16 ] )
        break
else :
    raise ValueError ( 'No host-only adapter is defined for the Dusty VM' )
for line in vm_config :
    if line . startswith ( 'macaddress{}' . format ( adapter_number ) ) :
        return line . split ( '=' ) [ 1 ] . strip ( '"' ) . lower ( )
raise ValueError ( 'Could not find MAC address for adapter number {}' . format ( adapter_number ) )


signature:
def _ip_for_mac_from_ip_addr_show ( ip_addr_show , target_mac ) :
    """Given the rather - complex output from an ' ip addr show ' command
 on the VM , parse the output to determine the IP address
 assigned to the interface with the given MAC ."""

implementation:
return_next_ip = False
for line in ip_addr_show . splitlines ( ) :
    line = line . strip ( )
    if line . startswith ( 'link/ether' ) :
        line_mac = line . split ( ' ' ) [ 1 ] . replace ( ':' , '' )
        if line_mac == target_mac :
            return_next_ip = True
    elif return_next_ip and line . startswith ( 'inet' ) and not line . startswith ( 'inet6' ) :
        ip = line . split ( ' ' ) [ 1 ] . split ( '/' ) [ 0 ]
        return ip


signature:
def _get_host_only_ip ( ) :
    """Determine the host - only IP of the Dusty VM through Virtualbox and SSH
 directly , bypassing Docker Machine . We do this because Docker Machine is
 much slower , taking about 600ms total . We are basically doing the same
 flow Docker Machine does in its own code ."""

implementation:
mac = _get_host_only_mac_address ( )
ip_addr_show = check_output_demoted ( [ 'ssh' , '-o' , 'StrictHostKeyChecking=no' , '-o' , 'UserKnownHostsFile=/dev/null' , '-i' , _vm_key_path ( ) , '-p' , _get_localhost_ssh_port ( ) , 'docker@127.0.0.1' , 'ip addr show' ] )
return _ip_for_mac_from_ip_addr_show ( ip_addr_show , mac )


signature:
def create_local_copy ( cookie_file ) :
    """Make a local copy of the sqlite cookie database and return the new filename .
 This is necessary in case this database is still being written to while the user browses
 to avoid sqlite locking errors ."""

implementation:
# if type of cookie _ file is a list , use the first element in the list
if isinstance ( cookie_file , list ) :
    cookie_file = cookie_file [ 0 ]
# check if cookie file exists
if os . path . exists ( cookie_file ) : # copy to random name in tmp folder
    tmp_cookie_file = tempfile . NamedTemporaryFile ( suffix = '.sqlite' ) . name
    open ( tmp_cookie_file , 'wb' ) . write ( open ( cookie_file , 'rb' ) . read ( ) )
    return tmp_cookie_file
else :
    raise BrowserCookieError ( 'Can not find cookie file at: ' + cookie_file )


signature:
def create_cookie ( host , path , secure , expires , name , value ) :
    """Shortcut function to create a cookie"""

implementation: return http . cookiejar . Cookie ( 0 , name , value , None , False , host , host . startswith ( '.' ) , host . startswith ( '.' ) , path , True , secure , expires , False , None , None , { } )


signature:
def load ( domain_name = "" ) :
    """Try to load cookies from all supported browsers and return combined cookiejar
 Optionally pass in a domain name to only load cookies from the specified domain"""

implementation:
cj = http . cookiejar . CookieJar ( )
for cookie_fn in [ chrome , firefox ] :
    try :
        for cookie in cookie_fn ( domain_name = domain_name ) :
            cj . set_cookie ( cookie )
    except BrowserCookieError :
        pass
return cj


signature:
def load ( self ) :
    """Load sqlite cookies into a cookiejar"""

implementation:
con = sqlite3 . connect ( self . tmp_cookie_file )
cur = con . cursor ( )
try : # chrome < = 55
    cur . execute ( 'SELECT host_key, path, secure, expires_utc, name, value, encrypted_value ' 'FROM cookies WHERE host_key like "%{}%";' . format ( self . domain_name ) )
except sqlite3 . OperationalError : # chrome > = 56
    cur . execute ( 'SELECT host_key, path, is_secure, expires_utc, name, value, encrypted_value ' 'FROM cookies WHERE host_key like "%{}%";' . format ( self . domain_name ) )
cj = http . cookiejar . CookieJar ( )
for item in cur . fetchall ( ) :
    host , path , secure , expires , name = item [ : 5 ]
    value = self . _decrypt ( item [ 5 ] , item [ 6 ] )
    c = create_cookie ( host , path , secure , expires , name , value )
    cj . set_cookie ( c )
con . close ( )
return cj


signature:
def _decrypt ( self , value , encrypted_value ) :
    """Decrypt encoded cookies"""

implementation:
if sys . platform == 'win32' :
    return self . _decrypt_windows_chrome ( value , encrypted_value )
if value or ( encrypted_value [ : 3 ] != b'v10' ) :
    return value
# Encrypted cookies should be prefixed with ' v10 ' according to the
# Chromium code . Strip it off .
encrypted_value = encrypted_value [ 3 : ]
encrypted_value_half_len = int ( len ( encrypted_value ) / 2 )
cipher = pyaes . Decrypter ( pyaes . AESModeOfOperationCBC ( self . key , self . iv ) )
decrypted = cipher . feed ( encrypted_value [ : encrypted_value_half_len ] )
decrypted += cipher . feed ( encrypted_value [ encrypted_value_half_len : ] )
decrypted += cipher . feed ( )
return decrypted . decode ( "utf-8" )


signature:
def _gen ( d , limit = 20 , count = False , grouprefs = None ) :
    """docstring for _ gen"""

implementation:
if grouprefs is None :
    grouprefs = { }
ret = [ '' ]
strings = 0
literal = False
for i in d :
    if i [ 0 ] == sre_parse . IN :
        subs = _in ( i [ 1 ] )
        if count :
            strings = ( strings or 1 ) * len ( subs )
        ret = comb ( ret , subs )
    elif i [ 0 ] == sre_parse . LITERAL :
        literal = True
        ret = mappend ( ret , unichr ( i [ 1 ] ) )
    elif i [ 0 ] == sre_parse . CATEGORY :
        subs = CATEGORIES . get ( i [ 1 ] , [ '' ] )
        if count :
            strings = ( strings or 1 ) * len ( subs )
        ret = comb ( ret , subs )
    elif i [ 0 ] == sre_parse . ANY :
        subs = CATEGORIES [ 'category_any' ]
        if count :
            strings = ( strings or 1 ) * len ( subs )
        ret = comb ( ret , subs )
    elif i [ 0 ] == sre_parse . MAX_REPEAT or i [ 0 ] == sre_parse . MIN_REPEAT :
        items = list ( i [ 1 ] [ 2 ] )
        if i [ 1 ] [ 1 ] + 1 - i [ 1 ] [ 0 ] >= limit :
            r1 = i [ 1 ] [ 0 ]
            r2 = i [ 1 ] [ 0 ] + limit
        else :
            r1 = i [ 1 ] [ 0 ]
            r2 = i [ 1 ] [ 1 ] + 1
        ran = range ( r1 , r2 )
        if count :
            branch_count = 0
            for p in ran :
                branch_count += pow ( _gen ( items , limit , True , grouprefs ) , p )
            strings = ( strings or 1 ) * branch_count
        ret = prods ( ret , ran , items , limit , grouprefs )
    elif i [ 0 ] == sre_parse . BRANCH :
        if count :
            for x in i [ 1 ] [ 1 ] :
                strings += _gen ( x , limit , True , grouprefs ) or 1
        ret = concit ( ret , i [ 1 ] [ 1 ] , limit , grouprefs )
    elif i [ 0 ] == sre_parse . SUBPATTERN or i [ 0 ] == sre_parse . ASSERT :
        subexpr = i [ 1 ] [ 1 ]
        if IS_PY36_OR_GREATER and i [ 0 ] == sre_parse . SUBPATTERN :
            subexpr = i [ 1 ] [ 3 ]
        if count :
            strings = ( strings or 1 ) * ( sum ( ggen ( [ 0 ] , _gen , subexpr , limit = limit , count = True , grouprefs = grouprefs ) ) or 1 )
        ret = ggen ( ret , _gen , subexpr , limit = limit , count = False , grouprefs = grouprefs , groupref = i [ 1 ] [ 0 ] )
    # ignore ^ and $
    elif i [ 0 ] == sre_parse . AT :
        continue
    elif i [ 0 ] == sre_parse . NOT_LITERAL :
        subs = list ( CATEGORIES [ 'category_any' ] )
        if unichr ( i [ 1 ] ) in subs :
            subs . remove ( unichr ( i [ 1 ] ) )
        if count :
            strings = ( strings or 1 ) * len ( subs )
        ret = comb ( ret , subs )
    elif i [ 0 ] == sre_parse . GROUPREF :
        ret = dappend ( ret , grouprefs , i [ 1 ] )
    elif i [ 0 ] == sre_parse . ASSERT_NOT :
        pass
    else :
        print ( '[!] cannot handle expression ' + repr ( i ) )
if count :
    if strings == 0 and literal :
        inc = True
        for i in d :
            if i [ 0 ] not in ( sre_parse . AT , sre_parse . LITERAL ) :
                inc = False
        if inc :
            strings = 1
    return strings
return ret


signature:
def _randone ( d , limit = 20 , grouprefs = None ) :
    if grouprefs is None :
        grouprefs = { }
    """docstring for _ randone"""

implementation:
ret = ''
for i in d :
    if i [ 0 ] == sre_parse . IN :
        ret += choice ( _in ( i [ 1 ] ) )
    elif i [ 0 ] == sre_parse . LITERAL :
        ret += unichr ( i [ 1 ] )
    elif i [ 0 ] == sre_parse . CATEGORY :
        ret += choice ( CATEGORIES . get ( i [ 1 ] , [ '' ] ) )
    elif i [ 0 ] == sre_parse . ANY :
        ret += choice ( CATEGORIES [ 'category_any' ] )
    elif i [ 0 ] == sre_parse . MAX_REPEAT or i [ 0 ] == sre_parse . MIN_REPEAT :
        if i [ 1 ] [ 1 ] + 1 - i [ 1 ] [ 0 ] >= limit :
            min , max = i [ 1 ] [ 0 ] , i [ 1 ] [ 0 ] + limit - 1
        else :
            min , max = i [ 1 ] [ 0 ] , i [ 1 ] [ 1 ]
        for _ in range ( randint ( min , max ) ) :
            ret += _randone ( list ( i [ 1 ] [ 2 ] ) , limit , grouprefs )
    elif i [ 0 ] == sre_parse . BRANCH :
        ret += _randone ( choice ( i [ 1 ] [ 1 ] ) , limit , grouprefs )
    elif i [ 0 ] == sre_parse . SUBPATTERN or i [ 0 ] == sre_parse . ASSERT :
        subexpr = i [ 1 ] [ 1 ]
        if IS_PY36_OR_GREATER and i [ 0 ] == sre_parse . SUBPATTERN :
            subexpr = i [ 1 ] [ 3 ]
        subp = _randone ( subexpr , limit , grouprefs )
        if i [ 1 ] [ 0 ] :
            grouprefs [ i [ 1 ] [ 0 ] ] = subp
        ret += subp
    elif i [ 0 ] == sre_parse . AT :
        continue
    elif i [ 0 ] == sre_parse . NOT_LITERAL :
        c = list ( CATEGORIES [ 'category_any' ] )
        if unichr ( i [ 1 ] ) in c :
            c . remove ( unichr ( i [ 1 ] ) )
        ret += choice ( c )
    elif i [ 0 ] == sre_parse . GROUPREF :
        ret += grouprefs [ i [ 1 ] ]
    elif i [ 0 ] == sre_parse . ASSERT_NOT :
        pass
    else :
        print ( '[!] cannot handle expression "%s"' % str ( i ) )
return ret


signature:
def sre_to_string ( sre_obj , paren = True ) :
    """sre _ parse object to string
 : param sre _ obj : Output of sre _ parse . parse ( )
 : type sre _ obj : list
 : rtype : str"""

implementation:
ret = u''
for i in sre_obj :
    if i [ 0 ] == sre_parse . IN :
        prefix = ''
        if len ( i [ 1 ] ) and i [ 1 ] [ 0 ] [ 0 ] == sre_parse . NEGATE :
            prefix = '^'
        ret += u'[{0}{1}]' . format ( prefix , sre_to_string ( i [ 1 ] , paren = paren ) )
    elif i [ 0 ] == sre_parse . LITERAL :
        u = unichr ( i [ 1 ] )
        ret += u if u not in sre_parse . SPECIAL_CHARS else '\\{0}' . format ( u )
    elif i [ 0 ] == sre_parse . CATEGORY :
        ret += REVERSE_CATEGORIES [ i [ 1 ] ]
    elif i [ 0 ] == sre_parse . ANY :
        ret += '.'
    elif i [ 0 ] == sre_parse . BRANCH : # TODO simplifications here
        parts = [ sre_to_string ( x , paren = paren ) for x in i [ 1 ] [ 1 ] ]
        if not any ( parts ) :
            continue
        if i [ 1 ] [ 0 ] :
            if len ( parts ) == 1 :
                paren = False
            prefix = ''
        else :
            prefix = '?:'
        branch = '|' . join ( parts )
        if paren :
            ret += '({0}{1})' . format ( prefix , branch )
        else :
            ret += '{0}' . format ( branch )
    elif i [ 0 ] == sre_parse . SUBPATTERN :
        subexpr = i [ 1 ] [ 1 ]
        if IS_PY36_OR_GREATER and i [ 0 ] == sre_parse . SUBPATTERN :
            subexpr = i [ 1 ] [ 3 ]
        if i [ 1 ] [ 0 ] :
            ret += '({0})' . format ( sre_to_string ( subexpr , paren = False ) )
        else :
            ret += '{0}' . format ( sre_to_string ( subexpr , paren = paren ) )
    elif i [ 0 ] == sre_parse . NOT_LITERAL :
        ret += '[^{0}]' . format ( unichr ( i [ 1 ] ) )
    elif i [ 0 ] == sre_parse . MAX_REPEAT :
        if i [ 1 ] [ 0 ] == i [ 1 ] [ 1 ] :
            range_str = '{{{0}}}' . format ( i [ 1 ] [ 0 ] )
        else :
            if i [ 1 ] [ 0 ] == 0 and i [ 1 ] [ 1 ] - i [ 1 ] [ 0 ] == sre_parse . MAXREPEAT :
                range_str = '*'
            elif i [ 1 ] [ 0 ] == 1 and i [ 1 ] [ 1 ] - i [ 1 ] [ 0 ] == sre_parse . MAXREPEAT - 1 :
                range_str = '+'
            else :
                range_str = '{{{0},{1}}}' . format ( i [ 1 ] [ 0 ] , i [ 1 ] [ 1 ] )
        ret += sre_to_string ( i [ 1 ] [ 2 ] , paren = paren ) + range_str
    elif i [ 0 ] == sre_parse . MIN_REPEAT :
        if i [ 1 ] [ 0 ] == 0 and i [ 1 ] [ 1 ] == sre_parse . MAXREPEAT :
            range_str = '*?'
        elif i [ 1 ] [ 0 ] == 1 and i [ 1 ] [ 1 ] == sre_parse . MAXREPEAT :
            range_str = '+?'
        elif i [ 1 ] [ 1 ] == sre_parse . MAXREPEAT :
            range_str = '{{{0},}}?' . format ( i [ 1 ] [ 0 ] )
        else :
            range_str = '{{{0},{1}}}?' . format ( i [ 1 ] [ 0 ] , i [ 1 ] [ 1 ] )
        ret += sre_to_string ( i [ 1 ] [ 2 ] , paren = paren ) + range_str
    elif i [ 0 ] == sre_parse . GROUPREF :
        ret += '\\{0}' . format ( i [ 1 ] )
    elif i [ 0 ] == sre_parse . AT :
        if i [ 1 ] == sre_parse . AT_BEGINNING :
            ret += '^'
        elif i [ 1 ] == sre_parse . AT_END :
            ret += '$'
    elif i [ 0 ] == sre_parse . NEGATE :
        pass
    elif i [ 0 ] == sre_parse . RANGE :
        ret += '{0}-{1}' . format ( unichr ( i [ 1 ] [ 0 ] ) , unichr ( i [ 1 ] [ 1 ] ) )
    elif i [ 0 ] == sre_parse . ASSERT :
        if i [ 1 ] [ 0 ] :
            ret += '(?={0})' . format ( sre_to_string ( i [ 1 ] [ 1 ] , paren = False ) )
        else :
            ret += '{0}' . format ( sre_to_string ( i [ 1 ] [ 1 ] , paren = paren ) )
    elif i [ 0 ] == sre_parse . ASSERT_NOT :
        pass
    else :
        print ( '[!] cannot handle expression "%s"' % str ( i ) )
return ret


signature:
def parse ( s ) :
    """Regular expression parser
 : param s : Regular expression
 : type s : str
 : rtype : list"""

implementation:
if IS_PY3 :
    r = sre_parse . parse ( s , flags = U )
else :
    r = sre_parse . parse ( s . decode ( 'utf-8' ) , flags = U )
return list ( r )


signature:
def ib64_patched ( self , attrsD , contentparams ) :
    """Patch isBase64 to prevent Base64 encoding of JSON content"""

implementation:
if attrsD . get ( "mode" , "" ) == "base64" :
    return 0
if self . contentparams [ "type" ] . startswith ( "text/" ) :
    return 0
if self . contentparams [ "type" ] . endswith ( "+xml" ) :
    return 0
if self . contentparams [ "type" ] . endswith ( "/xml" ) :
    return 0
if self . contentparams [ "type" ] . endswith ( "/json" ) :
    return 0
return 0


signature:
def ss_wrap ( func ) :
    """ensure that a SavedSearch object exists"""

implementation:
def wrapper ( self , * args , ** kwargs ) :
    if not self . savedsearch :
        self . savedsearch = SavedSearch ( self )
    return func ( self , * args , ** kwargs )
return wrapper


signature:
def default_headers ( self ) :
    """It ' s always OK to include these headers"""

implementation:
_headers = { "User-Agent" : "Pyzotero/%s" % __version__ , "Zotero-API-Version" : "%s" % __api_version__ , }
if self . api_key :
    _headers [ "Authorization" ] = "Bearer %s" % self . api_key
return _headers


signature:
def _cache ( self , response , key ) :
    """Add a retrieved template to the cache for 304 checking
 accepts a dict and key name , adds the retrieval time , and adds both
 to self . templates as a new dict using the specified key"""

implementation:
# cache template and retrieval time for subsequent calls
thetime = datetime . datetime . utcnow ( ) . replace ( tzinfo = pytz . timezone ( "GMT" ) )
self . templates [ key ] = { "tmplt" : response . json ( ) , "updated" : thetime }
return copy . deepcopy ( response . json ( ) )


signature:
def _cleanup ( self , to_clean , allow = ( ) ) :
    """Remove keys we added for internal use"""

implementation:
# this item ' s been retrieved from the API , we only need the ' data '
# entry
if to_clean . keys ( ) == [ "links" , "library" , "version" , "meta" , "key" , "data" ] :
    to_clean = to_clean [ "data" ]
return dict ( [ [ k , v ] for k , v in list ( to_clean . items ( ) ) if ( k in allow or k not in self . temp_keys ) ] )


signature:
def _retrieve_data ( self , request = None ) :
    """Retrieve Zotero items via the API
 Combine endpoint and request to access the specific resource
 Returns a JSON document"""

implementation:
full_url = "%s%s" % ( self . endpoint , request )
# The API doesn ' t return this any more , so we have to cheat
self . self_link = request
self . request = requests . get ( url = full_url , headers = self . default_headers ( ) )
self . request . encoding = "utf-8"
try :
    self . request . raise_for_status ( )
except requests . exceptions . HTTPError :
    error_handler ( self . request )
return self . request


signature:
def _extract_links ( self ) :
    """Extract self , first , next , last links from a request response"""

implementation:
extracted = dict ( )
try :
    for key , value in self . request . links . items ( ) :
        parsed = urlparse ( value [ "url" ] )
        fragment = "{path}?{query}" . format ( path = parsed [ 2 ] , query = parsed [ 4 ] )
        extracted [ key ] = fragment
    # add a ' self ' link
    parsed = list ( urlparse ( self . self_link ) )
    # strip ' format ' query parameter
    stripped = "&" . join ( [ "%s=%s" % ( p [ 0 ] , p [ 1 ] ) for p in parse_qsl ( parsed [ 4 ] ) if p [ 0 ] != "format" ] )
    # rebuild url fragment
    # this is a death march
    extracted [ "self" ] = urlunparse ( [ parsed [ 0 ] , parsed [ 1 ] , parsed [ 2 ] , parsed [ 3 ] , stripped , parsed [ 5 ] ] )
    return extracted
except KeyError : # No links present , because it ' s a single item
    return None


signature:
def _updated ( self , url , payload , template = None ) :
    """Generic call to see if a template request returns 304
 accepts :
 - a string to combine with the API endpoint
 - a dict of format values , in case they ' re required by ' url '
 - a template name to check for
 As per the API docs , a template less than 1 hour old is
 assumed to be fresh , and will immediately return False if found"""

implementation:
# If the template is more than an hour old , try a 304
if ( abs ( datetime . datetime . utcnow ( ) . replace ( tzinfo = pytz . timezone ( "GMT" ) ) - self . templates [ template ] [ "updated" ] ) . seconds > 3600 ) :
    query = self . endpoint + url . format ( u = self . library_id , t = self . library_type , ** payload )
    headers = { "If-Modified-Since" : payload [ "updated" ] . strftime ( "%a, %d %b %Y %H:%M:%S %Z" ) }
    headers . update ( self . default_headers ( ) )
    # perform the request , and check whether the response returns 304
    req = requests . get ( query , headers = headers )
    try :
        req . raise_for_status ( )
    except requests . exceptions . HTTPError :
        error_handler ( req )
    return req . status_code == 304
# Still plenty of life left in ' t
return False


signature:
def add_parameters ( self , ** params ) :
    """Add URL parameters
 Also ensure that only valid format / content combinations are requested"""

implementation:
self . url_params = None
# we want JSON by default
if not params . get ( "format" ) :
    params [ "format" ] = "json"
# non - standard content must be retrieved as Atom
if params . get ( "content" ) :
    params [ "format" ] = "atom"
# TODO : rewrite format = atom , content = json request
if "limit" not in params or params . get ( "limit" ) == 0 :
    params [ "limit" ] = 100
# Need ability to request arbitrary number of results for version
# response
# -1 value is hack that works with current version
elif params [ "limit" ] == - 1 or params [ "limit" ] is None :
    del params [ "limit" ]
# bib format can ' t have a limit
if params . get ( "format" ) == "bib" :
    del params [ "limit" ]
self . url_params = urlencode ( params , doseq = True )


signature:
def _build_query ( self , query_string , no_params = False ) :
    """Set request parameters . Will always add the user ID if it hasn ' t
 been specifically set by an API method"""

implementation:
try :
    query = quote ( query_string . format ( u = self . library_id , t = self . library_type ) )
except KeyError as err :
    raise ze . ParamNotPassed ( "There's a request parameter missing: %s" % err )
# Add the URL parameters and the user key , if necessary
if no_params is False :
    if not self . url_params :
        self . add_parameters ( )
    query = "%s?%s" % ( query , self . url_params )
return query


signature:
def publications ( self ) :
    """Return the contents of My Publications"""

implementation:
if self . library_type != "users" :
    raise ze . CallDoesNotExist ( "This API call does not exist for group libraries" )
query_string = "/{t}/{u}/publications/items"
return self . _build_query ( query_string )


signature:
def num_collectionitems ( self , collection ) :
    """Return the total number of items in the specified collection"""

implementation:
query = "/{t}/{u}/collections/{c}/items" . format ( u = self . library_id , t = self . library_type , c = collection . upper ( ) )
return self . _totals ( query )


signature:
def num_tagitems ( self , tag ) :
    """Return the total number of items for the specified tag"""

implementation:
query = "/{t}/{u}/tags/{ta}/items" . format ( u = self . library_id , t = self . library_type , ta = tag )
return self . _totals ( query )


signature:
def _totals ( self , query ) :
    """General method for returning total counts"""

implementation:
self . add_parameters ( limit = 1 )
query = self . _build_query ( query )
self . _retrieve_data ( query )
self . url_params = None
# extract the ' total items ' figure
return int ( self . request . headers [ "Total-Results" ] )


signature:
def key_info ( self , ** kwargs ) :
    """Retrieve info about the permissions associated with the
 key associated to the given Zotero instance"""

implementation:
query_string = "/keys/{k}" . format ( k = self . api_key )
return self . _build_query ( query_string )


signature:
def fulltext_item ( self , itemkey , ** kwargs ) :
    """Get full - text content for an item"""

implementation:
query_string = "/{t}/{u}/items/{itemkey}/fulltext" . format ( t = self . library_type , u = self . library_id , itemkey = itemkey )
return self . _build_query ( query_string )


signature:
def set_fulltext ( self , itemkey , payload ) :
    """Set full - text data for an item
 < itemkey > should correspond to an existing attachment item .
 payload should be a dict containing three keys :
 ' content ' : the full - text content and either
 For text documents , ' indexedChars ' and ' totalChars ' OR
 For PDFs , ' indexedPages ' and ' totalPages ' ."""

implementation:
headers = self . default_headers ( )
headers . update ( { "Content-Type" : "application/json" } )
req = requests . put ( url = self . endpoint + "/{t}/{u}/items/{k}/fulltext" . format ( t = self . library_type , u = self . library_id , k = itemkey ) , headers = headers , data = json . dumps ( payload ) , )
try :
    req . raise_for_status ( )
except requests . exceptions . HTTPError :
    error_handler ( req )
return True


signature:
def new_fulltext ( self , version ) :
    """Retrieve list of full - text content items and versions which are newer
 than < version >"""

implementation:
query_string = "/{t}/{u}/fulltext" . format ( t = self . library_type , u = self . library_id )
headers = { "since" : str ( version ) }
headers . update ( self . default_headers ( ) )
req = requests . get ( self . endpoint + query_string , headers = headers )
try :
    req . raise_for_status ( )
except requests . exceptions . HTTPError :
    error_handler ( req )
return req . json ( )


signature:
def last_modified_version ( self , ** kwargs ) :
    """Get the last modified version"""

implementation:
self . items ( ** kwargs )
return int ( self . request . headers . get ( "last-modified-version" , 0 ) )


signature:
def file ( self , item , ** kwargs ) :
    """Get the file from an specific item"""

implementation:
query_string = "/{t}/{u}/items/{i}/file" . format ( u = self . library_id , t = self . library_type , i = item . upper ( ) )
return self . _build_query ( query_string , no_params = True )


signature:
def dump ( self , itemkey , filename = None , path = None ) :
    """Dump a file attachment to disk , with optional filename and path"""

implementation:
if not filename :
    filename = self . item ( itemkey ) [ "data" ] [ "filename" ]
if path :
    pth = os . path . join ( path , filename )
else :
    pth = filename
file = self . file ( itemkey )
if self . snapshot :
    self . snapshot = False
    pth = pth + ".zip"
with open ( pth , "wb" ) as f :
    f . write ( file )


signature:
def collections_sub ( self , collection , ** kwargs ) :
    """Get subcollections for a specific collection"""

implementation:
query_string = "/{t}/{u}/collections/{c}/collections" . format ( u = self . library_id , t = self . library_type , c = collection . upper ( ) )
return self . _build_query ( query_string )


signature:
def tags ( self , ** kwargs ) :
    """Get tags"""

implementation:
query_string = "/{t}/{u}/tags"
self . tag_data = True
return self . _build_query ( query_string )


signature:
def iterfollow ( self ) :
    """Generator for self . follow ( )"""

implementation:
# use same criterion as self . follow ( )
if self . links is None :
    return
if self . links . get ( "next" ) :
    yield self . follow ( )
else :
    raise StopIteration


signature:
def everything ( self , query ) :
    """Retrieve all items in the library for a particular query
 This method will override the ' limit ' parameter if it ' s been set"""

implementation:
try :
    items = [ ]
    items . extend ( query )
    while self . links . get ( "next" ) :
        items . extend ( self . follow ( ) )
except TypeError : # we have a bibliography object ughh
    items = copy . deepcopy ( query )
    while self . links . get ( "next" ) :
        items . entries . extend ( self . follow ( ) . entries )
return items


signature:
def get_subset ( self , subset ) :
    """Retrieve a subset of items
 Accepts a single argument : a list of item IDs"""

implementation:
if len ( subset ) > 50 :
    raise ze . TooManyItems ( "You may only retrieve 50 items per call" )
# remember any url parameters that have been set
params = self . url_params
retr = [ ]
for itm in subset :
    retr . extend ( self . item ( itm ) )
    self . url_params = params
# clean up URL params when we ' re finished
self . url_params = None
return retr


signature:
def _json_processor ( self , retrieved ) :
    """Format and return data from API calls which return Items"""

implementation:
json_kwargs = { }
if self . preserve_json_order :
    json_kwargs [ "object_pairs_hook" ] = OrderedDict
# send entries to _ tags _ data if there ' s no JSON
try :
    items = [ json . loads ( e [ "content" ] [ 0 ] [ "value" ] , ** json_kwargs ) for e in retrieved . entries ]
except KeyError :
    return self . _tags_data ( retrieved )
return items


signature:
def _csljson_processor ( self , retrieved ) :
    """Return a list of dicts which are dumped CSL JSON"""

implementation:
items = [ ]
json_kwargs = { }
if self . preserve_json_order :
    json_kwargs [ "object_pairs_hook" ] = OrderedDict
for csl in retrieved . entries :
    items . append ( json . loads ( csl [ "content" ] [ 0 ] [ "value" ] , ** json_kwargs ) )
self . url_params = None
return items


signature:
def _bib_processor ( self , retrieved ) :
    """Return a list of strings formatted as HTML bibliography entries"""

implementation:
items = [ ]
for bib in retrieved . entries :
    items . append ( bib [ "content" ] [ 0 ] [ "value" ] )
self . url_params = None
return items


signature:
def _citation_processor ( self , retrieved ) :
    """Return a list of strings formatted as HTML citation entries"""

implementation:
items = [ ]
for cit in retrieved . entries :
    items . append ( cit [ "content" ] [ 0 ] [ "value" ] )
self . url_params = None
return items


signature:
def item_template ( self , itemtype ) :
    """Get a template for a new item"""

implementation:
# if we have a template and it hasn ' t been updated since we stored it
template_name = "item_template_" + itemtype
query_string = "/items/new?itemType={i}" . format ( i = itemtype )
if self . templates . get ( template_name ) and not self . _updated ( query_string , self . templates [ template_name ] , template_name ) :
    return copy . deepcopy ( self . templates [ template_name ] [ "tmplt" ] )
# otherwise perform a normal request and cache the response
retrieved = self . _retrieve_data ( query_string )
return self . _cache ( retrieved , template_name )


signature:
def _attachment ( self , payload , parentid = None ) :
    """Create attachments
 accepts a list of one or more attachment template dicts
 and an optional parent Item ID . If this is specified ,
 attachments are created under this ID"""

implementation:
attachment = Zupload ( self , payload , parentid )
res = attachment . upload ( )
return res


signature:
def show_condition_operators ( self , condition ) :
    """Show available operators for a given saved search condition"""

implementation:
# dict keys of allowed operators for the current condition
permitted_operators = self . savedsearch . conditions_operators . get ( condition )
# transform these into values
permitted_operators_list = set ( [ self . savedsearch . operators . get ( op ) for op in permitted_operators ] )
return permitted_operators_list


signature:
def saved_search ( self , name , conditions ) :
    """Create a saved search . conditions is a list of dicts
 containing search conditions , and must contain the following str keys :
 condition , operator , value"""

implementation:
self . savedsearch . _validate ( conditions )
payload = [ { "name" : name , "conditions" : conditions } ]
headers = { "Zotero-Write-Token" : token ( ) }
headers . update ( self . default_headers ( ) )
req = requests . post ( url = self . endpoint + "/{t}/{u}/searches" . format ( t = self . library_type , u = self . library_id ) , headers = headers , data = json . dumps ( payload ) , )
self . request = req
try :
    req . raise_for_status ( )
except requests . exceptions . HTTPError :
    error_handler ( req )
return req . json ( )


signature:
def delete_saved_search ( self , keys ) :
    """Delete one or more saved searches by passing a list of one or more
 unique search keys"""

implementation:
headers = { "Zotero-Write-Token" : token ( ) }
headers . update ( self . default_headers ( ) )
req = requests . delete ( url = self . endpoint + "/{t}/{u}/searches" . format ( t = self . library_type , u = self . library_id ) , headers = headers , params = { "searchKey" : "," . join ( keys ) } , )
self . request = req
try :
    req . raise_for_status ( )
except requests . exceptions . HTTPError :
    error_handler ( req )
return req . status_code


signature:
def upload_attachments ( self , attachments , parentid = None , basedir = None ) :
    """Upload files to the already created ( but never uploaded ) attachments"""

implementation: return Zupload ( self , attachments , parentid , basedir = basedir ) . upload ( )


signature:
def add_tags ( self , item , * tags ) :
    """Add one or more tags to a retrieved item ,
 then update it on the server
 Accepts a dict , and one or more tags to add to it
 Returns the updated item from the server"""

implementation:
# Make sure there ' s a tags field , or add one
try :
    assert item [ "data" ] [ "tags" ]
except AssertionError :
    item [ "data" ] [ "tags" ] = list ( )
for tag in tags :
    item [ "data" ] [ "tags" ] . append ( { "tag" : "%s" % tag } )
# make sure everything ' s OK
assert self . check_items ( [ item ] )
return self . update_item ( item )


signature:
def check_items ( self , items ) :
    """Check that items to be created contain no invalid dict keys
 Accepts a single argument : a list of one or more dicts
 The retrieved fields are cached and re - used until a 304 call fails"""

implementation:
# check for a valid cached version
if self . templates . get ( "item_fields" ) and not self . _updated ( "/itemFields" , self . templates [ "item_fields" ] , "item_fields" ) :
    template = set ( t [ "field" ] for t in self . templates [ "item_fields" ] [ "tmplt" ] )
else :
    template = set ( t [ "field" ] for t in self . item_fields ( ) )
# add fields we know to be OK
template = template | set ( [ "path" , "tags" , "notes" , "itemType" , "creators" , "mimeType" , "linkMode" , "note" , "charset" , "dateAdded" , "version" , "collections" , "dateModified" , "relations" , # attachment items
"parentItem" , "mtime" , "contentType" , "md5" , "filename" , ] )
template = template | set ( self . temp_keys )
for pos , item in enumerate ( items ) :
    if set ( item ) == set ( [ "links" , "library" , "version" , "meta" , "key" , "data" ] ) : # we have an item that was retrieved from the API
        item = item [ "data" ]
    to_check = set ( i for i in list ( item . keys ( ) ) )
    difference = to_check . difference ( template )
    if difference :
        raise ze . InvalidItemFields ( "Invalid keys present in item %s: %s" % ( pos + 1 , " " . join ( i for i in difference ) ) )
return items


signature:
def fields_types ( self , tname , qstring , itemtype ) :
    """Retrieve item fields or creator types"""

implementation:
# check for a valid cached version
template_name = tname + itemtype
query_string = qstring . format ( i = itemtype )
if self . templates . get ( template_name ) and not self . _updated ( query_string , self . templates [ template_name ] , template_name ) :
    return self . templates [ template_name ] [ "tmplt" ]
# otherwise perform a normal request and cache the response
retrieved = self . _retrieve_data ( query_string )
return self . _cache ( retrieved , template_name )


signature:
def item_fields ( self ) :
    """Get all available item fields"""

implementation:
# Check for a valid cached version
if self . templates . get ( "item_fields" ) and not self . _updated ( "/itemFields" , self . templates [ "item_fields" ] , "item_fields" ) :
    return self . templates [ "item_fields" ] [ "tmplt" ]
query_string = "/itemFields"
# otherwise perform a normal request and cache the response
retrieved = self . _retrieve_data ( query_string )
return self . _cache ( retrieved , "item_fields" )


signature:
def create_items ( self , payload , parentid = None , last_modified = None ) :
    """Create new Zotero items
 Accepts two arguments :
 a list containing one or more item dicts
 an optional parent item ID .
 Note that this can also be used to update existing items"""

implementation:
if len ( payload ) > 50 :
    raise ze . TooManyItems ( "You may only create up to 50 items per call" )
# TODO : strip extra data if it ' s an existing item
headers = { "Zotero-Write-Token" : token ( ) , "Content-Type" : "application/json" }
if last_modified is not None :
    headers [ "If-Unmodified-Since-Version" ] = str ( last_modified )
to_send = json . dumps ( [ i for i in self . _cleanup ( * payload , allow = ( "key" ) ) ] )
headers . update ( self . default_headers ( ) )
req = requests . post ( url = self . endpoint + "/{t}/{u}/items" . format ( t = self . library_type , u = self . library_id ) , data = to_send , headers = dict ( headers ) , )
self . request = req
try :
    req . raise_for_status ( )
except requests . exceptions . HTTPError :
    error_handler ( req )
resp = req . json ( )
if parentid : # we need to create child items using PATCH
# TODO : handle possibility of item creation + failed parent
# attachment
    uheaders = { "If-Unmodified-Since-Version" : req . headers [ "last-modified-version" ] }
    uheaders . update ( self . default_headers ( ) )
    for value in resp [ "success" ] . values ( ) :
        payload = json . dumps ( { "parentItem" : parentid } )
        presp = requests . patch ( url = self . endpoint + "/{t}/{u}/items/{v}" . format ( t = self . library_type , u = self . library_id , v = value ) , data = payload , headers = dict ( uheaders ) , )
        self . request = presp
        try :
            presp . raise_for_status ( )
        except requests . exceptions . HTTPError :
            error_handler ( presp )
return resp


signature:
def create_collections ( self , payload , last_modified = None ) :
    """Create new Zotero collections
 Accepts one argument , a list of dicts containing the following keys :
 ' name ' : the name of the collection
 ' parentCollection ' : OPTIONAL , the parent collection to which you wish to add this"""

implementation:
# no point in proceeding if there ' s no ' name ' key
for item in payload :
    if "name" not in item :
        raise ze . ParamNotPassed ( "The dict you pass must include a 'name' key" )
    # add a blank ' parentCollection ' key if it hasn ' t been passed
    if "parentCollection" not in item :
        item [ "parentCollection" ] = ""
headers = { "Zotero-Write-Token" : token ( ) }
if last_modified is not None :
    headers [ "If-Unmodified-Since-Version" ] = str ( last_modified )
headers . update ( self . default_headers ( ) )
req = requests . post ( url = self . endpoint + "/{t}/{u}/collections" . format ( t = self . library_type , u = self . library_id ) , headers = headers , data = json . dumps ( payload ) , )
self . request = req
try :
    req . raise_for_status ( )
except requests . exceptions . HTTPError :
    error_handler ( req )
return req . json ( )


signature:
def update_collection ( self , payload , last_modified = None ) :
    """Update a Zotero collection property such as ' name '
 Accepts one argument , a dict containing collection data retrieved
 using e . g . ' collections ( ) '"""

implementation:
modified = payload [ "version" ]
if last_modified is not None :
    modified = last_modified
key = payload [ "key" ]
headers = { "If-Unmodified-Since-Version" : str ( modified ) }
headers . update ( self . default_headers ( ) )
headers . update ( { "Content-Type" : "application/json" } )
req = requests . put ( url = self . endpoint + "/{t}/{u}/collections/{c}" . format ( t = self . library_type , u = self . library_id , c = key ) , headers = headers , data = json . dumps ( payload ) , )
self . request = req
try :
    req . raise_for_status ( )
except requests . exceptions . HTTPError :
    error_handler ( req )
return True


signature:
def attachment_simple ( self , files , parentid = None ) :
    """Add attachments using filenames as title
 Arguments :
 One or more file paths to add as attachments :
 An optional Item ID , which will create child attachments"""

implementation:
orig = self . _attachment_template ( "imported_file" )
to_add = [ orig . copy ( ) for fls in files ]
for idx , tmplt in enumerate ( to_add ) :
    tmplt [ "title" ] = os . path . basename ( files [ idx ] )
    tmplt [ "filename" ] = files [ idx ]
if parentid :
    return self . _attachment ( to_add , parentid )
else :
    return self . _attachment ( to_add )


signature:
def attachment_both ( self , files , parentid = None ) :
    """Add child attachments using title , filename
 Arguments :
 One or more lists or tuples containing title , file path
 An optional Item ID , which will create child attachments"""

implementation:
orig = self . _attachment_template ( "imported_file" )
to_add = [ orig . copy ( ) for f in files ]
for idx , tmplt in enumerate ( to_add ) :
    tmplt [ "title" ] = files [ idx ] [ 0 ]
    tmplt [ "filename" ] = files [ idx ] [ 1 ]
if parentid :
    return self . _attachment ( to_add , parentid )
else :
    return self . _attachment ( to_add )


signature:
def update_item ( self , payload , last_modified = None ) :
    """Update an existing item
 Accepts one argument , a dict containing Item data"""

implementation:
to_send = self . check_items ( [ payload ] ) [ 0 ]
if last_modified is None :
    modified = payload [ "version" ]
else :
    modified = last_modified
ident = payload [ "key" ]
headers = { "If-Unmodified-Since-Version" : str ( modified ) }
headers . update ( self . default_headers ( ) )
req = requests . patch ( url = self . endpoint + "/{t}/{u}/items/{id}" . format ( t = self . library_type , u = self . library_id , id = ident ) , headers = headers , data = json . dumps ( to_send ) , )
self . request = req
try :
    req . raise_for_status ( )
except requests . exceptions . HTTPError :
    error_handler ( req )
return True


signature:
def update_items ( self , payload ) :
    """Update existing items
 Accepts one argument , a list of dicts containing Item data"""

implementation:
to_send = [ self . check_items ( [ p ] ) [ 0 ] for p in payload ]
headers = { }
headers . update ( self . default_headers ( ) )
# the API only accepts 50 items at a time , so we have to split
# anything longer
for chunk in chunks ( to_send , 50 ) :
    req = requests . post ( url = self . endpoint + "/{t}/{u}/items/" . format ( t = self . library_type , u = self . library_id ) , headers = headers , data = json . dumps ( chunk ) , )
    self . request = req
    try :
        req . raise_for_status ( )
    except requests . exceptions . HTTPError :
        error_handler ( req )
    return True


signature:
def addto_collection ( self , collection , payload ) :
    """Add one or more items to a collection
 Accepts two arguments :
 The collection ID , and an item dict"""

implementation:
ident = payload [ "key" ]
modified = payload [ "version" ]
# add the collection data from the item
modified_collections = payload [ "data" ] [ "collections" ] + [ collection ]
headers = { "If-Unmodified-Since-Version" : str ( modified ) }
headers . update ( self . default_headers ( ) )
req = requests . patch ( url = self . endpoint + "/{t}/{u}/items/{i}" . format ( t = self . library_type , u = self . library_id , i = ident ) , data = json . dumps ( { "collections" : modified_collections } ) , headers = headers , )
self . request = req
try :
    req . raise_for_status ( )
except requests . exceptions . HTTPError :
    error_handler ( req )
return True


signature:
def delete_tags ( self , * payload ) :
    """Delete a group of tags
 pass in up to 50 tags , or use * [ tags ]"""

implementation:
if len ( payload ) > 50 :
    raise ze . TooManyItems ( "Only 50 tags or fewer may be deleted" )
modified_tags = " || " . join ( [ tag for tag in payload ] )
# first , get version data by getting one tag
self . tags ( limit = 1 )
headers = { "If-Unmodified-Since-Version" : self . request . headers [ "last-modified-version" ] }
headers . update ( self . default_headers ( ) )
req = requests . delete ( url = self . endpoint + "/{t}/{u}/tags" . format ( t = self . library_type , u = self . library_id ) , params = { "tag" : modified_tags } , headers = headers , )
self . request = req
try :
    req . raise_for_status ( )
except requests . exceptions . HTTPError :
    error_handler ( req )
return True


signature:
def delete_item ( self , payload , last_modified = None ) :
    """Delete Items from a Zotero library
 Accepts a single argument :
 a dict containing item data
 OR a list of dicts containing item data"""

implementation:
params = None
if isinstance ( payload , list ) :
    params = { "itemKey" : "," . join ( [ p [ "key" ] for p in payload ] ) }
    if last_modified is not None :
        modified = last_modified
    else :
        modified = payload [ 0 ] [ "version" ]
    url = self . endpoint + "/{t}/{u}/items" . format ( t = self . library_type , u = self . library_id )
else :
    ident = payload [ "key" ]
    if last_modified is not None :
        modified = last_modified
    else :
        modified = payload [ "version" ]
    url = self . endpoint + "/{t}/{u}/items/{c}" . format ( t = self . library_type , u = self . library_id , c = ident )
headers = { "If-Unmodified-Since-Version" : str ( modified ) }
headers . update ( self . default_headers ( ) )
req = requests . delete ( url = url , params = params , headers = headers )
self . request = req
try :
    req . raise_for_status ( )
except requests . exceptions . HTTPError :
    error_handler ( req )
return True


signature:
def _validate ( self , conditions ) :
    """Validate saved search conditions , raising an error if any contain invalid operators"""

implementation:
allowed_keys = set ( self . searchkeys )
operators_set = set ( self . operators . keys ( ) )
for condition in conditions :
    if set ( condition . keys ( ) ) != allowed_keys :
        raise ze . ParamNotPassed ( "Keys must be all of: %s" % ", " . join ( self . searchkeys ) )
    if condition . get ( "operator" ) not in operators_set :
        raise ze . ParamNotPassed ( "You have specified an unknown operator: %s" % condition . get ( "operator" ) )
    # dict keys of allowed operators for the current condition
    permitted_operators = self . conditions_operators . get ( condition . get ( "condition" ) )
    # transform these into values
    permitted_operators_list = set ( [ self . operators . get ( op ) for op in permitted_operators ] )
    if condition . get ( "operator" ) not in permitted_operators_list :
        raise ze . ParamNotPassed ( "You may not use the '%s' operator when selecting the '%s' condition. \nAllowed operators: %s" % ( condition . get ( "operator" ) , condition . get ( "condition" ) , ", " . join ( list ( permitted_operators_list ) ) , ) )


signature:
def _verify ( self , payload ) :
    """ensure that all files to be attached exist
 open ( ) ' s better than exists ( ) , cos it avoids a race condition"""

implementation:
if not payload : # Check payload has nonzero length
    raise ze . ParamNotPassed
for templt in payload :
    if os . path . isfile ( str ( self . basedir . joinpath ( templt [ "filename" ] ) ) ) :
        try : # if it is a file , try to open it , and catch the error
            with open ( str ( self . basedir . joinpath ( templt [ "filename" ] ) ) ) :
                pass
        except IOError :
            raise ze . FileDoesNotExist ( "The file at %s couldn't be opened or found." % str ( self . basedir . joinpath ( templt [ "filename" ] ) ) )
    # no point in continuing if the file isn ' t a file
    else :
        raise ze . FileDoesNotExist ( "The file at %s couldn't be opened or found." % str ( self . basedir . joinpath ( templt [ "filename" ] ) ) )


signature:
def _create_prelim ( self ) :
    """Step 0 : Register intent to upload files"""

implementation:
self . _verify ( self . payload )
if "key" in self . payload [ 0 ] and self . payload [ 0 ] [ "key" ] :
    if next ( ( i for i in self . payload if "key" not in i ) , False ) :
        raise ze . UnsupportedParams ( "Can't pass payload entries with and without keys to Zupload" )
    return None
    # Don ' t do anything if payload comes with keys
liblevel = "/{t}/{u}/items"
# Create one or more new attachments
headers = { "Zotero-Write-Token" : token ( ) , "Content-Type" : "application/json" }
headers . update ( self . zinstance . default_headers ( ) )
# If we have a Parent ID , add it as a parentItem
if self . parentid :
    for child in self . payload :
        child [ "parentItem" ] = self . parentid
to_send = json . dumps ( self . payload )
req = requests . post ( url = self . zinstance . endpoint + liblevel . format ( t = self . zinstance . library_type , u = self . zinstance . library_id ) , data = to_send , headers = headers , )
try :
    req . raise_for_status ( )
except requests . exceptions . HTTPError :
    error_handler ( req )
data = req . json ( )
for k in data [ "success" ] :
    self . payload [ int ( k ) ] [ "key" ] = data [ "success" ] [ k ]
return data


signature:
def _get_auth ( self , attachment , reg_key , md5 = None ) :
    """Step 1 : get upload authorisation for a file"""

implementation:
mtypes = mimetypes . guess_type ( attachment )
digest = hashlib . md5 ( )
with open ( attachment , "rb" ) as att :
    for chunk in iter ( lambda : att . read ( 8192 ) , b"" ) :
        digest . update ( chunk )
auth_headers = { "Content-Type" : "application/x-www-form-urlencoded" }
if not md5 :
    auth_headers [ "If-None-Match" ] = "*"
else : # docs specify that for existing file we use this
    auth_headers [ "If-Match" ] = md5
auth_headers . update ( self . zinstance . default_headers ( ) )
data = { "md5" : digest . hexdigest ( ) , "filename" : os . path . basename ( attachment ) , "filesize" : os . path . getsize ( attachment ) , "mtime" : str ( int ( os . path . getmtime ( attachment ) * 1000 ) ) , "contentType" : mtypes [ 0 ] or "application/octet-stream" , "charset" : mtypes [ 1 ] , "params" : 1 , }
auth_req = requests . post ( url = self . zinstance . endpoint + "/{t}/{u}/items/{i}/file" . format ( t = self . zinstance . library_type , u = self . zinstance . library_id , i = reg_key ) , data = data , headers = auth_headers , )
try :
    auth_req . raise_for_status ( )
except requests . exceptions . HTTPError :
    error_handler ( auth_req )
return auth_req . json ( )


signature:
def _upload_file ( self , authdata , attachment , reg_key ) :
    """Step 2 : auth successful , and file not on server
 zotero . org / support / dev / server _ api / file _ upload # a _ full _ upload
 reg _ key isn ' t used , but we need to pass it through to Step 3"""

implementation:
upload_dict = authdata [ "params" ]
# using params now since prefix / suffix concat was giving ConnectionError
# must pass tuple of tuples not dict to ensure key comes first
upload_list = [ ( "key" , upload_dict [ "key" ] ) ]
for k in upload_dict :
    if k != "key" :
        upload_list . append ( ( k , upload_dict [ k ] ) )
# The prior code for attaching file gave me content not match md5
# errors
upload_list . append ( ( "file" , open ( attachment , "rb" ) . read ( ) ) )
upload_pairs = tuple ( upload_list )
try :
    upload = requests . post ( url = authdata [ "url" ] , files = upload_pairs , headers = { # " Content - Type " : authdata [ ' contentType ' ] ,
    "User-Agent" : "Pyzotero/%s" % __version__ } , )
except ( requests . exceptions . ConnectionError ) :
    raise ze . UploadError ( "ConnectionError" )
try :
    upload . raise_for_status ( )
except requests . exceptions . HTTPError :
    error_handler ( upload )
# now check the responses
return self . _register_upload ( authdata , reg_key )


signature:
def _register_upload ( self , authdata , reg_key ) :
    """Step 3 : upload successful , so register it"""

implementation:
reg_headers = { "Content-Type" : "application/x-www-form-urlencoded" , "If-None-Match" : "*" , }
reg_headers . update ( self . zinstance . default_headers ( ) )
reg_data = { "upload" : authdata . get ( "uploadKey" ) }
upload_reg = requests . post ( url = self . zinstance . endpoint + "/{t}/{u}/items/{i}/file" . format ( t = self . zinstance . library_type , u = self . zinstance . library_id , i = reg_key ) , data = reg_data , headers = dict ( reg_headers ) , )
try :
    upload_reg . raise_for_status ( )
except requests . exceptions . HTTPError :
    error_handler ( upload_reg )


signature:
def upload ( self ) :
    """File upload functionality
 Goes through upload steps 0 - 3 ( private class methods ) , and returns
 a dict noting success , failure , or unchanged
 ( returning the payload entries with that property as a list for each status )"""

implementation:
result = { "success" : [ ] , "failure" : [ ] , "unchanged" : [ ] }
self . _create_prelim ( )
for item in self . payload :
    if "key" not in item :
        result [ "failure" ] . append ( item )
        continue
    attach = str ( self . basedir . joinpath ( item [ "filename" ] ) )
    authdata = self . _get_auth ( attach , item [ "key" ] , md5 = item . get ( "md5" , None ) )
    # no need to keep going if the file exists
    if authdata . get ( "exists" ) :
        result [ "unchanged" ] . append ( item )
        continue
    self . _upload_file ( authdata , attach , item [ "key" ] )
    result [ "success" ] . append ( item )
return result


signature:
def which ( program , win_allow_cross_arch = True ) :
    """Identify the location of an executable file ."""

implementation:
def is_exe ( path ) :
    return os . path . isfile ( path ) and os . access ( path , os . X_OK )
def _get_path_list ( ) :
    return os . environ [ 'PATH' ] . split ( os . pathsep )
if os . name == 'nt' :
    def find_exe ( program ) :
        root , ext = os . path . splitext ( program )
        if ext :
            if is_exe ( program ) :
                return program
        else :
            for ext in os . environ [ 'PATHEXT' ] . split ( os . pathsep ) :
                program_path = root + ext . lower ( )
                if is_exe ( program_path ) :
                    return program_path
        return None
    def get_path_list ( ) :
        paths = _get_path_list ( )
        if win_allow_cross_arch :
            alt_sys_path = os . path . expandvars ( r"$WINDIR\Sysnative" )
            if os . path . isdir ( alt_sys_path ) :
                paths . insert ( 0 , alt_sys_path )
            else :
                alt_sys_path = os . path . expandvars ( r"$WINDIR\SysWOW64" )
                if os . path . isdir ( alt_sys_path ) :
                    paths . append ( alt_sys_path )
        return paths
else :
    def find_exe ( program ) :
        return program if is_exe ( program ) else None
    get_path_list = _get_path_list
if os . path . split ( program ) [ 0 ] :
    program_path = find_exe ( program )
    if program_path :
        return program_path
else :
    for path in get_path_list ( ) :
        program_path = find_exe ( os . path . join ( path , program ) )
        if program_path :
            return program_path
return None


signature:
def split_multiline ( value ) :
    """Split a multiline string into a list , excluding blank lines ."""

implementation: return [ element for element in ( line . strip ( ) for line in value . split ( '\n' ) ) if element ]


signature:
def split_elements ( value ) :
    """Split a string with comma or space - separated elements into a list ."""

implementation:
items = [ v . strip ( ) for v in value . split ( ',' ) ]
if len ( items ) == 1 :
    items = value . split ( )
return items


signature:
def eval_environ ( value ) :
    """Evaluate environment markers ."""

implementation:
def eval_environ_str ( value ) :
    parts = value . split ( ';' )
    if len ( parts ) < 2 :
        return value
    expr = parts [ 1 ] . lstrip ( )
    if not re . match ( "^((\\w+(\\.\\w+)?|'.*?'|\".*?\")\\s+" '(in|==|!=|not in)\\s+' "(\\w+(\\.\\w+)?|'.*?'|\".*?\")" '(\\s+(or|and)\\s+)?)+$' , expr ) :
        raise ValueError ( 'bad environment marker: %r' % expr )
    expr = re . sub ( r"(platform\.\w+)" , r"\1()" , expr )
    return parts [ 0 ] if eval ( expr ) else ''
if isinstance ( value , list ) :
    new_value = [ ]
    for element in value :
        element = eval_environ_str ( element )
        if element :
            new_value . append ( element )
elif isinstance ( value , str ) :
    new_value = eval_environ_str ( value )
else :
    new_value = value
return new_value


signature:
def get_cfg_value ( config , section , option ) :
    """Get configuration value ."""

implementation:
try :
    value = config [ section ] [ option ]
except KeyError :
    if ( section , option ) in MULTI_OPTIONS :
        return [ ]
    else :
        return ''
if ( section , option ) in MULTI_OPTIONS :
    value = split_multiline ( value )
if ( section , option ) in ENVIRON_OPTIONS :
    value = eval_environ ( value )
return value


signature:
def set_cfg_value ( config , section , option , value ) :
    """Set configuration value ."""

implementation:
if isinstance ( value , list ) :
    value = '\n' . join ( value )
config [ section ] [ option ] = value


signature:
def cfg_to_args ( config ) :
    """Compatibility helper to use setup . cfg in setup . py ."""

implementation:
kwargs = { }
opts_to_args = { 'metadata' : [ ( 'name' , 'name' ) , ( 'author' , 'author' ) , ( 'author-email' , 'author_email' ) , ( 'maintainer' , 'maintainer' ) , ( 'maintainer-email' , 'maintainer_email' ) , ( 'home-page' , 'url' ) , ( 'summary' , 'description' ) , ( 'description' , 'long_description' ) , ( 'download-url' , 'download_url' ) , ( 'classifier' , 'classifiers' ) , ( 'platform' , 'platforms' ) , ( 'license' , 'license' ) , ( 'keywords' , 'keywords' ) , ] , 'files' : [ ( 'packages_root' , 'package_dir' ) , ( 'packages' , 'packages' ) , ( 'modules' , 'py_modules' ) , ( 'scripts' , 'scripts' ) , ( 'package_data' , 'package_data' ) , ( 'data_files' , 'data_files' ) , ] , }
opts_to_args [ 'metadata' ] . append ( ( 'requires-dist' , 'install_requires' ) )
if IS_PY2K and not which ( '3to2' ) :
    kwargs [ 'setup_requires' ] = [ '3to2' ]
kwargs [ 'zip_safe' ] = False
for section in opts_to_args :
    for option , argname in opts_to_args [ section ] :
        value = get_cfg_value ( config , section , option )
        if value :
            kwargs [ argname ] = value
if 'long_description' not in kwargs :
    kwargs [ 'long_description' ] = read_description_file ( config )
if 'package_dir' in kwargs :
    kwargs [ 'package_dir' ] = { '' : kwargs [ 'package_dir' ] }
if 'keywords' in kwargs :
    kwargs [ 'keywords' ] = split_elements ( kwargs [ 'keywords' ] )
if 'package_data' in kwargs :
    kwargs [ 'package_data' ] = get_package_data ( kwargs [ 'package_data' ] )
if 'data_files' in kwargs :
    kwargs [ 'data_files' ] = get_data_files ( kwargs [ 'data_files' ] )
kwargs [ 'version' ] = get_version ( )
if not IS_PY2K :
    kwargs [ 'test_suite' ] = 'test'
return kwargs


signature:
def run_3to2 ( args = None ) :
    """Convert Python files using lib3to2."""

implementation:
args = BASE_ARGS_3TO2 if args is None else BASE_ARGS_3TO2 + args
try :
    proc = subprocess . Popen ( [ '3to2' ] + args , stderr = subprocess . PIPE )
except OSError :
    for path in glob . glob ( '*.egg' ) :
        if os . path . isdir ( path ) and path not in sys . path :
            sys . path . append ( path )
    try :
        from lib3to2 . main import main as lib3to2_main
    except ImportError :
        raise OSError ( '3to2 script is unavailable.' )
    else :
        if lib3to2_main ( 'lib3to2.fixes' , args ) :
            raise Exception ( 'lib3to2 parsing error' )
else : # HACK : workaround for 3to2 never returning non - zero
# when using the - j option .
    num_errors = 0
    while proc . poll ( ) is None :
        line = proc . stderr . readline ( )
        sys . stderr . write ( line )
        num_errors += line . count ( ': ParseError: ' )
    if proc . returncode or num_errors :
        raise Exception ( 'lib3to2 parsing error' )


signature:
def write_py2k_header ( file_list ) :
    """Write Python 2 shebang and add encoding cookie if needed ."""

implementation:
if not isinstance ( file_list , list ) :
    file_list = [ file_list ]
python_re = re . compile ( br"^(#!.*\bpython)(.*)([\r\n]+)$" )
coding_re = re . compile ( br"coding[:=]\s*([-\w.]+)" )
new_line_re = re . compile ( br"([\r\n]+)$" )
version_3 = LooseVersion ( '3' )
for file in file_list :
    if not os . path . getsize ( file ) :
        continue
    rewrite_needed = False
    python_found = False
    coding_found = False
    lines = [ ]
    f = open ( file , 'rb' )
    try :
        while len ( lines ) < 2 :
            line = f . readline ( )
            match = python_re . match ( line )
            if match :
                python_found = True
                version = LooseVersion ( match . group ( 2 ) . decode ( ) or '2' )
                try :
                    version_test = version >= version_3
                except TypeError :
                    version_test = True
                if version_test :
                    line = python_re . sub ( br"\g<1>2\g<3>" , line )
                    rewrite_needed = True
            elif coding_re . search ( line ) :
                coding_found = True
            lines . append ( line )
        if not coding_found :
            match = new_line_re . search ( lines [ 0 ] )
            newline = match . group ( 1 ) if match else b"\n"
            line = b"# -*- coding: utf-8 -*-" + newline
            lines . insert ( 1 if python_found else 0 , line )
            rewrite_needed = True
        if rewrite_needed :
            lines += f . readlines ( )
    finally :
        f . close ( )
    if rewrite_needed :
        f = open ( file , 'wb' )
        try :
            f . writelines ( lines )
        finally :
            f . close ( )


signature:
def which ( program ) :
    """Identify the location of an executable file ."""

implementation:
if os . path . split ( program ) [ 0 ] :
    program_path = find_exe ( program )
    if program_path :
        return program_path
else :
    for path in get_path_list ( ) :
        program_path = find_exe ( os . path . join ( path , program ) )
        if program_path :
            return program_path
return None


signature:
def correct ( text : str , matches : [ Match ] ) -> str :
    """Automatically apply suggestions to the text ."""

implementation:
ltext = list ( text )
matches = [ match for match in matches if match . replacements ]
errors = [ ltext [ match . offset : match . offset + match . errorlength ] for match in matches ]
correct_offset = 0
for n , match in enumerate ( matches ) :
    frompos , topos = ( correct_offset + match . offset , correct_offset + match . offset + match . errorlength )
    if ltext [ frompos : topos ] != errors [ n ] :
        continue
    repl = match . replacements [ 0 ]
    ltext [ frompos : topos ] = list ( repl )
    correct_offset += len ( repl ) - len ( errors [ n ] )
return '' . join ( ltext )


signature:
def get_version ( ) :
    """Get LanguageTool version ."""

implementation:
version = _get_attrib ( ) . get ( 'version' )
if not version :
    match = re . search ( r"LanguageTool-?.*?(\S+)$" , get_directory ( ) )
    if match :
        version = match . group ( 1 )
return version


signature:
def get_languages ( ) -> set :
    """Get supported languages ."""

implementation:
try :
    languages = cache [ 'languages' ]
except KeyError :
    languages = LanguageTool . _get_languages ( )
    cache [ 'languages' ] = languages
return languages


signature:
def get_directory ( ) :
    """Get LanguageTool directory ."""

implementation:
try :
    language_check_dir = cache [ 'language_check_dir' ]
except KeyError :
    def version_key ( string ) :
        return [ int ( e ) if e . isdigit ( ) else e for e in re . split ( r"(\d+)" , string ) ]
    def get_lt_dir ( base_dir ) :
        paths = [ path for path in glob . glob ( os . path . join ( base_dir , 'LanguageTool*' ) ) if os . path . isdir ( path ) ]
        return max ( paths , key = version_key ) if paths else None
    base_dir = os . path . dirname ( sys . argv [ 0 ] )
    language_check_dir = get_lt_dir ( base_dir )
    if not language_check_dir :
        try :
            base_dir = os . path . dirname ( os . path . abspath ( __file__ ) )
        except NameError :
            pass
        else :
            language_check_dir = get_lt_dir ( base_dir )
        if not language_check_dir :
            raise PathError ( "can't find LanguageTool directory in {!r}" . format ( base_dir ) )
    cache [ 'language_check_dir' ] = language_check_dir
return language_check_dir


signature:
def set_directory ( path = None ) :
    """Set LanguageTool directory ."""

implementation:
old_path = get_directory ( )
terminate_server ( )
cache . clear ( )
if path :
    cache [ 'language_check_dir' ] = path
    try :
        get_jar_info ( )
    except Error :
        cache [ 'language_check_dir' ] = old_path
        raise


signature:
def check ( self , text : str , srctext = None ) -> [ Match ] :
    """Match text against enabled rules ."""

implementation:
root = self . _get_root ( self . _url , self . _encode ( text , srctext ) )
return [ Match ( e . attrib ) for e in root if e . tag == 'error' ]


signature:
def _check_api ( self , text : str , srctext = None ) -> bytes :
    """Match text against enabled rules ( result in XML format ) ."""

implementation:
root = self . _get_root ( self . _url , self . _encode ( text , srctext ) )
return ( b'<?xml version="1.0" encoding="UTF-8"?>\n' + ElementTree . tostring ( root ) + b"\n" )


signature:
def correct ( self , text : str , srctext = None ) -> str :
    """Automatically apply suggestions to the text ."""

implementation: return correct ( text , self . check ( text , srctext ) )


signature:
def parse_java_version ( version_text ) :
    """Return Java version ( major1 , major2 ) .
 > > > parse _ java _ version ( ' ' ' java version " 1.6.0_65"
 . . . Java ( TM ) SE Runtime Environment ( build 1.6.0_65 - b14-462-11M4609)
 . . . Java HotSpot ( TM ) 64 - Bit Server VM ( build 20.65 - b04-462 , mixed mode ) )
 (1 , 6)
 > > > parse _ java _ version ( ' ' '
 . . . openjdk version " 1.8.0_60"
 . . . OpenJDK Runtime Environment ( build 1.8.0_60 - b27)
 . . . OpenJDK 64 - Bit Server VM ( build 25.60 - b23 , mixed mode ) )
 (1 , 8)"""

implementation:
match = re . search ( JAVA_VERSION_REGEX , version_text )
if not match :
    raise SystemExit ( 'Could not parse Java version from """{}""".' . format ( version_text ) )
return ( int ( match . group ( 'major1' ) ) , int ( match . group ( 'major2' ) ) )


signature:
def get_newest_possible_languagetool_version ( ) :
    """Return newest compatible version .
 > > > version = get _ newest _ possible _ languagetool _ version ( )
 > > > version in [ JAVA _ 6 _ COMPATIBLE _ VERSION ,
 . . . JAVA _ 7 _ COMPATIBLE _ VERSION ,
 . . . LATEST _ VERSION ]
 True"""

implementation:
java_path = find_executable ( 'java' )
if not java_path : # Just ignore this and assume an old version of Java . It might not be
# found because of a PATHEXT - related issue
# ( https : / / bugs . python . org / issue2200 ) .
    return JAVA_6_COMPATIBLE_VERSION
output = subprocess . check_output ( [ java_path , '-version' ] , stderr = subprocess . STDOUT , universal_newlines = True )
java_version = parse_java_version ( output )
if java_version >= ( 1 , 8 ) :
    return LATEST_VERSION
elif java_version >= ( 1 , 7 ) :
    return JAVA_7_COMPATIBLE_VERSION
elif java_version >= ( 1 , 6 ) :
    warn ( 'language-check would be able to use a newer version of ' 'LanguageTool if you had Java 7 or newer installed' )
    return JAVA_6_COMPATIBLE_VERSION
else :
    raise SystemExit ( 'You need at least Java 6 to use language-check' )


signature:
def get_common_prefix ( z ) :
    """Get common directory in a zip file if any ."""

implementation:
name_list = z . namelist ( )
if name_list and all ( n . startswith ( name_list [ 0 ] ) for n in name_list [ 1 : ] ) :
    return name_list [ 0 ]
return None


signature:
def _process_events ( self , events ) :
    """Process events from proactor ."""

implementation:
for f , callback , transferred , key , ov in events :
    try :
        self . _logger . debug ( 'Invoking event callback {}' . format ( callback ) )
        value = callback ( transferred , key , ov )
    except OSError :
        self . _logger . warning ( 'Event callback failed' , exc_info = sys . exc_info ( ) )
    else :
        f . set_result ( value )


signature:
def asyncClose ( fn ) :
    """Allow to run async code before application is closed ."""

implementation:
@ functools . wraps ( fn )
def wrapper ( * args , ** kwargs ) :
    f = asyncio . ensure_future ( fn ( * args , ** kwargs ) )
    while not f . done ( ) :
        QApplication . instance ( ) . processEvents ( )
return wrapper


signature:
def asyncSlot ( * args ) :
    """Make a Qt async slot run on asyncio loop ."""

implementation:
def outer_decorator ( fn ) :
    @ Slot ( * args )
    @ functools . wraps ( fn )
    def wrapper ( * args , ** kwargs ) :
        asyncio . ensure_future ( fn ( * args , ** kwargs ) )
    return wrapper
return outer_decorator


signature:
def with_logger ( cls ) :
    """Class decorator to add a logger to a class ."""

implementation:
attr_name = '_logger'
cls_name = cls . __qualname__
module = cls . __module__
if module is not None :
    cls_name = module + '.' + cls_name
else :
    raise AssertionError
setattr ( cls , attr_name , logging . getLogger ( cls_name ) )
return cls


signature:
def _process_event ( self , key , mask ) :
    """Selector has delivered us an event ."""

implementation:
self . _logger . debug ( 'Processing event with key {} and mask {}' . format ( key , mask ) )
fileobj , ( reader , writer ) = key . fileobj , key . data
if mask & selectors . EVENT_READ and reader is not None :
    if reader . _cancelled :
        self . remove_reader ( fileobj )
    else :
        self . _logger . debug ( 'Invoking reader callback: {}' . format ( reader ) )
        reader . _run ( )
if mask & selectors . EVENT_WRITE and writer is not None :
    if writer . _cancelled :
        self . remove_writer ( fileobj )
    else :
        self . _logger . debug ( 'Invoking writer callback: {}' . format ( writer ) )
        writer . _run ( )


signature:
def addSources ( self , * sources ) :
    """Add more ASN . 1 MIB source repositories .
 MibCompiler . compile will invoke each of configured source objects
 in order of their addition asking each to fetch MIB module specified
 by name .
 Args :
 sources : reader object ( s )
 Returns :
 reference to itself ( can be used for call chaining )"""

implementation:
self . _sources . extend ( sources )
debug . logger & debug . flagCompiler and debug . logger ( 'current MIB source(s): %s' % ', ' . join ( [ str ( x ) for x in self . _sources ] ) )
return self


signature:
def addSearchers ( self , * searchers ) :
    """Add more transformed MIBs repositories .
 MibCompiler . compile will invoke each of configured searcher objects
 in order of their addition asking each if already transformed MIB
 module already exists and is more recent than specified .
 Args :
 searchers : searcher object ( s )
 Returns :
 reference to itself ( can be used for call chaining )"""

implementation:
self . _searchers . extend ( searchers )
debug . logger & debug . flagCompiler and debug . logger ( 'current compiled MIBs location(s): %s' % ', ' . join ( [ str ( x ) for x in self . _searchers ] ) )
return self


signature:
def addBorrowers ( self , * borrowers ) :
    """Add more transformed MIBs repositories to borrow MIBs from .
 Whenever MibCompiler . compile encounters MIB module which neither of
 the * searchers * can find or fetched ASN . 1 MIB module can not be
 parsed ( due to syntax errors ) , these * borrowers * objects will be
 invoked in order of their addition asking each if already transformed
 MIB can be fetched ( borrowed ) .
 Args :
 borrowers : borrower object ( s )
 Returns :
 reference to itself ( can be used for call chaining )"""

implementation:
self . _borrowers . extend ( borrowers )
debug . logger & debug . flagCompiler and debug . logger ( 'current MIB borrower(s): %s' % ', ' . join ( [ str ( x ) for x in self . _borrowers ] ) )
return self


signature:
def compile ( self , * mibnames , ** options ) :
    """Transform requested and possibly referred MIBs .
 The * compile * method should be invoked when * MibCompiler * object
 is operational meaning at least * sources * are specified .
 Once called with a MIB module name , * compile * will :
 * fetch ASN . 1 MIB module with given name by calling * sources *
 * make sure no such transformed MIB already exists ( with * searchers * )
 * parse ASN . 1 MIB text with * parser *
 * perform actual MIB transformation into target format with * code generator *
 * may attempt to borrow pre - transformed MIB through * borrowers *
 * write transformed MIB through * writer *
 The above sequence will be performed for each MIB name given in
 * mibnames * and may be performed for all MIBs referred to from
 MIBs being processed .
 Args :
 mibnames : list of ASN . 1 MIBs names
 options : options that affect the way PySMI components work
 Returns :
 A dictionary of MIB module names processed ( keys ) and * MibStatus *
 class instances ( values )"""

implementation:
processed = { }
parsedMibs = { }
failedMibs = { }
borrowedMibs = { }
builtMibs = { }
symbolTableMap = { }
mibsToParse = [ x for x in mibnames ]
canonicalMibNames = { }
while mibsToParse :
    mibname = mibsToParse . pop ( 0 )
    if mibname in parsedMibs :
        debug . logger & debug . flagCompiler and debug . logger ( 'MIB %s already parsed' % mibname )
        continue
    if mibname in failedMibs :
        debug . logger & debug . flagCompiler and debug . logger ( 'MIB %s already failed' % mibname )
        continue
    for source in self . _sources :
        debug . logger & debug . flagCompiler and debug . logger ( 'trying source %s' % source )
        try :
            fileInfo , fileData = source . getData ( mibname )
            for mibTree in self . _parser . parse ( fileData ) :
                mibInfo , symbolTable = self . _symbolgen . genCode ( mibTree , symbolTableMap )
                symbolTableMap [ mibInfo . name ] = symbolTable
                parsedMibs [ mibInfo . name ] = fileInfo , mibInfo , mibTree
                if mibname in failedMibs :
                    del failedMibs [ mibname ]
                mibsToParse . extend ( mibInfo . imported )
                if fileInfo . name in mibnames :
                    if mibInfo . name not in canonicalMibNames :
                        canonicalMibNames [ mibInfo . name ] = [ ]
                    canonicalMibNames [ mibInfo . name ] . append ( fileInfo . name )
                debug . logger & debug . flagCompiler and debug . logger ( '%s (%s) read from %s, immediate dependencies: %s' % ( mibInfo . name , mibname , fileInfo . path , ', ' . join ( mibInfo . imported ) or '<none>' ) )
            break
        except error . PySmiReaderFileNotFoundError :
            debug . logger & debug . flagCompiler and debug . logger ( 'no %s found at %s' % ( mibname , source ) )
            continue
        except error . PySmiError :
            exc_class , exc , tb = sys . exc_info ( )
            exc . source = source
            exc . mibname = mibname
            exc . msg += ' at MIB %s' % mibname
            debug . logger & debug . flagCompiler and debug . logger ( '%serror %s from %s' % ( options . get ( 'ignoreErrors' ) and 'ignoring ' or 'failing on ' , exc , source ) )
            failedMibs [ mibname ] = exc
            processed [ mibname ] = statusFailed . setOptions ( error = exc )
    else :
        exc = error . PySmiError ( 'MIB source %s not found' % mibname )
        exc . mibname = mibname
        debug . logger & debug . flagCompiler and debug . logger ( 'no %s found everywhere' % mibname )
        if mibname not in failedMibs :
            failedMibs [ mibname ] = exc
        if mibname not in processed :
            processed [ mibname ] = statusMissing
debug . logger & debug . flagCompiler and debug . logger ( 'MIBs analyzed %s, MIBs failed %s' % ( len ( parsedMibs ) , len ( failedMibs ) ) )
# See what MIBs need generating
for mibname in tuple ( parsedMibs ) :
    fileInfo , mibInfo , mibTree = parsedMibs [ mibname ]
    debug . logger & debug . flagCompiler and debug . logger ( 'checking if %s requires updating' % mibname )
    for searcher in self . _searchers :
        try :
            searcher . fileExists ( mibname , fileInfo . mtime , rebuild = options . get ( 'rebuild' ) )
        except error . PySmiFileNotFoundError :
            debug . logger & debug . flagCompiler and debug . logger ( 'no compiled MIB %s available through %s' % ( mibname , searcher ) )
            continue
        except error . PySmiFileNotModifiedError :
            debug . logger & debug . flagCompiler and debug . logger ( 'will be using existing compiled MIB %s found by %s' % ( mibname , searcher ) )
            del parsedMibs [ mibname ]
            processed [ mibname ] = statusUntouched
            break
        except error . PySmiError :
            exc_class , exc , tb = sys . exc_info ( )
            exc . searcher = searcher
            exc . mibname = mibname
            exc . msg += ' at MIB %s' % mibname
            debug . logger & debug . flagCompiler and debug . logger ( 'error from %s: %s' % ( searcher , exc ) )
            continue
    else :
        debug . logger & debug . flagCompiler and debug . logger ( 'no suitable compiled MIB %s found anywhere' % mibname )
        if options . get ( 'noDeps' ) and mibname not in canonicalMibNames :
            debug . logger & debug . flagCompiler and debug . logger ( 'excluding imported MIB %s from code generation' % mibname )
            del parsedMibs [ mibname ]
            processed [ mibname ] = statusUntouched
            continue
debug . logger & debug . flagCompiler and debug . logger ( 'MIBs parsed %s, MIBs failed %s' % ( len ( parsedMibs ) , len ( failedMibs ) ) )
# Generate code for parsed MIBs
for mibname in parsedMibs . copy ( ) :
    fileInfo , mibInfo , mibTree = parsedMibs [ mibname ]
    debug . logger & debug . flagCompiler and debug . logger ( 'compiling %s read from %s' % ( mibname , fileInfo . path ) )
    platform_info , user_info = self . _get_system_info ( )
    comments = [ 'ASN.1 source %s' % fileInfo . path , 'Produced by %s-%s at %s' % ( packageName , packageVersion , time . asctime ( ) ) , 'On host %s platform %s version %s by user %s' % ( platform_info [ 1 ] , platform_info [ 0 ] , platform_info [ 2 ] , user_info [ 0 ] ) , 'Using Python version %s' % sys . version . split ( '\n' ) [ 0 ] ]
    try :
        mibInfo , mibData = self . _codegen . genCode ( mibTree , symbolTableMap , comments = comments , dstTemplate = options . get ( 'dstTemplate' ) , genTexts = options . get ( 'genTexts' ) , textFilter = options . get ( 'textFilter' ) )
        builtMibs [ mibname ] = fileInfo , mibInfo , mibData
        del parsedMibs [ mibname ]
        debug . logger & debug . flagCompiler and debug . logger ( '%s read from %s and compiled by %s' % ( mibname , fileInfo . path , self . _writer ) )
    except error . PySmiError :
        exc_class , exc , tb = sys . exc_info ( )
        exc . handler = self . _codegen
        exc . mibname = mibname
        exc . msg += ' at MIB %s' % mibname
        debug . logger & debug . flagCompiler and debug . logger ( 'error from %s: %s' % ( self . _codegen , exc ) )
        processed [ mibname ] = statusFailed . setOptions ( error = exc )
        failedMibs [ mibname ] = exc
        del parsedMibs [ mibname ]
debug . logger & debug . flagCompiler and debug . logger ( 'MIBs built %s, MIBs failed %s' % ( len ( parsedMibs ) , len ( failedMibs ) ) )
# Try to borrow pre - compiled MIBs for failed ones
for mibname in failedMibs . copy ( ) :
    if options . get ( 'noDeps' ) and mibname not in canonicalMibNames :
        debug . logger & debug . flagCompiler and debug . logger ( 'excluding imported MIB %s from borrowing' % mibname )
        continue
    for borrower in self . _borrowers :
        debug . logger & debug . flagCompiler and debug . logger ( 'trying to borrow %s from %s' % ( mibname , borrower ) )
        try :
            fileInfo , fileData = borrower . getData ( mibname , genTexts = options . get ( 'genTexts' ) )
            borrowedMibs [ mibname ] = fileInfo , MibInfo ( name = mibname , imported = [ ] ) , fileData
            del failedMibs [ mibname ]
            debug . logger & debug . flagCompiler and debug . logger ( '%s borrowed with %s' % ( mibname , borrower ) )
            break
        except error . PySmiError :
            debug . logger & debug . flagCompiler and debug . logger ( 'error from %s: %s' % ( borrower , sys . exc_info ( ) [ 1 ] ) )
debug . logger & debug . flagCompiler and debug . logger ( 'MIBs available for borrowing %s, MIBs failed %s' % ( len ( borrowedMibs ) , len ( failedMibs ) ) )
# See what MIBs need borrowing
for mibname in borrowedMibs . copy ( ) :
    debug . logger & debug . flagCompiler and debug . logger ( 'checking if failed MIB %s requires borrowing' % mibname )
    fileInfo , mibInfo , mibData = borrowedMibs [ mibname ]
    for searcher in self . _searchers :
        try :
            searcher . fileExists ( mibname , fileInfo . mtime , rebuild = options . get ( 'rebuild' ) )
        except error . PySmiFileNotFoundError :
            debug . logger & debug . flagCompiler and debug . logger ( 'no compiled MIB %s available through %s' % ( mibname , searcher ) )
            continue
        except error . PySmiFileNotModifiedError :
            debug . logger & debug . flagCompiler and debug . logger ( 'will be using existing compiled MIB %s found by %s' % ( mibname , searcher ) )
            del borrowedMibs [ mibname ]
            processed [ mibname ] = statusUntouched
            break
        except error . PySmiError :
            exc_class , exc , tb = sys . exc_info ( )
            exc . searcher = searcher
            exc . mibname = mibname
            exc . msg += ' at MIB %s' % mibname
            debug . logger & debug . flagCompiler and debug . logger ( 'error from %s: %s' % ( searcher , exc ) )
            continue
    else :
        debug . logger & debug . flagCompiler and debug . logger ( 'no suitable compiled MIB %s found anywhere' % mibname )
        if options . get ( 'noDeps' ) and mibname not in canonicalMibNames :
            debug . logger & debug . flagCompiler and debug . logger ( 'excluding imported MIB %s from borrowing' % mibname )
            processed [ mibname ] = statusUntouched
        else :
            debug . logger & debug . flagCompiler and debug . logger ( 'will borrow MIB %s' % mibname )
            builtMibs [ mibname ] = borrowedMibs [ mibname ]
            processed [ mibname ] = statusBorrowed . setOptions ( path = fileInfo . path , file = fileInfo . file , alias = fileInfo . name )
        del borrowedMibs [ mibname ]
debug . logger & debug . flagCompiler and debug . logger ( 'MIBs built %s, MIBs failed %s' % ( len ( builtMibs ) , len ( failedMibs ) ) )
# We could attempt to ignore missing / failed MIBs
if failedMibs and not options . get ( 'ignoreErrors' ) :
    debug . logger & debug . flagCompiler and debug . logger ( 'failing with problem MIBs %s' % ', ' . join ( failedMibs ) )
    for mibname in builtMibs :
        processed [ mibname ] = statusUnprocessed
    return processed
debug . logger & debug . flagCompiler and debug . logger ( 'proceeding with built MIBs %s, failed MIBs %s' % ( ', ' . join ( builtMibs ) , ', ' . join ( failedMibs ) ) )
# Store compiled MIBs
for mibname in builtMibs . copy ( ) :
    fileInfo , mibInfo , mibData = builtMibs [ mibname ]
    try :
        if options . get ( 'writeMibs' , True ) :
            self . _writer . putData ( mibname , mibData , dryRun = options . get ( 'dryRun' ) )
        debug . logger & debug . flagCompiler and debug . logger ( '%s stored by %s' % ( mibname , self . _writer ) )
        del builtMibs [ mibname ]
        if mibname not in processed :
            processed [ mibname ] = statusCompiled . setOptions ( path = fileInfo . path , file = fileInfo . file , alias = fileInfo . name , oid = mibInfo . oid , oids = mibInfo . oids , identity = mibInfo . identity , revision = mibInfo . revision , enterprise = mibInfo . enterprise , compliance = mibInfo . compliance , )
    except error . PySmiError :
        exc_class , exc , tb = sys . exc_info ( )
        exc . handler = self . _codegen
        exc . mibname = mibname
        exc . msg += ' at MIB %s' % mibname
        debug . logger & debug . flagCompiler and debug . logger ( 'error %s from %s' % ( exc , self . _writer ) )
        processed [ mibname ] = statusFailed . setOptions ( error = exc )
        failedMibs [ mibname ] = exc
        del builtMibs [ mibname ]
debug . logger & debug . flagCompiler and debug . logger ( 'MIBs modified: %s' % ', ' . join ( [ x for x in processed if processed [ x ] in ( 'compiled' , 'borrowed' ) ] ) )
return processed


signature:
def t_UPPERCASE_IDENTIFIER ( self , t ) :
    r'[ A - Z ] [ - a - zA - z0-9 ] *'

implementation:
if t . value in self . forbidden_words :
    raise error . PySmiLexerError ( "%s is forbidden" % t . value , lineno = t . lineno )
if t . value [ - 1 ] == '-' :
    raise error . PySmiLexerError ( "Identifier should not end with '-': %s" % t . value , lineno = t . lineno )
t . type = self . reserved . get ( t . value , 'UPPERCASE_IDENTIFIER' )
return t


signature:
def t_LOWERCASE_IDENTIFIER ( self , t ) :
    r'[0-9 ] * [ a - z ] [ - a - zA - z0-9 ] *'

implementation:
if t . value [ - 1 ] == '-' :
    raise error . PySmiLexerError ( "Identifier should not end with '-': %s" % t . value , lineno = t . lineno )
return t


signature:
def t_NUMBER ( self , t ) :
    r'- ? [ 0-9 ] +'

implementation:
t . value = int ( t . value )
neg = 0
if t . value < 0 :
    neg = 1
val = abs ( t . value )
if val <= UNSIGNED32_MAX :
    if neg :
        t . type = 'NEGATIVENUMBER'
elif val <= UNSIGNED64_MAX :
    if neg :
        t . type = 'NEGATIVENUMBER64'
    else :
        t . type = 'NUMBER64'
else :
    raise error . PySmiLexerError ( "Number %s is too big" % t . value , lineno = t . lineno )
return t


signature:
def t_BIN_STRING ( self , t ) :
    r'\ ' [ 01 ] * \ ' [ bB ]'

implementation:
value = t . value [ 1 : - 2 ]
while value and value [ 0 ] == '0' and len ( value ) % 8 :
    value = value [ 1 : ]
# XXX raise in strict mode
# if len ( value ) % 8:
# raise error . PySmiLexerError ( " Number of 0s and 1s have to divide by 8 in binary string % s " % t . value , lineno = t . lineno )
return t


signature:
def t_HEX_STRING ( self , t ) :
    r'\ ' [ 0-9a - fA - F ] * \ ' [ hH ]'

implementation:
value = t . value [ 1 : - 2 ]
while value and value [ 0 ] == '0' and len ( value ) % 2 :
    value = value [ 1 : ]
# XXX raise in strict mode
# if len ( value ) % 2:
# raise error . PySmiLexerError ( " Number of symbols have to be even in hex string % s " % t . value , lineno = t . lineno )
return t


signature:
def p_modules ( self , p ) :
    """modules : modules module
 | module"""


signature:
def p_importPart ( self , p ) :
    """importPart : imports
 | empty"""

implementation:
# libsmi : TODO : ` ` IMPORTS ; ' ' allowed ? refer ASN . 1!
if p [ 1 ] :
    importDict = { }
    for imp in p [ 1 ] : # don ' t do just dict ( ) because moduleNames may be repeated
        fromModule , symbols = imp
        if fromModule in importDict :
            importDict [ fromModule ] += symbols
        else :
            importDict [ fromModule ] = symbols
    p [ 0 ] = importDict


signature:
def p_imports ( self , p ) :
    """imports : imports import
 | import"""


signature:
def p_importIdentifiers ( self , p ) :
    """importIdentifiers : importIdentifiers ' , ' importIdentifier
 | importIdentifier"""


signature:
def p_declarations ( self , p ) :
    """declarations : declarations declaration
 | declaration"""


signature:
def p_typeDeclarationRHS ( self , p ) :
    """typeDeclarationRHS : Syntax
 | TEXTUAL _ CONVENTION DisplayPart STATUS Status DESCRIPTION Text ReferPart SYNTAX Syntax
 | choiceClause"""

implementation:
if p [ 1 ] :
    if p [ 1 ] == 'TEXTUAL-CONVENTION' :
        p [ 0 ] = ( 'typeDeclarationRHS' , p [ 2 ] , # display
        p [ 4 ] , # status
        ( p [ 5 ] , p [ 6 ] ) , # description
        p [ 7 ] , # reference
        p [ 9 ] )
        # syntax
    else :
        p [ 0 ] = ( 'typeDeclarationRHS' , p [ 1 ] )


signature:
def p_sequenceItems ( self , p ) :
    """sequenceItems : sequenceItems ' , ' sequenceItem
 | sequenceItem"""

implementation:
# libsmi : TODO : might this list be emtpy ?
n = len ( p )
if n == 4 :
    p [ 0 ] = p [ 1 ] + [ p [ 3 ] ]
elif n == 2 :
    p [ 0 ] = [ p [ 1 ] ]


signature:
def p_Syntax ( self , p ) :
    """Syntax : ObjectSyntax
 | BITS ' { ' NamedBits ' } '"""

implementation:
# libsmi : TODO : standalone ` BITS ' ok ? seen in RMON2 - MIB
# libsmi : - > no , it ' s only allowed in a SEQUENCE { . . . }
n = len ( p )
if n == 2 :
    p [ 0 ] = p [ 1 ]
elif n == 5 :
    p [ 0 ] = ( p [ 1 ] , p [ 3 ] )


signature:
def p_NamedBits ( self , p ) :
    """NamedBits : NamedBits ' , ' NamedBit
 | NamedBit"""


signature:
def p_objectIdentityClause ( self , p ) :
    """objectIdentityClause : LOWERCASE _ IDENTIFIER OBJECT _ IDENTITY STATUS Status DESCRIPTION Text ReferPart COLON _ COLON _ EQUAL ' { ' objectIdentifier ' } '"""

implementation:
p [ 0 ] = ( 'objectIdentityClause' , p [ 1 ] , # id
# p [ 2 ] , # OBJECT _ IDENTITY
p [ 4 ] , # status
( p [ 5 ] , p [ 6 ] ) , # description
p [ 7 ] , # reference
p [ 10 ] )


signature:
def p_objectTypeClause ( self , p ) :
    """objectTypeClause : LOWERCASE _ IDENTIFIER OBJECT _ TYPE SYNTAX Syntax UnitsPart MaxOrPIBAccessPart STATUS Status descriptionClause ReferPart IndexPart MibIndex DefValPart COLON _ COLON _ EQUAL ' { ' ObjectName ' } '"""

implementation:
p [ 0 ] = ( 'objectTypeClause' , p [ 1 ] , # id
# p [ 2 ] , # OBJECT _ TYPE
p [ 4 ] , # syntax
p [ 5 ] , # UnitsPart
p [ 6 ] , # MaxOrPIBAccessPart
p [ 8 ] , # status
p [ 9 ] , # descriptionClause
p [ 10 ] , # reference
p [ 11 ] , # augmentions
p [ 12 ] , # index
p [ 13 ] , # DefValPart
p [ 16 ] )


signature:
def p_VarTypes ( self , p ) :
    """VarTypes : VarTypes ' , ' VarType
 | VarType"""