das_client Namespace Reference

Classes
class	DASOptionParser
class	HTTPSClientAuthHandler

Functions
def	check_auth (key)
def	check_glidein ()
def	convert_time (val)
def	extract_value (row, key, base=10)
def	fullpath (path)
def	get_data (host, query, idx, limit, debug, threshold=300, ckey=None, cert=None, capath=None, qcache=0, das_headers=True)
def	get_value (data, filters, base=10)
def	keys_attrs (lkey, oformat, host, ckey, cert, debug=0)
def	main ()
def	prim_value (row)
def	print_from_cache (cache, query)
def	print_summary (rec)
def	size_format (uinput, ibase=0)
def	unique_filter (rows)
def	x509 ()

Variables
	__author__
	DAS_CLIENT
	EX__BASE
	EX_CANTCREAT
	EX_CONFIG
	EX_DATAERR
	EX_IOERR
	EX_NOHOST
	EX_NOINPUT
	EX_NOPERM
	EX_NOUSER
	EX_OK
	EX_OSERR
	EX_OSFILE
	EX_PROTOCOL
	EX_SOFTWARE
	EX_TEMPFAIL
	EX_UNAVAILABLE
	EX_USAGE

Function Documentation

check_auth()

def das_client.check_auth

(

key

)

Definition at line 102 of file das_client.py.

References print().

Referenced by main().

def check_auth(key):
    "Check if user runs das_client with key/cert and warn users to switch"
    if  not key:
        msg  = "WARNING: das_client is running without user credentials/X509 proxy, create proxy via 'voms-proxy-init -voms cms -rfc'"
        print(msg, file=sys.stderr)

check_glidein()

def das_client.check_glidein

(

)

Definition at line 94 of file das_client.py.

References print().

Referenced by main().

def check_glidein():
    "Check glideine environment and exit if it is set"
    glidein = os.environ.get('GLIDEIN_CMSSite', '')
    if  glidein:
        msg = "ERROR: das_client is running from GLIDEIN environment, it is prohibited"
        print(msg)
        sys.exit(EX__BASE)

convert_time()

def das_client.convert_time

(

val

)

Definition at line 176 of file das_client.py.

Referenced by extract_value().

def convert_time(val):
    "Convert given timestamp into human readable format"
    if  isinstance(val, int) or isinstance(val, float):
        return time.strftime('%d/%b/%Y_%H:%M:%S_GMT', time.gmtime(val))
    return val

extract_value()

def das_client.extract_value	(		row,
			key,
			base = 10
	)

Generator which extracts row[key] value

Definition at line 233 of file das_client.py.

References convert_time(), and size_format().

Referenced by get_value().

def extract_value(row, key, base=10):
    """Generator which extracts row[key] value"""
    if  isinstance(row, dict) and key in row:
        if  key == 'creation_time':
            row = convert_time(row[key])
        elif  key == 'size':
            row = size_format(row[key], base)
        else:
            row = row[key]
        yield row
    if  isinstance(row, list) or isinstance(row, GeneratorType):
        for item in row:
            for vvv in extract_value(item, key, base):
                yield vvv

fullpath()

def das_client.fullpath

(

path

)

Definition at line 267 of file das_client.py.

References get_data().

Referenced by get_data(), and keys_attrs().

def fullpath(path):
    "Expand path to full path"
    if  path and path[0] == '~':
        path = path.replace('~', '')
        path = path[1:] if path[0] == '/' else path
        path = os.path.join(os.environ['HOME'], path)
    return path

get_data()

def das_client.get_data	(		host,
			query,
			idx,
			limit,
			debug,
			threshold = 300,
			ckey = None,
			cert = None,
			capath = None,
			qcache = 0,
			das_headers = True
	)

Contact DAS server and retrieve data for given DAS query

Definition at line 276 of file das_client.py.

References fullpath(), createfilelist.int, print(), and str.

Referenced by cmsswVersionTools.PickRelValInputFiles.apply(), electronDataDiscovery.common_search(), dasFileQuery.dasFileQuery(), dataset.dasquery(), util.getRunInfo.dasQuery(), fullpath(), edmPickEvents.getFileNames_das_client(), and main().

        cert=None, capath=None, qcache=0, das_headers=True):
    """Contact DAS server and retrieve data for given DAS query"""
    params  = {'input':query, 'idx':idx, 'limit':limit}
    if  qcache:
        params['qcache'] = qcache
    path    = '/das/cache'
    pat     = re.compile('http[s]{0,1}://')
    if  not pat.match(host):
        msg = 'Invalid hostname: %s' % host
        raise Exception(msg)
    url = host + path
    client = '%s (%s)' % (DAS_CLIENT, os.environ.get('USER', ''))
    headers = {"Accept": "application/json", "User-Agent": client}
    encoded_data = urllib.urlencode(params, doseq=True)
    url += '?%s' % encoded_data
    req  = urllib2.Request(url=url, headers=headers)
    if  ckey and cert:
        ckey = fullpath(ckey)
        cert = fullpath(cert)
        http_hdlr  = HTTPSClientAuthHandler(ckey, cert, capath, debug)
    elif cert and capath:
        cert = fullpath(cert)
        http_hdlr  = HTTPSClientAuthHandler(ckey, cert, capath, debug)
    else:
        http_hdlr  = urllib2.HTTPHandler(debuglevel=debug)
    proxy_handler  = urllib2.ProxyHandler({})
    cookie_jar     = cookielib.CookieJar()
    cookie_handler = urllib2.HTTPCookieProcessor(cookie_jar)
    try:
        opener = urllib2.build_opener(http_hdlr, proxy_handler, cookie_handler)
        fdesc = opener.open(req)
        data = fdesc.read()
        fdesc.close()
    except urllib2.HTTPError as error:
    print(error.read())
    sys.exit(1)

    pat = re.compile(r'^[a-z0-9]{32}')
    if  data and isinstance(data, str) and pat.match(data) and len(data) == 32:
        pid = data
    else:
        pid = None
    iwtime  = 2  # initial waiting time in seconds
    wtime   = 20 # final waiting time in seconds
    sleep   = iwtime
    time0   = time.time()
    while pid:
        params.update({'pid':data})
        encoded_data = urllib.urlencode(params, doseq=True)
        url  = host + path + '?%s' % encoded_data
        req  = urllib2.Request(url=url, headers=headers)
        try:
            fdesc = opener.open(req)
            data = fdesc.read()
            fdesc.close()
        except urllib2.HTTPError as err:
            return {"status":"fail", "reason":str(err)}
        if  data and isinstance(data, str) and pat.match(data) and len(data) == 32:
            pid = data
        else:
            pid = None
        time.sleep(sleep)
        if  sleep < wtime:
            sleep *= 2
        elif sleep == wtime:
            sleep = iwtime # start new cycle
        else:
            sleep = wtime
        if  (time.time()-time0) > threshold:
            reason = "client timeout after %s sec" % int(time.time()-time0)
            return {"status":"fail", "reason":reason}
    jsondict = json.loads(data)
    return jsondict

get_value()

def das_client.get_value	(		data,
			filters,
			base = 10
	)

Filter data from a row for given list of filters

Definition at line 248 of file das_client.py.

References extract_value().

Referenced by cmsswVersionTools.PickRelValInputFiles.apply(), edmPickEvents.getFileNames_das_client(), and main().

def get_value(data, filters, base=10):
    """Filter data from a row for given list of filters"""
    for ftr in filters:
        if  ftr.find('>') != -1 or ftr.find('<') != -1 or ftr.find('=') != -1:
            continue
        row = dict(data)
        values = []
        keys = ftr.split('.')
        for key in keys:
            val = [v for v in extract_value(row, key, base)]
            if  key == keys[-1]: # we collect all values at last key
                values += [json.dumps(i) for i in val]
            else:
                row = val
        if  len(values) == 1:
            yield values[0]
        else:
            yield values

keys_attrs()

def das_client.keys_attrs	(		lkey,
			oformat,
			host,
			ckey,
			cert,
			debug = 0
	)

Definition at line 387 of file das_client.py.

References fullpath(), and print().

Referenced by main().

def keys_attrs(lkey, oformat, host, ckey, cert, debug=0):
    "Contact host for list of key/attributes pairs"
    url = '%s/das/keys?view=json' % host
    headers = {"Accept": "application/json", "User-Agent": DAS_CLIENT}
    req  = urllib2.Request(url=url, headers=headers)
    if  ckey and cert:
        ckey = fullpath(ckey)
        cert = fullpath(cert)
        http_hdlr  = HTTPSClientAuthHandler(ckey, cert, debug)
    else:
        http_hdlr  = urllib2.HTTPHandler(debuglevel=debug)
    proxy_handler  = urllib2.ProxyHandler({})
    cookie_jar     = cookielib.CookieJar()
    cookie_handler = urllib2.HTTPCookieProcessor(cookie_jar)
    opener = urllib2.build_opener(http_hdlr, proxy_handler, cookie_handler)
    fdesc = opener.open(req)
    data = json.load(fdesc)
    fdesc.close()
    if  oformat.lower() == 'json':
        if  lkey == 'all':
            print(json.dumps(data))
        else:
            print(json.dumps({lkey:data[lkey]}))
        return
    for key, vdict in data.items():
        if  lkey == 'all':
            pass
        elif lkey != key:
            continue
        print()
        print("DAS key:", key)
        for attr, examples in vdict.items():
            prefix = '    '
            print('%s%s' % (prefix, attr))
            for item in examples:
                print('%s%s%s' % (prefix, prefix, item))

main()

def das_client.main

(

)

Main function

Definition at line 424 of file das_client.py.

References check_auth(), check_glidein(), spr.find(), get_data(), get_value(), createfilelist.int, join(), keys_attrs(), dqm-mbProfile.log, prim_value(), print(), print_from_cache(), print_summary(), size_format(), str, and unique_filter().

def main():
    """Main function"""
    optmgr  = DASOptionParser()
    opts, _ = optmgr.get_opt()
    host    = opts.host
    debug   = opts.verbose
    query   = opts.query
    idx     = opts.idx
    limit   = opts.limit
    thr     = opts.threshold
    ckey    = opts.ckey
    cert    = opts.cert
    capath  = opts.capath
    base    = opts.base
    qcache  = opts.qcache
    check_glidein()
    check_auth(ckey)
    if  opts.keys_attrs:
        keys_attrs(opts.keys_attrs, opts.format, host, ckey, cert, debug)
        return
    if  not query:
        print('Input query is missing')
        sys.exit(EX_USAGE)
    if  opts.format == 'plain':
        jsondict = get_data(host, query, idx, limit, debug, thr, ckey, cert, capath, qcache)
        cli_msg  = jsondict.get('client_message', None)
        if  cli_msg:
            print("DAS CLIENT WARNING: %s" % cli_msg)
        if  'status' not in jsondict and opts.cache:
            print_from_cache(opts.cache, query)
        if  'status' not in jsondict:
            print('DAS record without status field:\n%s' % jsondict)
            sys.exit(EX_PROTOCOL)
        if  jsondict["status"] != 'ok' and opts.cache:
            print_from_cache(opts.cache, query)
        if  jsondict['status'] != 'ok':
            print("status: %s, reason: %s" \
                % (jsondict.get('status'), jsondict.get('reason', 'N/A')))
            if  opts.retry:
                found = False
                for attempt in xrange(1, int(opts.retry)):
                    interval = log(attempt)**5
                    print("Retry in %5.3f sec" % interval)
                    time.sleep(interval)
                    data = get_data(host, query, idx, limit, debug, thr, ckey, cert, capath, qcache)
                    jsondict = json.loads(data)
                    if  jsondict.get('status', 'fail') == 'ok':
                        found = True
                        break
            else:
                sys.exit(EX_TEMPFAIL)
            if  not found:
                sys.exit(EX_TEMPFAIL)
        nres = jsondict.get('nresults', 0)
        if  not limit:
            drange = '%s' % nres
        else:
            drange = '%s-%s out of %s' % (idx+1, idx+limit, nres)
        if  opts.limit:
            msg  = "\nShowing %s results" % drange
            msg += ", for more results use --idx/--limit options\n"
            print(msg)
        mongo_query = jsondict.get('mongo_query', {})
        unique  = False
        fdict   = mongo_query.get('filters', {})
        filters = fdict.get('grep', [])
        aggregators = mongo_query.get('aggregators', [])
        if  'unique' in fdict.keys():
            unique = True
        if  filters and not aggregators:
            data = jsondict['data']
            if  isinstance(data, dict):
                rows = [r for r in get_value(data, filters, base)]
                print(' '.join(rows))
            elif isinstance(data, list):
                if  unique:
                    data = unique_filter(data)
                for row in data:
                    rows = [r for r in get_value(row, filters, base)]
                    types = [type(r) for r in rows]
                    if  len(types)>1: # mixed types print as is
                        print(' '.join([str(r) for r in rows]))
                    elif isinstance(rows[0], list):
                        out = set()
                        for item in rows:
                            for elem in item:
                                out.add(elem)
                        print(' '.join(out))
                    else:
                        print(' '.join(rows))
            else:
                print(json.dumps(jsondict))
        elif aggregators:
            data = jsondict['data']
            if  unique:
                data = unique_filter(data)
            for row in data:
                if  row['key'].find('size') != -1 and \
                    row['function'] == 'sum':
                    val = size_format(row['result']['value'], base)
                else:
                    val = row['result']['value']
                print('%s(%s)=%s' \
                % (row['function'], row['key'], val))
        else:
            data = jsondict['data']
            if  isinstance(data, list):
                old = None
                val = None
                for row in data:
                    prim_key = row.get('das', {}).get('primary_key', None)
                    if  prim_key == 'summary':
                        print_summary(row)
                        return
                    val = prim_value(row)
                    if  not opts.limit:
                        if  val != old:
                            print(val)
                            old = val
                    else:
                        print(val)
                if  val != old and not opts.limit:
                    print(val)
            elif isinstance(data, dict):
                print(prim_value(data))
            else:
                print(data)
    else:
        jsondict = get_data(\
                host, query, idx, limit, debug, thr, ckey, cert, capath, qcache)
        print(json.dumps(jsondict))

#
# main
#

prim_value()

def das_client.prim_value

(

row

)

Extract primary key value from DAS record

Definition at line 350 of file das_client.py.

Referenced by main().

def prim_value(row):
    """Extract primary key value from DAS record"""
    prim_key = row['das']['primary_key']
    if  prim_key == 'summary':
        return row.get(prim_key, None)
    key, att = prim_key.split('.')
    if  isinstance(row[key], list):
        for item in row[key]:
            if  att in item:
                return item[att]
    else:
        if  key in row:
            if  att in row[key]:
                return row[key][att]

print_from_cache()

def das_client.print_from_cache	(		cache,
			query
	)

Definition at line 378 of file das_client.py.

References beamvalidation.exit(), join(), print(), and fileinputsource_cfi.read.

Referenced by main().

def print_from_cache(cache, query):
    "print the list of files reading it from cache"
    data = open(cache).read()
    jsondict = json.loads(data)
    if query in jsondict:
      print("\n".join(jsondict[query]))
      exit(0)
    exit(1)

print_summary()

def das_client.print_summary

(

rec

)

Definition at line 365 of file das_client.py.

References SiStripPI.max, and print().

Referenced by main().

def print_summary(rec):
    "Print summary record information on stdout"
    if  'summary' not in rec:
        msg = 'Summary information is not found in record:\n', rec
        raise Exception(msg)
    for row in rec['summary']:
        keys = [k for k in row.keys()]
        maxlen = max([len(k) for k in keys])
        for key, val in row.items():
            pkey = '%s%s' % (key, ' '*(maxlen-len(key)))
            print('%s: %s' % (pkey, val))
        print()

size_format()

def das_client.size_format	(		uinput,
			ibase = 0
	)

Format file size utility, it converts file size into KB, MB, GB, TB, PB units

Definition at line 182 of file das_client.py.

References nano_mu_digi_cff.float.

Referenced by extract_value(), and main().

def size_format(uinput, ibase=0):
    """
    Format file size utility, it converts file size into KB, MB, GB, TB, PB units
    """
    if  not ibase:
        return uinput
    try:
        num = float(uinput)
    except Exception as _exc:
        return uinput
    if  ibase == 2.: # power of 2
        base  = 1024.
        xlist = ['', 'KiB', 'MiB', 'GiB', 'TiB', 'PiB']
    else: # default base is 10
        base  = 1000.
        xlist = ['', 'KB', 'MB', 'GB', 'TB', 'PB']
    for xxx in xlist:
        if  num < base:
            return "%3.1f%s" % (num, xxx)
        num /= base

unique_filter()

def das_client.unique_filter

(

rows

)

Unique filter drop duplicate rows.

Definition at line 203 of file das_client.py.

Referenced by main().

def unique_filter(rows):
    """
    Unique filter drop duplicate rows.
    """
    old_row = {}
    row = None
    for row in rows:
        row_data = dict(row)
        try:
            del row_data['_id']
            del row_data['das']
            del row_data['das_id']
            del row_data['cache_id']
        except:
            pass
        old_data = dict(old_row)
        try:
            del old_data['_id']
            del old_data['das']
            del old_data['das_id']
            del old_data['cache_id']
        except:
            pass
        if  row_data == old_data:
            continue
        if  old_row:
            yield old_row
        old_row = row
    yield row

x509()

def das_client.x509

(

)

Definition at line 85 of file das_client.py.

def x509():
    "Helper function to get x509 either from env or tmp file"
    proxy = os.environ.get('X509_USER_PROXY', '')
    if  not proxy:
        proxy = '/tmp/x509up_u%s' % pwd.getpwuid( os.getuid() ).pw_uid
        if  not os.path.isfile(proxy):
            return ''
    return proxy

Variable Documentation

__author__

das_client.__author__

private

Definition at line 9 of file das_client.py.

DAS_CLIENT

das_client.DAS_CLIENT

Definition at line 18 of file das_client.py.

EX__BASE

das_client.EX__BASE

Definition at line 35 of file das_client.py.

EX_CANTCREAT

das_client.EX_CANTCREAT

Definition at line 45 of file das_client.py.

EX_CONFIG

das_client.EX_CONFIG

Definition at line 50 of file das_client.py.

EX_DATAERR

das_client.EX_DATAERR

Definition at line 37 of file das_client.py.

EX_IOERR

das_client.EX_IOERR

Definition at line 46 of file das_client.py.

EX_NOHOST

das_client.EX_NOHOST

Definition at line 40 of file das_client.py.

EX_NOINPUT

das_client.EX_NOINPUT

Definition at line 38 of file das_client.py.

EX_NOPERM

das_client.EX_NOPERM

Definition at line 49 of file das_client.py.

EX_NOUSER

das_client.EX_NOUSER

Definition at line 39 of file das_client.py.

EX_OK

das_client.EX_OK

Definition at line 34 of file das_client.py.

EX_OSERR

das_client.EX_OSERR

Definition at line 43 of file das_client.py.

EX_OSFILE

das_client.EX_OSFILE

Definition at line 44 of file das_client.py.

EX_PROTOCOL

das_client.EX_PROTOCOL

Definition at line 48 of file das_client.py.

EX_SOFTWARE

das_client.EX_SOFTWARE

Definition at line 42 of file das_client.py.

EX_TEMPFAIL

das_client.EX_TEMPFAIL

Definition at line 47 of file das_client.py.

EX_UNAVAILABLE

das_client.EX_UNAVAILABLE

Definition at line 41 of file das_client.py.

EX_USAGE

das_client.EX_USAGE

Definition at line 36 of file das_client.py.