def _get_external_data(url):
result = {}
try:
# urlopen might fail if it runs into redirections,
# because of Python issue #13696. Fixed in locators
# using a custom redirect handler.
resp = urlopen(url)
headers = resp.info()
ct = headers.get('Content-Type')
if not ct.startswith('application/json'):
logger.debug('Unexpected response for JSON request: %s', ct)
else:
reader = codecs.getreader('utf-8')(resp)
#data = reader.read().decode('utf-8')
#result = json.loads(data)
result = json.load(reader)
except Exception as e:
logger.exception('Failed to get external data for %s: %s', url, e)
return result
python类reader()的实例源码
def reader(self, stream, context):
"""
Read lines from a subprocess' output stream and either pass to a progress
callable (if specified) or write progress information to sys.stderr.
"""
progress = self.progress
verbose = self.verbose
while True:
s = stream.readline()
if not s:
break
if progress is not None:
progress(s, context)
else:
if not verbose:
sys.stderr.write('.')
else:
sys.stderr.write(s.decode('utf-8'))
sys.stderr.flush()
stream.close()
def _get_external_data(url):
result = {}
try:
# urlopen might fail if it runs into redirections,
# because of Python issue #13696. Fixed in locators
# using a custom redirect handler.
resp = urlopen(url)
headers = resp.info()
ct = headers.get('Content-Type')
if not ct.startswith('application/json'):
logger.debug('Unexpected response for JSON request: %s', ct)
else:
reader = codecs.getreader('utf-8')(resp)
#data = reader.read().decode('utf-8')
#result = json.loads(data)
result = json.load(reader)
except Exception as e:
logger.exception('Failed to get external data for %s: %s', url, e)
return result
def reader(self, stream, context):
"""
Read lines from a subprocess' output stream and either pass to a progress
callable (if specified) or write progress information to sys.stderr.
"""
progress = self.progress
verbose = self.verbose
while True:
s = stream.readline()
if not s:
break
if progress is not None:
progress(s, context)
else:
if not verbose:
sys.stderr.write('.')
else:
sys.stderr.write(s.decode('utf-8'))
sys.stderr.flush()
stream.close()
def _get_external_data(url):
result = {}
try:
# urlopen might fail if it runs into redirections,
# because of Python issue #13696. Fixed in locators
# using a custom redirect handler.
resp = urlopen(url)
headers = resp.info()
ct = headers.get('Content-Type')
if not ct.startswith('application/json'):
logger.debug('Unexpected response for JSON request: %s', ct)
else:
reader = codecs.getreader('utf-8')(resp)
#data = reader.read().decode('utf-8')
#result = json.loads(data)
result = json.load(reader)
except Exception as e:
logger.exception('Failed to get external data for %s: %s', url, e)
return result
def reader(self, stream, context):
"""
Read lines from a subprocess' output stream and either pass to a progress
callable (if specified) or write progress information to sys.stderr.
"""
progress = self.progress
verbose = self.verbose
while True:
s = stream.readline()
if not s:
break
if progress is not None:
progress(s, context)
else:
if not verbose:
sys.stderr.write('.')
else:
sys.stderr.write(s.decode('utf-8'))
sys.stderr.flush()
stream.close()
def uninstallation_paths(dist):
"""
Yield all the uninstallation paths for dist based on RECORD-without-.pyc
Yield paths to all the files in RECORD. For each .py file in RECORD, add
the .pyc in the same directory.
UninstallPathSet.add() takes care of the __pycache__ .pyc.
"""
from pip.utils import FakeFile # circular import
r = csv.reader(FakeFile(dist.get_metadata_lines('RECORD')))
for row in r:
path = os.path.join(dist.location, row[0])
yield path
if path.endswith('.py'):
dn, fn = os.path.split(path)
base = fn[:-3]
path = os.path.join(dn, base + '.pyc')
yield path
def _get_external_data(url):
result = {}
try:
# urlopen might fail if it runs into redirections,
# because of Python issue #13696. Fixed in locators
# using a custom redirect handler.
resp = urlopen(url)
headers = resp.info()
ct = headers.get('Content-Type')
if not ct.startswith('application/json'):
logger.debug('Unexpected response for JSON request: %s', ct)
else:
reader = codecs.getreader('utf-8')(resp)
#data = reader.read().decode('utf-8')
#result = json.loads(data)
result = json.load(reader)
except Exception as e:
logger.exception('Failed to get external data for %s: %s', url, e)
return result
def uninstallation_paths(dist):
"""
Yield all the uninstallation paths for dist based on RECORD-without-.pyc
Yield paths to all the files in RECORD. For each .py file in RECORD, add
the .pyc in the same directory.
UninstallPathSet.add() takes care of the __pycache__ .pyc.
"""
from pip.utils import FakeFile # circular import
r = csv.reader(FakeFile(dist.get_metadata_lines('RECORD')))
for row in r:
path = os.path.join(dist.location, row[0])
yield path
if path.endswith('.py'):
dn, fn = os.path.split(path)
base = fn[:-3]
path = os.path.join(dn, base + '.pyc')
yield path
def _get_external_data(url):
result = {}
try:
# urlopen might fail if it runs into redirections,
# because of Python issue #13696. Fixed in locators
# using a custom redirect handler.
resp = urlopen(url)
headers = resp.info()
ct = headers.get('Content-Type')
if not ct.startswith('application/json'):
logger.debug('Unexpected response for JSON request: %s', ct)
else:
reader = codecs.getreader('utf-8')(resp)
#data = reader.read().decode('utf-8')
#result = json.loads(data)
result = json.load(reader)
except Exception as e:
logger.exception('Failed to get external data for %s: %s', url, e)
return result
def reader(self, stream, context):
"""
Read lines from a subprocess' output stream and either pass to a progress
callable (if specified) or write progress information to sys.stderr.
"""
progress = self.progress
verbose = self.verbose
while True:
s = stream.readline()
if not s:
break
if progress is not None:
progress(s, context)
else:
if not verbose:
sys.stderr.write('.')
else:
sys.stderr.write(s.decode('utf-8'))
sys.stderr.flush()
stream.close()
def uninstallation_paths(dist):
"""
Yield all the uninstallation paths for dist based on RECORD-without-.pyc
Yield paths to all the files in RECORD. For each .py file in RECORD, add
the .pyc in the same directory.
UninstallPathSet.add() takes care of the __pycache__ .pyc.
"""
from pip.utils import FakeFile # circular import
r = csv.reader(FakeFile(dist.get_metadata_lines('RECORD')))
for row in r:
path = os.path.join(dist.location, row[0])
yield path
if path.endswith('.py'):
dn, fn = os.path.split(path)
base = fn[:-3]
path = os.path.join(dn, base + '.pyc')
yield path
def _insert_zone_names(self):
"""
Args:
_zonetree (str): set in __init__
Returns:
List of dicts (str: str) devices by zone
"""
self._zone_name = None
self._zonetree_io = StringIO(self._zonetree)
self._zonetree_csv = csv.reader(self._zonetree_io, delimiter=',')
self._zonetree_lod = []
for self._row in self._zonetree_csv:
if self._row[0] == '1':
self._zone_name = self._row[1]
if self._zone_name == 'Undefined':
self._zone_name = ''
continue
for self._dev in self._devtree:
if self._dev['ds_id'] == self._row[2]:
self._dev['zone_name'] = self._zone_name
return self._devtree
cisco_meraki_fw_ap_rules_api.py 文件源码
项目：cisco-meraki-fw-ap-rules-api
作者: robertcsapo
项目源码
文件源码
阅读 34
收藏 0
点赞 0
评论 0
def meraki_ap_fw_rules_import_csv(filename):
fw_ap_import_csv_list = []
try:
with open(filename) as csvfile:
fw_ap_import_csv = csv.reader(csvfile, delimiter=',')
key=0
for item in fw_ap_import_csv:
if key != 0:
print(item)
fw_ap_import_csv_list.append(item)
if len(item) < 5:
key +=1
print("")
sys.exit("- Error: Row %s isn't correct, exiting... -" % key)
key +=1
return fw_ap_import_csv_list
except FileNotFoundError:
sys.exit("- Error: CSV file not found %s, exiting... -" % filename)
def _get_external_data(url):
result = {}
try:
# urlopen might fail if it runs into redirections,
# because of Python issue #13696. Fixed in locators
# using a custom redirect handler.
resp = urlopen(url)
headers = resp.info()
ct = headers.get('Content-Type')
if not ct.startswith('application/json'):
logger.debug('Unexpected response for JSON request: %s', ct)
else:
reader = codecs.getreader('utf-8')(resp)
#data = reader.read().decode('utf-8')
#result = json.loads(data)
result = json.load(reader)
except Exception as e:
logger.exception('Failed to get external data for %s: %s', url, e)
return result
def reader(self, stream, context):
"""
Read lines from a subprocess' output stream and either pass to a progress
callable (if specified) or write progress information to sys.stderr.
"""
progress = self.progress
verbose = self.verbose
while True:
s = stream.readline()
if not s:
break
if progress is not None:
progress(s, context)
else:
if not verbose:
sys.stderr.write('.')
else:
sys.stderr.write(s.decode('utf-8'))
sys.stderr.flush()
stream.close()
def readAdults(file_name):
adult_file = open(file_name, 'r')
adults_csv = csv.reader(adult_file, delimiter=',')
# ignore the headers
next(adults_csv, None)
# create the dict we need
adults = {}
for adult_idx in adults_csv:
adult = adults[int(adult_idx[0])] = {}
adult['age'] = int(adult_idx[1])
adult['workclass'] = adult_idx[2]
adult['native-country'] = adult_idx[3]
adult['sex'] = adult_idx[4]
adult['race'] = adult_idx[5]
adult['marital-status'] = adult_idx[6]
adult_file.close()
return adults
def test_serial():
assert s.serialize_value(None) == 'x'
assert s.serialize_value(True) == 'true'
assert s.serialize_value(False) == 'false'
assert s.serialize_value(5) == 'i:5'
assert s.serialize_value(5.0) == 'f:5.0'
assert s.serialize_value(decimal.Decimal('5.5')) == 'n:5.5'
assert s.serialize_value('abc') == 's:abc'
assert s.serialize_value(b'abc') == 'b:YWJj'
assert s.serialize_value(b'abc') == 'b:YWJj'
assert s.serialize_value(datetime.date(2007, 12, 5)) == 'd:2007-12-05'
assert s.serialize_value(datetime.datetime(2007, 12, 5, 12, 30, 30, tzinfo=utc)) \
== 'dt:2007-12-05 12:30:30+00:00'
assert s.serialize_value(datetime.time(12, 34, 56)) == 't:12:34:56'
with raises(NotImplementedError):
s.serialize_value(csv.reader)
def get_countries_from_csv(filepath):
""" Process a csv countaining country data.
Args:
filepath (str): A path to a csv file.
Returns:
list: sanitized country names as strings.
Raises:
OSError: If file does not exist at the filepath.
"""
if not path.exists(filepath):
raise OSError('Path to file: "{}" does not exist.'.format(filepath))
with open(filepath, 'r') as csv_file:
reader = csv.reader(csv_file)
return [sanitize_string(''.join(row)) for row in reader]
def build_list_of_people_from_csv(filename, countries):
people = []
with open(filename, 'r') as opened_file:
reader = csv.reader(opened_file)
for row in reader:
excluded_countries = []
if len(row) > 1:
excluded_countries = [
sanitize_string(name)
for name in row[1].split(',')
]
people.append(
Person(
name=row[0],
countries=countries,
excluded_countries=excluded_countries
)
)
return people
