def read_json_file(filename, path):
"""Accepts a file name and loads it as a json object.
Args:
filename (str): Filename to be loaded.
path (str): Directory path to use.
Returns:
obj: json object
"""
result = []
try:
with open(path + filename + ".json", "r") as entry:
result = ujson.load(entry)
except IOError as ex:
print "I/O error({0}): {1}".format(ex.errno, ex.strerror)
else:
entry.close()
return result
python类load()的实例源码
def test_should_fail_missing_timestamp_in_body(self, bulk_processor):
events_resource = _init_resource(self)
events_resource._processor = bulk_processor
unit_test_patch = os.path.dirname(__file__)
json_file_path = 'event_template_json/req_simple_event.json'
patch_to_req_simple_event_file = os.path.join(unit_test_patch,
json_file_path)
with open(patch_to_req_simple_event_file, 'r') as fi:
events = json.load(fi)['events']
body = {'events': [events]}
self.simulate_request(
path=ENDPOINT,
method='POST',
headers={
'Content-Type': 'application/json',
'X_ROLES': 'monasca'
},
body=json.dumps(body)
)
self.assertEqual(falcon.HTTP_422, self.srmock.status)
def parse_postman_file(path: str,
environment_vars: List[Dict]) -> APITest:
"""
This function parse a Postman file and return an APITest object instance
:param path: path to postman file
:type path: str
:return: APITest instance
:rtype: APITest
"""
assert isinstance(path, str)
with open(path, "r") as f:
json_info = load(f)
return postman_parser(json_info,
environment_vars)
def postman_parser_form_file(file_path: str,
environment_vars: dict = None):
"""
Get a parser collection, in JSON input format, and parser it
:param file_path: file path
:type file_path: str
:return: a Postman object
:rtype: APITest
:raise ApitestInvalidFormatError: When the Postman JSON file has wrong format
"""
assert file_path is not None
with open(file_path, "r") as f:
try:
loaded_data = load(f)
except (ValueError, TypeError) as e:
raise ApitestInvalidFormatError from e
return postman_parser(loaded_data,
environment_vars)
def get_database_connection():
global logger
s3 = boto3.resource('s3')
metasrcs = ujson.load(
s3.Object('net-mozaws-prod-us-west-2-pipeline-metadata',
'sources.json').get()['Body'])
creds = ujson.load(
s3.Object(
'net-mozaws-prod-us-west-2-pipeline-metadata',
'%s/write/credentials.json'
% metasrcs[DB]['metadata_prefix']
).get()['Body'])
conn = psycopg2.connect(connection_factory=LoggingConnection,
host=creds['host'], port=creds.get('port', 5432),
user=creds['username'], password=creds['password'],
dbname=creds['db_name'])
conn.initialize(logger)
return conn
def setUp(self):
self.test_subject = IIIFValidator()
self.test_subject.logger.setLevel("CRITICAL")
self.test_subject.fail_fast = False
self.test_subject.collect_errors = True
self.test_subject.collect_warnings = True
self.test_subject.debug = True
self.base_dir = os.path.dirname(os.path.realpath(__file__))
with open(os.path.join(self.base_dir, 'fixtures/valid_manifest')) as f:
self.valid_manifest = json.load(f)
with open(os.path.join(self.base_dir, 'fixtures/error_collection.json')) as f:
self.error_collection = json.load(f)
self.man_with_warnings_and_errors = self.error_collection['manifests'][-1]['manifest']
def read_vqa(vqa_dir, section):
ans = 'mscoco_%s2014_annotations.json' % section
with (vqa_dir / ans).open() as file_:
ans_data = json.load(file_)
answers_by_id = {}
for answer in ans_data['annotations']:
mca = answer['multiple_choice_answer']
answers_by_id[answer['question_id']] = mca
filename = ('MultipleChoice_mscoco_'
'%s2014_questions.json' % section)
with (vqa_dir / filename).open() as file_:
ques_data = json.load(file_)
for question in ques_data['questions']:
text = question['question']
ques_id = question['question_id']
options = question['multiple_choices']
yield text, options, answers_by_id[ques_id]
def convert_image(options):
print 'loading cache'
options['cache'] = json.load(open('color_cache.json'))
print 'beginning conversion'
im = Image.open( options['filename'] )
o = convert_frame(im, options)
o += ANSI_RESET + '\n\n'
# Save to ANSI file and add SAUCE record
if options['output_file'] is not sys.stdout:
save_frame(o,options)
add_sauce(o,options)
# Output to console (unicode)
else:
print_frame(o,options)
# json.dump(options['cache'], open('color_cache.json','w'))
# MAIN FUNCTION
def _set_asset_paths(self, app):
"""
Read in the manifest json file which acts as a manifest for assets.
This allows us to get the asset path as well as hashed names.
:param app: Sanic application
:return: None
"""
webpack_stats = app.config['WEBPACK_MANIFEST_PATH']
try:
with open(webpack_stats) as stats_json:
stats = json.load(stats_json)
if app.config['WEBPACK_ASSETS_URL']:
self.assets_url = app.config['WEBPACK_ASSETS_URL']
else:
self.assets_url = stats['publicPath']
self.assets = stats['assets']
except IOError:
raise RuntimeError(
"Sanic-Webpack requires 'WEBPACK_MANIFEST_PATH' to be set and "
"it must point to a valid json file.")
def query(self, *args, onlyCached=False, **kwargs):
queryString, hashString = self._queryString(*args, **kwargs)
filename = self.__cacheDir + '/' + self._prefix + '-' + self.__hash(hashString)
if not os.path.exists(self.__cacheDir):
os.makedirs(self.__cacheDir)
if os.path.exists(filename):
with open(filename, 'r') as file:
data = ujson.load(file)
elif onlyCached:
print('[' + self._prefix + '] data not cached: ' + queryString)
return None
else:
print('[' + self._prefix + '] downloading data: ' + queryString)
if self._waitForReady() == None:
if self.__lastQuery and self.__waitBetweenQueries and time.time() - self.__lastQuery < self.__waitBetweenQueries:
time.sleep(self.__waitBetweenQueries - time.time() + self.__lastQuery)
self.__lastQuery = time.time()
data = self.__query(queryString)
with open(filename, 'w') as file:
ujson.dump(data, file)
result = self._rawToResult(data, queryString)
if not self._isValid(result):
raise(Exception('[' + self._prefix + '] error in result (' + filename + '): ' + queryString))
return result
def _generate_fsm(language):
cond_path = os.path.join(CONFIG_PATH, language)
cond_path = os.path.join(cond_path, 'cond_config.json')
with open(cond_path) as cond_file:
cond_config = ujson.load(cond_file)
cls_name = language + '_FSM'
fsm_cls = generate_fsm_cls(cls_name, cond_config)
fsm_config_path = os.path.join(CONFIG_PATH, 'FSM.json')
with open(fsm_config_path, 'r') as fsm_config_file:
data = ujson.loads(jsmin(fsm_config_file.read()))
states = data['states']
transitions = data['transitions']
return fsm_cls(
states=states,
transitions=transitions,
bot_client=line_bot_api,
template_path=os.path.join(BOT_TEMPLATE_PATH, language),
)
def testCase000(self):
"""Load persistent data from files into into memory.
"""
global jval
global datafile
# data
datafile = os.path.abspath(os.path.dirname(__file__))+os.sep+str('testdata.json')
if not os.path.isfile(datafile):
raise BaseException("Missing JSON data:file="+str(datafile))
# load data
with open(datafile) as data_file:
jval = myjson.load(data_file)
if jval == None:
raise BaseException("Failed to load data:"+str(data_file))
jval = jval
assert jval
pass
def testCase000(self):
"""Load persistent data from files into into memory.
"""
global jval
global datafile
# data
datafile = os.path.abspath(os.path.dirname(__file__))+os.sep+str('testdata.json')
if not os.path.isfile(datafile):
raise BaseException("Missing JSON data:file="+str(datafile))
# load data
with open(datafile) as data_file:
jval = myjson.load(data_file)
if jval == None:
raise BaseException("Failed to load data:"+str(data_file))
jval = jval
assert jval
pass
def testCase001(self):
"""Load persistent schema from files into into memory.
"""
global sval
global datafile
global schemafile
# schema
schemafile = os.path.abspath(os.path.dirname(__file__))+os.sep+str('schema.jsd')
if not os.path.isfile(schemafile):
raise BaseException("Missing JSONschema:file="+str(schemafile))
with open(schemafile) as schema_file:
sval = myjson.load(schema_file)
if sval == None:
raise BaseException("Failed to load schema:"+str(schema_file))
#
# Create by object
#
def testCase000(self):
"""Load persistent data from files into into memory.
"""
global jval
global datafile
# data
datafile = os.path.abspath(os.path.dirname(__file__))+os.sep+str('testdata.json')
if not os.path.isfile(datafile):
raise BaseException("Missing JSON data:file="+str(datafile))
# load data
with open(datafile) as data_file:
jval = myjson.load(data_file)
if jval == None:
raise BaseException("Failed to load data:"+str(data_file))
jval = jval
assert jval
pass
#
# Create by object
#
def testCase000(self):
"""Load persistent data from files into into memory.
"""
global jval
global datafile
# data
datafile = os.path.abspath(os.path.dirname(__file__))+os.sep+str('testdata.json')
if not os.path.isfile(datafile):
raise BaseException("Missing JSON data:file="+str(datafile))
# load data
with open(datafile) as data_file:
jval = myjson.load(data_file)
if jval == None:
raise BaseException("Failed to load data:"+str(data_file))
jval = jval
assert jval
pass
def testCase001(self):
"""Load persistent schema from files into into memory.
"""
global sval
global datafile
global schemafile
# schema
schemafile = os.path.abspath(os.path.dirname(__file__))+os.sep+str('schema.jsd')
if not os.path.isfile(schemafile):
raise BaseException("Missing JSONschema:file="+str(schemafile))
with open(schemafile) as schema_file:
sval = myjson.load(schema_file)
if sval == None:
raise BaseException("Failed to load schema:"+str(schema_file))
#
# Create by object
#
def testCase000(self):
"""Load persistent data from files into into memory.
"""
global jval
global datafile
# data
datafile = os.path.abspath(os.path.dirname(__file__))+os.sep+str('testdata.json')
if not os.path.isfile(datafile):
raise BaseException("Missing JSON data:file="+str(datafile))
# load data
with open(datafile) as data_file:
jval = myjson.load(data_file)
if jval == None:
raise BaseException("Failed to load data:"+str(data_file))
jval = jval
assert jval
pass
def testCase001(self):
"""Load persistent schema from files into into memory.
"""
global sval
global datafile
global schemafile
# schema
schemafile = os.path.abspath(os.path.dirname(__file__))+os.sep+str('schema.jsd')
if not os.path.isfile(schemafile):
raise BaseException("Missing JSONschema:file="+str(schemafile))
with open(schemafile) as schema_file:
sval = myjson.load(schema_file)
if sval == None:
raise BaseException("Failed to load schema:"+str(schema_file))
#
# Create by object
#
def testCase000(self):
"""Load a data file.
"""
global jval
global datafile
# data
datafile = os.path.abspath(os.path.dirname(__file__))+os.sep+str('testdata.json')
if not os.path.isfile(datafile):
raise BaseException("Missing JSON data:file="+str(datafile))
# load data
with open(datafile) as data_file:
jval = myjson.load(data_file)
if jval == None:
raise BaseException("Failed to load data:"+str(data_file))
jval = jval
assert jval
pass
def testCase000(self):
"""Load a data file.
"""
global jval
global datafile
# data
datafile = os.path.abspath(os.path.dirname(__file__))+os.sep+str('datafile.json')
if not os.path.isfile(datafile):
raise BaseException("Missing JSON data:file="+str(datafile))
# load data
with open(datafile) as data_file:
jval = myjson.load(data_file)
if jval == None:
raise BaseException("Failed to load data:"+str(data_file))
jval = jval
assert jval
pass
def testCase000(self):
"""Load a data file.
"""
global jval
global datafile
# data
datafile = os.path.abspath(os.path.dirname(__file__))+os.sep+str('testdata.json')
if not os.path.isfile(datafile):
raise BaseException("Missing JSON data:file="+str(datafile))
# load data
with open(datafile) as data_file:
jval = myjson.load(data_file)
if jval == None:
raise BaseException("Failed to load data:"+str(data_file))
jval = jval
assert jval
pass
def testCase000(self):
"""Load a data file.
"""
global jval
global datafile
# data
datafile = os.path.abspath(os.path.dirname(__file__))+os.sep+str('testdata.json')
if not os.path.isfile(datafile):
raise BaseException("Missing JSON data:file="+str(datafile))
# load data
with open(datafile) as data_file:
jval = myjson.load(data_file)
if jval == None:
raise BaseException("Failed to load data:"+str(data_file))
jval = jval
assert jval
pass
def testCase000(self):
"""Load a data file.
"""
global jval
global datafile
# data
datafile = os.path.abspath(os.path.dirname(__file__))+os.sep+str('testdata.json')
if not os.path.isfile(datafile):
raise BaseException("Missing JSON data:file="+str(datafile))
# load data
with open(datafile) as data_file:
jval = myjson.load(data_file)
if jval == None:
raise BaseException("Failed to load data:"+str(data_file))
jval = jval
assert jval
pass
def testCase000(self):
"""Load a data file.
"""
global jval
global datafile
# data
datafile = os.path.abspath(os.path.dirname(__file__))+os.sep+str('testdata.json')
if not os.path.isfile(datafile):
raise BaseException("Missing JSON data:file="+str(datafile))
# load data
with open(datafile) as data_file:
jval = myjson.load(data_file)
if jval == None:
raise BaseException("Failed to load data:"+str(data_file))
jval = jval
assert jval
pass
def testCase000(self):
"""Load a data file.
"""
global jval
global datafile
# data
datafile = os.path.abspath(os.path.dirname(__file__))+os.sep+str('testdata.json')
if not os.path.isfile(datafile):
raise BaseException("Missing JSON data:file="+str(datafile))
# load data
with open(datafile) as data_file:
jval = myjson.load(data_file)
if jval == None:
raise BaseException("Failed to load data:"+str(data_file))
jval = jval
assert jval
pass
def testCase000(self):
"""Load a data file.
"""
global jval
global datafile
# data
datafile = os.path.abspath(os.path.dirname(__file__))+os.sep+str('testdata.json')
if not os.path.isfile(datafile):
raise BaseException("Missing JSON data:file="+str(datafile))
# load data
with open(datafile) as data_file:
jval = myjson.load(data_file)
if jval == None:
raise BaseException("Failed to load data:"+str(data_file))
jval = jval
assert jval
pass
def testCase000(self):
"""Load a data file.
"""
global jval
global datafile
# data
datafile = os.path.abspath(os.path.dirname(__file__))+os.sep+str('testdata.json')
if not os.path.isfile(datafile):
raise BaseException("Missing JSON data:file="+str(datafile))
# load data
with open(datafile) as data_file:
jval = myjson.load(data_file)
if jval == None:
raise BaseException("Failed to load data:"+str(data_file))
jval = jval
assert jval
pass
def testCase000(self):
"""Load a data file.
"""
global jval
global datafile
# data
datafile = os.path.abspath(os.path.dirname(__file__))+os.sep+str('testdata.json')
if not os.path.isfile(datafile):
raise BaseException("Missing JSON data:file="+str(datafile))
# load data
with open(datafile) as data_file:
jval = myjson.load(data_file)
if jval == None:
raise BaseException("Failed to load data:"+str(data_file))
jval = jval
assert jval
pass
def testCase000(self):
"""Load a data file.
"""
global jval
global datafile
# data
datafile = os.path.abspath(os.path.dirname(__file__))+os.sep+str('testdata.json')
if not os.path.isfile(datafile):
raise BaseException("Missing JSON data:file="+str(datafile))
# load data
with open(datafile) as data_file:
jval = myjson.load(data_file)
if jval == None:
raise BaseException("Failed to load data:"+str(data_file))
jval = jval
assert jval
pass
