stream_remote_resources.py

importos
importlogging
importtime
fromdatetimeimportdate, timeasdatetime_time
importitertools
fromdecimalimportDecimal
importrequests

importtabulator

fromtableschemaimportSchema

fromdataflows.helpers.resource_matcherimportResourceMatcher

fromdatapackage_pipelines.wrapperimportingest, spew
fromdatapackage_pipelines.utilities.resourcesimportstreamable, PATH_PLACEHOLDER, get_path, \
PROP_STREAMED_FROM, PROP_STREAMING, streaming
fromdatapackage_pipelines.utilities.extended_jsonimportjson
fromdatapackage_pipelines.utilities.tabulator_txt_parserimportTXTParser


def_tostr(value):
ifisinstance(value, str):
returnvalue
elifvalueisNone:
return''
elifisinstance(value, (int, float, bool, Decimal)):
returnstr(value)
elifisinstance(value, date):
returnvalue.isoformat()
elifisinstance(value, datetime_time):
returnvalue.isoformat()
elifisinstance(value, (list, dict)):
returnjson.dumps(value)

assertFalse, "Internal error - don't know how to handle %r of type %r"% (value, type(value))


def_reader(opener, _url, max_row=-1):
yieldNone
filename=os.path.basename(_url)
logging.info('%s: OPENING %s', filename, _url)
_schema, _headers, _, _reader, _close=opener()
num_headers=len(_headers)
i=0
fori, rowinenumerate(_reader):
row= [_tostr(x).strip() forxinrow]
values=set(row)
iflen(values) ==1and''invalues:
# In case of empty rows, just skip them
continue
output=dict(zip(_headers, _schema.cast_row(row[:num_headers])))
yieldoutput

i+=1
ifi%10000==0:
logging.info('%s: %d rows', filename, i)
# break
ifi==max_row:
break

_close()
logging.info('%s: TOTAL %d rows', filename, i)


defdedupe(headers):
_dedupped_headers= []
forhdrinheaders:
ifhdrisNone:
break
hdr=str(hdr).strip()
iflen(hdr) ==0:
continue
ifhdrin_dedupped_headers:
i=0
deduped_hdr=hdr
whilededuped_hdrin_dedupped_headers:
i+=1
deduped_hdr='%s_%s'% (hdr, i)
hdr=deduped_hdr
_dedupped_headers.append(hdr)
return_dedupped_headers


defadd_constants(extra_headers, extra_values, columns):
def_func(extended_rows):
fornumber, headers, rowinextended_rows:
ifcolumnsisNone:
row=row[:len(headers)] +extra_values
else:
row=row[:columns] +extra_values
yieldnumber, headers+extra_headers, row
return_func


defsuffix_remover(format):
defempty_row(row):
returnall(visNoneor (isinstance(v, str) andv=='')
forvinrow)

def_func(extended_rows):
suffix=False
fornumber, headers, rowinextended_rows:
ifformat=='txt':
yieldnumber, headers, row
continue

ifsuffix:
iflen(row) >=len(headers):
logging.warning('Expected an empty row, but got %r instead'%row)
else:
ifempty_row(row):
continue
eliflen(row) <len(headers):
suffix=True
else:
yieldnumber, headers, row

return_func


defstream_reader(_resource, _url, _ignore_missing, limit_rows, http_headers):
defget_opener(__url, __resource, columns=None):
_columns=columns

defopener():
_params=dict(headers=1)
format=__resource.get("format")
ifformat=="txt":
# datapackage-pipelines processing requires having a header row
# for txt format we add a single "data" column
_params["headers"] = ["data"]
_params["custom_parsers"] = {"txt": TXTParser}
_params["allow_html"] =True
else:
ifformatisNone:
_, format=tabulator.helpers.detect_scheme_and_format(__url)
ifformatintabulator.config.SUPPORTED_COMPRESSION:
format=None
else:
try:
parser_cls=tabulator.helpers.import_attribute(tabulator.config.PARSERS[format])
exceptKeyError:
logging.error("Unknown format %r", format)
raise
_params.update(
dict(xforxin__resource.items()
ifx[0] inparser_cls.options))
_params.update(
dict(xforxin__resource.items()
ifx[0] in {'headers', 'scheme', 'encoding', 'sample_size', 'allow_html',
'force_strings', 'force_parse', 'skip_rows', 'compression',
'http_timeout'}))
ifisinstance(_params.get('skip_rows'), int): # Backwards compatibility
_params['skip_rows'] =list(range(1, _params.get('skip_rows') +1))

ifformatisnotNone:
_params['format'] =format

ifhttp_headers:
http_session=requests.Session()
http_session.headers=http_headers
_params['http_session'] =http_session

constants=_resource.get('constants', {})
constant_headers=list(constants.keys())
constant_values= [constants.get(k) forkinconstant_headers]
_stream=tabulator.Stream(__url, **_params,
post_parse=[suffix_remover(format),
add_constants(constant_headers, constant_values, _columns)])
retry=0
backoff=2
whileTrue:
try:
_stream.open()
_headers=dedupe(_stream.headers)
__columns=len(_headers)
_headers=dedupe(_headers+constant_headers)
_schema=__resource.get('schema')
if_schemaisnotNone:
_schema=Schema(_schema)
return_schema, _headers, __columns, _stream, _stream.close
excepttabulator.exceptions.TabulatorExceptionase:
logging.warning("Error while opening resource from url %s: %r",
_url, e)
_stream.close()
retry+=1
ifretry<=3:
logging.warning("Retrying after %d seconds (%d/3)", backoff, retry)
time.sleep(backoff)
backoff*=2
continue
else:
ifnot_ignore_missing:
raise
return {}, [], 0, [], lambda: None
returnopener

schema, headers, columns, stream, close=get_opener(url, _resource)()
ifschemaisNone:
schema= {
'fields': [
 {'name': header, 'type': 'string'}
forheaderinheaders
 ]
 }
_resource['schema'] =schema

close()
delstream

returnitertools\
 .islice(
_reader(
get_opener(_url, _resource, columns),
_url,
max_row=limit_rows),
1, None)


parameters, datapackage, resource_iterator=ingest()

resources=ResourceMatcher(parameters.get('resources'), datapackage)
ignore_missing=parameters.get('ignore-missing', False)
limit_rows=parameters.get('limit-rows', -1)

new_resource_iterator= []
forresourceindatapackage['resources']:

ifstreamable(resource):
url=resource[PROP_STREAMED_FROM]

name=resource['name']
ifnotresources.match(name):
continue

path=get_path(resource)
ifpathisNoneorpath==PATH_PLACEHOLDER:
path=os.path.join('data', name+'.csv')
resource['path'] =path

resource[PROP_STREAMING] =True

rows=stream_reader(resource, url, ignore_missingorurl=="", limit_rows,
resource.pop('http_headers', None))

new_resource_iterator.append(rows)

elifstreaming(resource):
new_resource_iterator.append(next(resource_iterator))

spew(datapackage, new_resource_iterator)