Python aiohttp.web 模块，json_response() 实例源码
我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用aiohttp.web.json_response()。
def register(*, name, email, sha1_pw, oid=None, image=None):
check_string(name=name)
check_email_and_password(email, sha1_pw)
users = await User.findAll('email = ?', [email])
if users:
raise APIValueError('email', 'Email is already in used.')
user = User(name=name.strip(), email=email, password=sha1_pw, image=image or '/static/img/user.png')
await user.save()
if oid:
o = Oauth(id=oid, user_id=user.id)
await o.save()
# register ok,signin
return user.signin(web.json_response({'signin user': user.name}))
# ????
def post(self) -> None:
request_data = await self.request.json()
args = require_dict(request_data['args'], str, None)
if request_data.get('use_monitor_def_name', False):
monitor_def = get_monitor_def_by_name(
self.request.app['active_monitor_manager'],
require_str(request_data['monitor_def']))
else:
monitor_def = self.request.app['active_monitor_manager'].monitor_defs.get(
require_int(request_data['monitor_def']))
if not monitor_def:
raise errors.InvalidData('Monitor def not found')
monitor = await create_active_monitor(self.request.app['active_monitor_manager'], args, monitor_def)
if not monitor:
raise errors.InvalidData('invalid monitor arguments')
return web.json_response(monitor.id)
def get(self) -> web.Response:
dbcon = self.request.app['dbcon']
if 'id' in self.request.rel_url.query:
contact_id = require_int(get_request_param(self.request, 'id'))
c = await contact.get_contact(dbcon, contact_id)
contact_list = [] # type: Iterable[object_models.Contact]
if c:
contact_list = [c]
Metadata_list = await Metadata.get_Metadata_for_object(dbcon, 'contact', contact_id)
elif 'Meta_key' in self.request.rel_url.query:
Meta_key = require_str(get_request_param(self.request, 'Meta_key'))
Meta_value = require_str(get_request_param(self.request, 'Meta_value'))
contact_list = await contact.get_contacts_for_Metadata(dbcon, Meta_key, Meta_value)
Metadata_list = await Metadata.get_Metadata_for_object_Metadata(
dbcon, Meta_value, 'contacts')
else:
contact_list = await contact.get_all_contacts(dbcon)
Metadata_list = await Metadata.get_Metadata_for_object_type(dbcon, 'contact')
return web.json_response(apply_Metadata_to_model_list(contact_list, Metadata_list))
def get(self) -> web.Response:
dbcon = self.request.app['dbcon']
if 'id' in self.request.rel_url.query:
contact_group_id = require_int(get_request_param(self.request, 'id'))
contact_group_item = await contact.get_contact_group(dbcon, contact_group_id)
contact_group_list = [] # type: Iterable[object_models.ContactGroup]
if contact_group_item:
contact_group_list = [contact_group_item]
Metadata_list = await Metadata.get_Metadata_for_object(dbcon, 'contact_group', contact_group_id)
elif 'Meta_key' in self.request.rel_url.query:
Meta_key = require_str(get_request_param(self.request, 'Meta_value'))
contact_group_list = await contact.get_contact_groups_for_Metadata(dbcon, 'contact_groups')
else:
contact_group_list = await contact.get_all_contact_groups(dbcon)
Metadata_list = await Metadata.get_Metadata_for_object_type(dbcon, 'monitor_group')
return web.json_response(apply_Metadata_to_model_list(contact_group_list, Metadata_list))
def get(self) -> web.Response:
dbcon = self.request.app['dbcon']
if 'id' in self.request.rel_url.query:
monitor_group_id = require_int(get_request_param(self.request, 'id'))
monitor_group_item = await monitor_group.get_monitor_group(dbcon, monitor_group_id)
monitor_group_list = [] # type: Iterable[object_models.MonitorGroup]
if monitor_group_item:
monitor_group_list = [monitor_group_item]
Metadata_list = await Metadata.get_Metadata_for_object(dbcon, 'monitor_group', monitor_group_id)
elif 'Meta_key' in self.request.rel_url.query:
Meta_key = require_str(get_request_param(self.request, 'Meta_value'))
monitor_group_list = await monitor_group.get_monitor_groups_for_Metadata(dbcon, 'monitor_groups')
else:
monitor_group_list = await monitor_group.get_all_monitor_groups(dbcon)
Metadata_list = await Metadata.get_Metadata_for_object_type(dbcon, 'monitor_group')
return web.json_response(apply_Metadata_to_model_list(monitor_group_list, Metadata_list))
def search(self, request):
text = request.GET.get("text")
offset = int(request.GET.get("offset", 0))
limit = int(request.GET.get("limit", 10))
cursor = text_search(text) if text else db.tracks.find({})
total = await cursor.count()
results = await cursor.skip(offset).limit(limit).to_list(limit)
for r in results:
del r["_id"]
return web.json_response({
"tracks": results,
"offset": offset,
"limit": limit,
"total": total
})
def auth_through_token(app: web.Application, handler):
async def middleware_handler(request: web.Request):
headers = request.headers
x_auth_token = headers.get("X-Auth-Token")
project_id = request.match_info.get('project_id')
c = config.Config.config_instance()
try:
auth = identity.Token(c.auth_url,
token=x_auth_token,
project_id=project_id)
sess = session.Session(auth=auth)
ks = client.Client(session=sess,
project_id=project_id)
ks.authenticate(token=x_auth_token)
except Exception as ex:
return web.json_response(status=401, data={
"error": {
"message": ("Not authorized. Reason: {}"
.format(str(ex)))
}
})
return await handler(request)
return middleware_handler
def hello(request):
data = {
"config" : {
"polling" : {
"sleep" : "12:00:00"
}
},
"_links" : {
"deploymentBase" : {
"href" : "https://rollouts-cs.apps.bosch-iot-cloud.com/TENANT_ID/controller/v1/CONTROLLER_ID/deploymentBase/3?c=-2129030598"
},
"configData" : {
"href" : "https://rollouts-cs.apps.bosch-iot-cloud.com/TENANT_ID/controller/v1/CONTROLLER_ID/configData"
}
}
}
return web.json_response(data)
def heartbeat(request):
info = await asyncio.gather(archives.heartbeat(),
balrog.heartbeat(),
bedrock.heartbeat(),
bouncer.heartbeat(),
buildhub.heartbeat(),
crash_stats.heartbeat(),
product_details.heartbeat(),
telemetry.heartbeat())
status = all(info) and 200 or 503
return web.json_response({"archive": info[0],
"balrog": info[1],
"bedrock": info[2],
"bouncer": info[3],
"buildhub": info[4],
"crash-stats": info[5],
"product-details": info[6],
"telemetry": info[7]},
status=status)
def validate_product_version(func):
async def decorate(request):
product = request.match_info['product']
version = request.match_info.get('version')
if product not in PRODUCTS:
return web.json_response({
'status': 404,
'message': 'Invalid product: {} not in {}'.format(product, PRODUCTS)
}, status=404)
if version and not is_valid_version(version):
return web.json_response({
'status': 404,
'message': 'Invalid version number: {}'.format(version)
}, status=404)
if version:
return await func(request, product, version)
return await func(request, product)
return decorate
def status_response(task):
@validate_product_version
async def wrapped(request, version):
try:
response = await task(product, version)
except Exception as e: # In case something went bad,we return an error status message
logger.exception(e)
body = {
'status': 'error',
'message': str(e)
}
if hasattr(e, 'url') and e.url is not None:
body['link'] = e.url
return web.json_response(body)
return web.json_response(response)
return wrapped
def handle(self, request):
body = await request.json()
req_id = body['id']
method = body['method']
hash_or_number, _ = body['params']
if method == 'eth_getBlockByNumber':
if hash_or_number == "latest":
head = self.chain.get_canonical_head()
number = head.block_number
else:
number = int(hash_or_number, 16)
block = await self.chain.get_canonical_block_by_number(number)
elif method == 'eth_getBlockByHash':
block_hash = decode_hex(hash_or_number)
block = await self.chain.get_block_by_hash(block_hash)
else:
raise HTTPMethodNotAllowed(method, self.allowed_methods)
block_dict = self._block_to_dict(block)
response = {"jsonrpc": "2.0", "id": req_id, "result": block_dict}
return web.json_response(response)
def rest_error(message:str="UnkNow", code:str="0", error_id:str=""):
"""
Build the REST error response
:param message: The short "friendly user" error message
:param code: The code of the error type
:param error_id: The id of the error,to return to the end-user.
This code will allow admins to find in logs where exactly this error occure
"""
results = {
"success": False,
"msg": message,
"error_code": code,
"error_url": code,
"error_id": error_id
}
return web.json_response(results)
def test_set_get(loop, test_client):
app = web.Application()
app.router.add_get(
'/test/1',
lambda x: web.json_response(["Python"]))
client = await test_client(app)
url = client.make_url('/')
data = 'Python'
config = MergeDict(
storage=MergeDict(
cls='aioworkers.storage.http.Storage',
prefix=str(url),
semaphore=1,
format='json',
),
)
async with Context(config=config, loop=loop) as context:
storage = context.storage
assert data in await storage.get('test/1')
with pytest.raises(StorageError):
await storage.set('test/1', data)
def _make_response(self, request, response):
if callable(response):
if asyncio.iscoroutinefunction(response):
return await response(request)
else:
return response(request)
elif isinstance(response, dict):
return web.json_response(response)
rqdata = await request.json()
if 'method' in rqdata and rqdata['method'] == 'session-get':
return web.json_response(SESSION_GET_RESPONSE)
elif response is None:
raise RuntimeError('Set the response property before making a request!')
else:
return web.Response(text=response)
def test_timeout_plus_one(self):
delay = self.client.timeout+1
# NOTE: This function is only called sometimes,probably depending on
# which task finishes first,advance() or client.connect().
async def delay_response(request):
await asyncio.sleep(delay, loop=self.loop)
return web.json_response(rsrc.SESSION_GET_RESPONSE)
self.daemon.response = delay_response
with self.assertRaises(ConnectionError) as cm:
await asyncio.gather(self.advance(delay),
self.client.connect(),
loop=self.loop)
self.assertIn('timeout', str(cm.exception).lower())
self.assertIn(str(self.client.timeout), str(cm.exception))
self.assertIn(str(self.url), str(cm.exception))
self.assert_cb_connected_called(calls=0)
self.assert_cb_disconnected_called(calls=0)
self.assert_cb_error_called(calls=1,
args=[(self.url,)],
kwargs=[{'error': r'{}.*{}'.format(self.client.timeout,
self.url)}])
def login(self, request):
try:
assert self.config["auth"]["user"]
assert self.config["auth"]["password"]
except AssertionError:
logger.error("HTTP Auth credentials are missing!")
return web.json_response({"error": "Auth credentials are missing."}, status=400)
params = await request.post()
user = params.get('username', None)
if (user == self.config["auth"]["user"] and
params.get('password', None) == self.config["auth"]["password"]):
# User is in our database,remember their login details
_tokens[user] = str(uuid.uuid4())
return web.json_response({"token": _tokens[user]})
return web.json_response({"error": "Unauthorized"}, status=401)
def bookshelf_new(request):
"""
Add the given book to the shelf event lake.
"""
payload = await request.content.read()
# ensure the book gets an id
book = json.loads(payload.decode('utf-8'))
book["id"] = str(uuid.uuid4())
# create an event from this request
event = make_event(name="book-added",
payload=json.dumps(book),
safe=False, idempotent=False)
# let's push it
await send_event(b"bookshelf", event)
return web.json_response(status=201, data=book)
def upload_pet_image(request):
with open(os.path.join(os.path.dirname(__file__), 'sample.jpg'), 'rb') as f:
data = await request.post()
file_data = data.get('file')
content = file_data.file.read()
expected_content = f.read()
if content != expected_content:
return web.HTTPBadRequest()
if not (
request.match_info['petId'] == '42'
and data.get('userId') == '12'
):
return web.HTTPBadRequest()
return web.json_response({})
def hook(request):
body = await request.read()
check_signature = hmac.compare_digest(
get_signature(body),
request.headers.get('X-Hub-Signature', '')
)
if not check_signature:
return web.HTTPBadRequest()
body = await request.json()
headers = dict(request.headers.items())
del headers['X-Hub-Signature']
ref = get_hook_ctx(headers, body, clean=True)
if ref:
request.app.loop.create_task(ci(ref))
return web.json_response(ref)
def dps_baglan(request):
link=""
global tox
durum=""
data = yield from request.post()
toxid = data['kdugum']
print (toxid)
port =33999
lport=random.randrange(38000,40000)
komut="./tuntox -i "+str(toxid)+" -L "+str(lport)+":127.0.0.1:"+str(port)
print ("dugumler aras? tunel ac?l?yor.")
#tunel id kaydetmek için-?u an iptal
#open("yenidugum","w").write(toxid)
durum=yield from komutar(komut)
link=lokalhost+":"+str(lport)
return web.json_response(data=link)
def _post_(self, request):
data = await request.post()
domain = data.get('domain', "")
domain_expiration = DomainExpiration(domain=domain)
if not domain_expiration.name:
response = {"response": "KO", "message": self.EMPTY_DOMAIN}
elif not domain_expiration.allowed:
response = {"response": "KO", "message": self.ALLOWED}
elif self.db.domains.find_one({"name": domain_expiration.name}):
response = {"response": "KO", "message": self.EXISTS}
else:
domain_expiration.save()
response = {"response": "OK"}
return web.json_response(response)
def handle_slack_button(request):
payload = multidict_to_dict(await request.post())
body = json.loads(payload['payload'])
print(f'INFO: Button request body:\n{body}.')
try:
verify_request_token(body)
except (KeyError, TokenError) as err:
print(f'INFO: Invalid token: {err}')
return web.json_response({'text': 'Unverified message.'})
if body['actions'][0]['name'] == 'Cancel':
if 'text' not in body:
body['text'] = ''
if 'user_id' not in body:
body['user_id'] = body['user']['id']
iwant_object = IwantRequest(body, (), _slack_user_pattern)
iwant_object.cancel_iwant_task()
return web.json_response({'text': 'Request was cancelled.'})
def handler(request):
try:
print(f'Requested: {request.path_qs}')
_, symbol, price, volume = request.path_qs.split('/')
except ValueError:
return web.Response(status=500)
t0 = time.time() # for profiling
SHARED_MEMORY[symbol].append(price)
strength = STRATEGY_FUNC(SHARED_MEMORY, volume)
elapsed_time = time.time() - t0 # for profiling
# Send a signal
print(f'Analyzed {symbol} => {strength} ({elapsed_time:.4f})')
result = {
'symbol': symbol,
'price': price,
'strategy': ARGS['strategy'],
'strength': strength,
'stoploss': int(price) - 3000,
'elapsedtime': elapsed_time
}
return web.json_response(result)
def geass(context, tmpl=None, *, app_key=APP_KEY, encoding='utf-8', status=200):
# print('path->',request.url)
if tmpl is None:
return web.json_response(context)
try:
if 'identifier' in context:
context['PAGE_IDENTIFIER'] = request.app.router[context['identifier']].url()
# Auth
context['drawer_category'] = CONST.CATEGORY
except:
raise RuntimeError
response = render_template(tmpl, context,
app_key=app_key, encoding=encoding)
response.set_status(status)
return response
def webhook_emit(self, request):
name = request.match_info.get('name', None)
token = request.match_info.get('token', None)
post_data = await request.post()
if not name:
if 'name' not in request.GET or 'name' not in post_data:
return web.json_response({'success': False, 'message': 'Error: required parameter "name" not found.'})
if not token and not any(['token' not in request.GET, 'X-Hub-Signature' not in request.headers]):
return web.json_response({'success': False, 'message': 'Error: required parameter "token" not found.'})
if not name:
name = request.GET.get('name', None) or post_data['name']
if not token:
token = request.GET.get('token', None)
if token != TOKEN is False:
return web.json_response({'success': False, 'message': 'Error: authentication Failed. Invalid token.'})
data = {k: v for k, v in request.GET.items() if k not in ['name', 'token']}
if post_data:
data.update({k: v for k, v in post_data.items() if k not in ['name', 'token']})
self.emit(name, data=data)
return web.json_response({'success': True})
def get_statistics(request):
"""
Args:
request:
Returns:
a general set of statistics
:Example:
curl -X GET http://localhost:8081/foglamp/statistics
"""
payload = PayloadBuilder().SELECT(("key", "description", "value")).ORDER_BY(["key"]).payload()
storage_client = connect.get_storage()
results = storage_client.query_tbl_with_payload('statistics', payload)
return web.json_response(results['rows'])
def get_categories(request):
"""
Args:
request:
Returns:
the list of kNown categories in the configuration database
:Example:
curl -X GET http://localhost:8081/foglamp/categories
"""
# Todo: make it optimized and elegant
cf_mgr = ConfigurationManager(connect.get_storage())
categories = await cf_mgr.get_all_category_names()
categories_json = [{"key": c[0], "description": c[1]} for c in categories]
return web.json_response({'categories': categories_json})
def get_category(request):
"""
Args:
request: category_name is required
Returns:
the configuration items in the given category.
:Example:
curl -X GET http://localhost:8081/category/PURGE_READ
"""
category_name = request.match_info.get('category_name', None)
if not category_name:
raise web.HTTPBadRequest(reason="Category Name is required")
# Todo: make it optimized and elegant
cf_mgr = ConfigurationManager(connect.get_storage())
category = await cf_mgr.get_category_all_items(category_name)
if category is None:
raise web.HTTPNotFound(reason="No such Category Found for {}".format(category_name))
return web.json_response(category)
def get_category_item(request):
"""
Args:
request: category_name & config_item are required
Returns:
the configuration item in the given category.
:Example:
curl -X GET http://localhost:8081/foglamp/category/PURGE_READ/age
"""
category_name = request.match_info.get('category_name', None)
config_item = request.match_info.get('config_item', None)
if not category_name or not config_item:
raise web.HTTPBadRequest(reason="Both Category Name and Config items are required")
# Todo: make it optimized and elegant
cf_mgr = ConfigurationManager(connect.get_storage())
category_item = await cf_mgr.get_category_item(category_name, config_item)
if category_item is None:
raise web.HTTPNotFound(reason="No Category Item Found")
return web.json_response(category_item)
def get_backups(request):
"""
Returns a list of all backups
:Example: curl -X GET http://localhost:8082/foglamp/backup
:Example: curl -X GET http://localhost:8082/foglamp/backup?limit=2&skip=1&status=complete
"""
try:
limit = int(request.query['limit']) if 'limit' in request.query else None
skip = int(request.query['skip']) if 'skip' in request.query else None
status = request.query['status'] if 'status' in request.query else None
# Todo : Fix after actual implementation
Backup.get_backup_list.return_value = [{'id': 28, 'date': '2017-08-30 04:05:10.382', 'status': 'running'},
{'id': 27, 'date': '2017-08-29 04:05:13.392', 'status': 'Failed'},
{'id': 26, 'date': '2017-08-28 04:05:08.201', 'status': 'complete'}]
# backup_json = [{"id": b[0],"date": b[1],"status": b[2]}
# for b in Backup.get_backup_list(limit=limit,skip=skip,status=status)]
backup_json = Backup.get_backup_list(limit=limit, skip=skip, status=status)
except Backup.DoesNotExist:
raise web.HTTPNotFound(reason='No backups found for queried parameters')
return web.json_response({"backups": backup_json})
def get_backup_details(request):
"""
Returns the details of a backup
:Example: curl -X GET http://localhost:8082/foglamp/backup/1
"""
backup_id = request.match_info.get('backup_id', None)
if not backup_id:
raise web.HTTPBadRequest(reason='Backup id is required')
else:
try:
backup_id = int(backup_id)
except ValueError:
raise web.HTTPBadRequest(reason='Invalid backup id')
try:
# Todo : Fix after actual implementation
Backup.get_backup_details.return_value = \
{"date": '2017-08-30 04:05:10.382', "status": "running"}
except Backup.DoesNotExist:
raise web.HTTPNotFound(reason='Backup with {} does not exist'.format(backup_id))
_resp = Backup.get_backup_details(id=backup_id)
_resp["id"] = backup_id
return web.json_response(_resp)
def delete_backup(request):
"""
Delete a backup
:Example: curl -X DELETE http://localhost:8082/foglamp/backup/1
"""
backup_id = request.match_info.get('backup_id', None)
if not backup_id:
raise web.HTTPBadRequest(reason='Backup id is required')
else:
try:
backup_id = int(backup_id)
except ValueError:
raise web.HTTPBadRequest(reason='Invalid backup id')
try:
# Todo : Fix after actual implementation
Backup.delete_backup.return_value = "Backup deleted successfully"
except Backup.DoesNotExist:
raise web.HTTPNotFound(reason='Backup with {} does not exist'.format(backup_id))
_resp = Backup.delete_backup(id=backup_id)
return web.json_response({'message': _resp})
def restore_backup(request):
"""
Restore from a backup
:Example: curl -X PUT http://localhost:8082/foglamp/backup/1/restore
"""
backup_id = request.match_info.get('backup_id', None)
if not backup_id:
raise web.HTTPBadRequest(reason='Backup id is required')
else:
try:
backup_id = int(backup_id)
except ValueError:
raise web.HTTPBadRequest(reason='Invalid backup id')
try:
# Todo : Fix after actual implementation
Backup.restore_backup.return_value = 1
except Backup.DoesNotExist:
raise web.HTTPNotFound(reason='Backup with {} does not exist'.format(backup_id))
try:
Backup.restore_backup(id=backup_id)
return web.json_response({'message': 'Restore backup with id {} started successfully'.format(backup_id)})
except Backup.RestoreFailed as ex:
return web.json_response({'error': 'Restore backup with id {} Failed,reason {}'.format(backup_id, ex)})
def get_scheduled_process(request):
"""
Returns a list of all the defined scheduled_processes from scheduled_processes table
"""
scheduled_process_name = request.match_info.get('scheduled_process_name', None)
if not scheduled_process_name:
raise web.HTTPBadRequest(reason='No Scheduled Process Name given')
payload = PayloadBuilder().SELECT(("name")).WHERE(["name", "=", scheduled_process_name]).payload()
_storage = connect.get_storage()
scheduled_process = _storage.query_tbl_with_payload('scheduled_processes', payload)
if len(scheduled_process['rows']) == 0:
raise web.HTTPNotFound(reason='No such Scheduled Process: {}.'.format(scheduled_process_name))
return web.json_response(scheduled_process['rows'][0].get("name"))
#################################
# Schedules
#################################
def get_schedules(request):
"""
Returns a list of all the defined schedules from schedules table
"""
schedule_list = await server.Server.scheduler.get_schedules()
schedules = []
for sch in schedule_list:
schedules.append({
'id': str(sch.schedule_id),
'name': sch.name,
'process_name': sch.process_name,
'type': Schedule.Type(int(sch.schedule_type)).name,
'repeat': sch.repeat.total_seconds() if sch.repeat else 0,
'time': (sch.time.hour * 60 * 60 + sch.time.minute * 60 + sch.time.second) if sch.time else 0 ,
'day': sch.day,
'exclusive': sch.exclusive
})
return web.json_response({'schedules': schedules})
def delete_schedule(request):
"""
Delete a schedule from schedules table
:Example: curl -X DELETE http://localhost:8082/foglamp/schedule/dc9bfc01-066a-4cc0-b068-9c35486db87f
"""
try:
schedule_id = request.match_info.get('schedule_id', None)
if not schedule_id:
raise web.HTTPBadRequest(reason='Schedule ID is required.')
try:
assert uuid.UUID(schedule_id)
except ValueError as ex:
raise web.HTTPNotFound(reason="Invalid Schedule ID {}".format(schedule_id))
await server.Server.scheduler.delete_schedule(uuid.UUID(schedule_id))
return web.json_response({'message': 'Schedule deleted successfully', 'id': schedule_id})
except (ValueError, ScheduleNotFoundError) as ex:
raise web.HTTPNotFound(reason=str(ex))
def get_schedule_type(request):
"""
Args:
request:
Returns:
an array of Schedule type enumeration key index values
:Example: curl -X GET http://localhost:8082/foglamp/schedule/type
"""
results = []
for _type in Schedule.Type:
data = {'index': _type.value, 'name': _type.name}
results.append(data)
return web.json_response({'schedule_type': results})
#################################
# Tasks
#################################
def get_task_state(request):
"""
Args:
request:
Returns:
an array of Task State enumeration key index values
:Example: curl -X GET http://localhost:8082/foglamp/task/state
"""
results = []
for _state in Task.State:
data = {'index': _state.value, 'name': _state.name}
results.append(data)
return web.json_response({'task_state': results})
def asset_counts(request):
""" browse all the assets for which we have recorded readings and
return a readings count.
Return the result of the query
SELECT asset_code,count(*) FROM readings GROUP BY asset_code;
"""
# Todo: FOGL-643 - Aggregate with alias support needed to use payload builder
# PayloadBuilder().AGGREGATE(["count","*"]).GROUP_BY('asset_code')
aggregate = {"operation": "count", "column": "*", "alias": "count"}
d = OrderedDict()
d['aggregate'] = aggregate
d['group'] = "asset_code"
payload = json.dumps(d)
_storage = connect.get_storage()
results = _storage.query_tbl_with_payload('readings', payload)
return web.json_response(results['rows'])
def get_audit_log_codes(request):
"""
Args:
request:
Returns:
an array of log codes with description
:Example:
curl -X GET http://localhost:8081/foglamp/audit/logcode
"""
storage_client = connect.get_storage()
result = storage_client.query_tbl('log_codes')
return web.json_response({'log_code': result['rows']})
def get_audit_log_severity(request):
"""
Args:
request:
Returns:
an array of audit severity enumeration key index values
:Example:
curl -X GET http://localhost:8081/foglamp/audit/severity
"""
results = []
for _severity in Severity:
data = {'index': _severity.value, 'name': _severity.name}
results.append(data)
return web.json_response({"log_severity": results})
def unregister(request):
""" Deregister a service
:Example: curl -X DELETE http://localhost:8082/foglamp/service/dc9bfc01-066a-4cc0-b068-9c35486db87f
"""
try:
service_id = request.match_info.get('service_id', None)
if not service_id:
raise web.HTTPBadRequest(reason='Service id is required')
try:
Service.Instances.get(idx=service_id)
except Service.DoesNotExist:
raise web.HTTPBadRequest(reason='Service with {} does not exist'.format(service_id))
Service.Instances.unregister(service_id)
_resp = {'id': str(service_id), 'message': 'Service unregistered'}
return web.json_response(_resp)
except ValueError as ex:
raise web.HTTPNotFound(reason=str(ex))
def _process_request(self, method, default_status=200):
if not method:
raise web.HTTPMethodNotAllowed
response = {
'code': 0,
'data': {}
}
try:
await method.from_http(self.request)
await method.call_pre_hooks()
output = await method.get_output()
await method.call_post_hooks(response=output)
response['data']['result'] = output.get('result')
response['data']['pagination'] = output.get('pagination')
except BaseApiException as exc:
response['code'] = exc.status_code
self.set_errors(response, method.errors, exc.message)
except Exception as exc:
response['code'] = 500
self.set_errors(response, str(exc))
self._response = response
return web.json_response(response, status=default_status)
def authenticate(*, sha1_pw):
check_email_and_password(email, [email])
if len(users) == 0:
raise APIValueError('email', 'Email not exist.')
user = users[0]
# check password
if not user.verify_password(sha1_pw):
raise APIValueError('password', 'Invalid password')
# authenticate ok,signin
return user.signin(web.json_response({'signin user': user.name}))
def ajax_page(self, status, page):
context = await self.get_page_context(page)
record_list = aiohttp_jinja2.render_string(
'admin/partials/_record_list.html', self.request, context)
return web.json_response({'status': status,
'record_list': record_list})
def ajax_empty(self, status):
return web.json_response(dict(status=status))
def simple_webhook(request):
logger.info("Processing simple webhook event.")
url = request.GET['head']
priority = ('10-webhook', url)
yield from WORKERS.enqueue(
ProcessUrlTask(priority, url, callable_=process_url)
)
return web.json_response({'message': 'Event processing in progress.'})
def github_webhook(request):
logger.info("Processing GitHub webhook event.")
payload = yield from request.read()
yield from request.release()
try:
validate_signature(request.headers, payload)
except DenySignature:
return web.json_response({'message': 'Invalid signature.'}, status=403)
payload = json.loads(payload.decode('utf-8'))
if 'hook_id' in payload:
logger.debug("Ping from GitHub.")
return web.json_response({'message': 'Hookaïda !'}, status=200)
try:
url = infer_url_from_event(payload)
except SkipEvent:
return web.json_response({'message': 'Event processed.'})
priority = ('10-webhook', url)
logger.info("Queuing %s.", callable_=process_url)
)
return web.json_response({'message': 'Event processing in progress.'})
def get(self) -> web.Response:
dbcon = self.request.app['dbcon']
monitor_ids = await self._get_monitor_ids(dbcon)
Metadata_dict = await self._get_monitor_Metadata(dbcon)
monitors = []
for monitor_id in monitor_ids:
monitor = self.request.app['active_monitor_manager'].monitors.get(monitor_id, None)
if not monitor:
continue
data = self._collect_monitor_data(monitor, Metadata_dict)
monitors.append(data)
return web.json_response(monitors)
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 [email protected] 举报，一经查实，本站将立刻删除。