本文整理汇总了Python中redash.handlers.base.get_object_or_404函数的典型用法代码示例。如果您正苦于以下问题:Python get_object_or_404函数的具体用法?Python get_object_or_404怎么用?Python get_object_or_404使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了get_object_or_404函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: post
def post(self, data_source_id):
req = request.get_json(True)
if req:
if not req.has_key("type"):
abort(400)
if req["type"] == "column":
data_source = get_object_or_404(models.DataSourceColumn.get_by_id, data_source_id)
elif req["type"] == "table":
data_source = get_object_or_404(models.DataSourceTable.get_by_id, data_source_id)
else:
abort(400)
if req["type"] == "column" and req.has_key("joins"):
data_source.joins = req["joins"]
if req.has_key("description"):
data_source.description = req["description"]
if req.has_key("tags"):
data_source.tags = req["tags"]
data_source.save()
return data_source.to_dict(all=True)
else:
abort(400)
开发者ID:hudl,项目名称:redash,代码行数:27,代码来源:data_sources.py
示例2: get
def get(self, query_id=None, query_result_id=None, filetype='json'):
# TODO:
# This method handles two cases: retrieving result by id & retrieving result by query id.
# They need to be split, as they have different logic (for example, retrieving by query id
# should check for query parameters and shouldn't cache the result).
should_cache = query_result_id is not None
if query_result_id is None and query_id is not None:
query = get_object_or_404(models.Query.get_by_id_and_org, query_id, self.current_org)
if query:
query_result_id = query._data['latest_query_data']
if query_result_id:
query_result = get_object_or_404(models.QueryResult.get_by_id_and_org, query_result_id, self.current_org)
else:
query_result = None
if query_result:
require_access(query_result.data_source.groups, self.current_user, view_only)
if isinstance(self.current_user, models.ApiUser):
event = {
'user_id': None,
'org_id': self.current_org.id,
'action': 'api_get',
'timestamp': int(time.time()),
'api_key': self.current_user.name,
'file_type': filetype,
'user_agent': request.user_agent.string,
'ip': request.remote_addr
}
if query_id:
event['object_type'] = 'query'
event['object_id'] = query_id
else:
event['object_type'] = 'query_result'
event['object_id'] = query_result_id
record_event.delay(event)
if filetype == 'json':
response = self.make_json_response(query_result)
elif filetype == 'xlsx':
response = self.make_excel_response(query_result)
else:
response = self.make_csv_response(query_result)
if len(settings.ACCESS_CONTROL_ALLOW_ORIGIN) > 0:
self.add_cors_headers(response.headers)
if should_cache:
response.headers.add_header('Cache-Control', 'max-age=%d' % ONE_YEAR)
return response
else:
abort(404, message='No cached result found for this query.')
开发者ID:5t111111,项目名称:redash,代码行数:57,代码来源:query_results.py
示例3: get
def get(self, query_id, dropdown_query_id):
query = get_object_or_404(models.Query.get_by_id_and_org, query_id, self.current_org)
require_access(query, current_user, view_only)
related_queries_ids = [p['queryId'] for p in query.parameters if p['type'] == 'query']
if int(dropdown_query_id) not in related_queries_ids:
dropdown_query = get_object_or_404(models.Query.get_by_id_and_org, dropdown_query_id, self.current_org)
require_access(dropdown_query.data_source, current_user, view_only)
return dropdown_values(dropdown_query_id)
开发者ID:getredash,项目名称:redash,代码行数:10,代码来源:query_results.py
示例4: get
def get(self, data_source_id):
data_source = get_object_or_404(models.DataSource.get_by_id_and_org, data_source_id, self.current_org)
require_access(data_source.groups, self.current_user, view_only)
refresh = request.args.get('refresh') is not None
schema = data_source.get_schema(refresh)
return schema
开发者ID:kitsuyui,项目名称:redash,代码行数:7,代码来源:data_sources.py
示例5: post
def post(self, query_id):
query = get_object_or_404(models.Query.get_by_id_and_org, query_id, self.current_org)
query_def = request.get_json(force=True)
require_object_modify_permission(query, self.current_user)
for field in ['id', 'created_at', 'api_key', 'visualizations', 'latest_query_data', 'user', 'last_modified_by', 'org']:
query_def.pop(field, None)
if 'latest_query_data_id' in query_def:
query_def['latest_query_data'] = query_def.pop('latest_query_data_id')
if 'data_source_id' in query_def:
query_def['data_source'] = query_def.pop('data_source_id')
query_def['last_modified_by'] = self.current_user
query_def['changed_by'] = self.current_user
try:
query.update_instance(**query_def)
except models.ConflictDetectedError:
abort(409)
# old_query = copy.deepcopy(query.to_dict())
# new_change = query.update_instance_tracked(changing_user=self.current_user, old_object=old_query, **query_def)
# abort(409) # HTTP 'Conflict' status code
result = query.to_dict(with_visualizations=True)
return result
开发者ID:AntoineAugusti,项目名称:redash,代码行数:29,代码来源:queries.py
示例6: get
def get(self, query_id):
q = get_object_or_404(models.Query.get_by_id_and_org, query_id, self.current_org)
require_access(q.groups, self.current_user, view_only)
result = q.to_dict(with_visualizations=True)
result['can_edit'] = can_modify(q, self.current_user)
return result
开发者ID:AntoineAugusti,项目名称:redash,代码行数:7,代码来源:queries.py
示例7: post
def post(self, query_id):
"""
Execute a saved query.
:param number query_id: The ID of the query whose results should be fetched.
:param object parameters: The parameter values to apply to the query.
:qparam number max_age: If query results less than `max_age` seconds old are available,
return them, otherwise execute the query; if omitted or -1, returns
any cached result, or executes if not available. Set to zero to
always execute.
"""
params = request.get_json(force=True)
parameter_values = params.get('parameters')
max_age = params.get('max_age', -1)
# max_age might have the value of None, in which case calling int(None) will fail
if max_age is None:
max_age = -1
max_age = int(max_age)
query = get_object_or_404(models.Query.get_by_id_and_org, query_id, self.current_org)
allow_executing_with_view_only_permissions = query.parameterized.is_safe
if has_access(query, self.current_user, allow_executing_with_view_only_permissions):
return run_query(query.parameterized, parameter_values, query.data_source, query_id, max_age)
else:
return {'job': {'status': 4, 'error': 'You do not have permission to run queries with this data source.'}}, 403
开发者ID:ariarijp,项目名称:redash,代码行数:28,代码来源:query_results.py
示例8: public_dashboard
def public_dashboard(token, org_slug=None):
# TODO: verify object is a dashboard?
if not isinstance(current_user, models.ApiUser):
api_key = get_object_or_404(models.ApiKey.get_by_api_key, token)
dashboard = api_key.object
else:
dashboard = current_user.object
user = {
'permissions': [],
'apiKey': current_user.id
}
headers = {
'Cache-Control': 'no-cache, no-store, max-age=0, must-revalidate'
}
record_event(current_org, current_user, {
'action': 'view',
'object_id': dashboard.id,
'object_type': 'dashboard',
'public': True,
'headless': 'embed' in request.args,
'referer': request.headers.get('Referer')
})
response = render_template("public.html",
headless='embed' in request.args,
user=json.dumps(user),
seed_data=json_dumps({
'dashboard': serializers.public_dashboard(dashboard)
}),
client_config=json.dumps(settings.COMMON_CLIENT_CONFIG))
return response, 200, headers
开发者ID:AntoineAugusti,项目名称:redash,代码行数:35,代码来源:embed.py
示例9: get
def get(self, object_type, object_id, access_type):
model = get_model_from_type(object_type)
obj = get_object_or_404(model.get_by_id_and_org, object_id, self.current_org)
has_access = AccessPermission.exists(obj, access_type, self.current_user)
return {'response': has_access}
开发者ID:AntoineAugusti,项目名称:redash,代码行数:7,代码来源:permissions.py
示例10: post
def post(self, object_type, object_id):
model = get_model_from_type(object_type)
obj = get_object_or_404(model.get_by_id_and_org, object_id, self.current_org)
require_admin_or_owner(obj.user_id)
req = request.get_json(True)
access_type = req['access_type']
if access_type not in ACCESS_TYPES:
abort(400, message='Unknown access type.')
try:
grantee = User.get_by_id_and_org(req['user_id'], self.current_org)
except User.DoesNotExist:
abort(400, message='User not found.')
permission = AccessPermission.grant(obj, access_type, grantee, self.current_user)
self.record_event({
'action': 'grant_permission',
'object_id': object_id,
'object_type': object_type,
'access_type': access_type,
'grantee': grantee.id
})
return permission.to_dict()
开发者ID:AntoineAugusti,项目名称:redash,代码行数:29,代码来源:permissions.py
示例11: post
def post(self, query_id):
query = get_object_or_404(models.Query.get_by_id_and_org, query_id, self.current_org)
require_access(query.groups, self.current_user, not_view_only)
parameter_values = collect_parameters_from_request(request.args)
return run_query(query.data_source, parameter_values, query.query, query.id)
开发者ID:5t111111,项目名称:redash,代码行数:7,代码来源:queries.py
示例12: get
def get(self, query_id=None, query_result_id=None, filetype='json'):
should_cache = query_result_id is not None
if query_result_id is None and query_id is not None:
query = get_object_or_404(models.Query.get_by_id_and_org, query_id, self.current_org)
if query:
query_result_id = query._data['latest_query_data']
if query_result_id:
query_result = get_object_or_404(models.QueryResult.get_by_id_and_org, query_result_id, self.current_org)
if query_result:
if isinstance(self.current_user, models.ApiUser):
event = {
'user_id': None,
'org_id': self.current_org.id,
'action': 'api_get',
'timestamp': int(time.time()),
'api_key': self.current_user.id,
'file_type': filetype
}
if query_id:
event['object_type'] = 'query'
event['object_id'] = query_id
else:
event['object_type'] = 'query_result'
event['object_id'] = query_result_id
record_event.delay(event)
if filetype == 'json':
response = self.make_json_response(query_result)
elif filetype == 'xlsx':
response = self.make_excel_response(query_result)
else:
response = self.make_csv_response(query_result)
if len(settings.ACCESS_CONTROL_ALLOW_ORIGIN) > 0:
self.add_cors_headers(response.headers)
if should_cache:
response.headers.add_header('Cache-Control', 'max-age=%d' % ONE_YEAR)
return response
else:
abort(404)
开发者ID:MoveOnOrg,项目名称:redash,代码行数:47,代码来源:query_results.py
示例13: get
def get(self, query_id, dropdown_query_id):
query = get_object_or_404(models.Query.get_by_id_and_org, query_id, self.current_org)
related_queries_ids = [p['queryId'] for p in query.parameters if p['type'] == 'query']
if int(dropdown_query_id) not in related_queries_ids:
abort(403)
return dropdown_values(dropdown_query_id, should_require_access=False)
开发者ID:ariarijp,项目名称:redash,代码行数:8,代码来源:query_results.py
示例14: get
def get(self, query_id):
q = get_object_or_404(models.Query.get_by_id_and_org, query_id, self.current_org)
require_access(q.groups, self.current_user, view_only)
if q:
return q.to_dict(with_visualizations=True)
else:
abort(404, message="Query not found.")
开发者ID:5t111111,项目名称:redash,代码行数:8,代码来源:queries.py
示例15: post
def post(self, data_source_id):
data_source = get_object_or_404(models.DataSource.get_by_id_and_org, data_source_id, self.current_org)
try:
data_source.query_runner.test_connection()
except Exception as e:
return {"message": unicode(e), "ok": False}
else:
return {"message": "success", "ok": True}
开发者ID:Captricity,项目名称:redash,代码行数:9,代码来源:data_sources.py
示例16: delete
def delete(self, object_id):
dashboard = get_object_or_404(models.Dashboard.get_by_slug_and_org, object_id, self.current_org)
models.Favorite.query.filter(models.Favorite.object==dashboard, models.Favorite.user==self.current_user).delete()
models.db.session.commit()
self.record_event({
'action': 'unfavorite',
'object_id': dashboard.id,
'object_type': 'dashboard'
})
开发者ID:jcox92,项目名称:redash,代码行数:9,代码来源:favorites.py
示例17: get
def get(self, group_id):
group = get_object_or_404(models.Group.get_by_id_and_org, group_id, self.current_org)
# TOOD: move to models
data_sources = models.DataSource.select(models.DataSource, models.DataSourceGroup.view_only)\
.join(models.DataSourceGroup)\
.where(models.DataSourceGroup.group == group)
return [ds.to_dict(with_permissions=True) for ds in data_sources]
开发者ID:olivetree123,项目名称:redash-x,代码行数:9,代码来源:groups.py
示例18: delete
def delete(self, query_id):
"""
Archives a query.
:param query_id: ID of query to archive
"""
query = get_object_or_404(models.Query.get_by_id_and_org, query_id, self.current_org)
require_admin_or_owner(query.user_id)
query.archive(self.current_user)
models.db.session.commit()
开发者ID:moritz9,项目名称:redash,代码行数:10,代码来源:queries.py
示例19: delete
def delete(self, snippet_id):
snippet = get_object_or_404(models.QuerySnippet.get_by_id_and_org, snippet_id, self.current_org)
require_admin_or_owner(snippet.user.id)
snippet.delete_instance()
self.record_event({
'action': 'delete',
'object_id': snippet.id,
'object_type': 'query_snippet'
})
开发者ID:AntoineAugusti,项目名称:redash,代码行数:10,代码来源:query_snippets.py
示例20: get
def get(self, dashboard_slug=None):
dashboard = get_object_or_404(models.Dashboard.get_by_slug_and_org, dashboard_slug, self.current_org)
response = dashboard.to_dict(with_widgets=True, user=self.current_user)
api_key = models.ApiKey.get_by_object(dashboard)
if api_key:
response['public_url'] = url_for('redash.public_dashboard', token=api_key.api_key, org_slug=self.current_org.slug, _external=True)
response['api_key'] = api_key.api_key
return response
开发者ID:5t111111,项目名称:redash,代码行数:10,代码来源:dashboards.py
注:本文中的redash.handlers.base.get_object_or_404函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论