Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
path_with_added_args(
request,
{"_shape": "objects"},
path=request.path.rsplit(".jsono", 1)[0] + ".json",
),
forward_querystring=False,
)
if _format in self.ds.renderers.keys():
# Dispatch request to the correct output format renderer
# (CSV is not handled here due to streaming)
result = self.ds.renderers[_format](request.args, data, self.name)
if result is None:
raise NotFound("No data")
r = Response(
body=result.get("body"),
status=result.get("status_code", 200),
content_type=result.get("content_type", "text/plain"),
)
else:
extras = {}
if callable(extra_template_data):
extras = extra_template_data()
if asyncio.iscoroutine(extras):
extras = await extras
else:
extras = extra_template_data
url_labels_extra = {}
if data.get("expandable_columns"):
url_labels_extra = {"_labels": "on"}
**context,
**{
"app_css_hash": self.ds.app_css_hash(),
"select_templates": select_templates,
"zip": zip,
"body_scripts": body_scripts,
"extra_css_urls": self._asset_urls("extra_css_urls", template, context),
"extra_js_urls": self._asset_urls("extra_js_urls", template, context),
"format_bytes": format_bytes,
"database_url": self.database_url,
"database_color": self.database_color,
},
**extra_template_vars,
}
if request.args.get("_context") and self.ds.config("template_debug"):
return Response.html(
"<pre>{}</pre>".format(
escape(json.dumps(template_context, default=repr, indent=4))
)
)
return Response.html(await template.render_async(template_context))
async def get(self, request, as_format):
data = self.data_callback()
if as_format:
headers = {}
if self.ds.cors:
headers["Access-Control-Allow-Origin"] = "*"
return Response(
json.dumps(data),
content_type="application/json; charset=utf-8",
headers=headers,
)
else:
return await self.render(
["show_json.html"],
request=request,
context={
"filename": self.filename,
"data_json": json.dumps(data, indent=4),
},
"body_scripts": body_scripts,
"extra_css_urls": self._asset_urls("extra_css_urls", template, context),
"extra_js_urls": self._asset_urls("extra_js_urls", template, context),
"format_bytes": format_bytes,
"database_url": self.database_url,
"database_color": self.database_color,
},
**extra_template_vars,
}
if request.args.get("_context") and self.ds.config("template_debug"):
return Response.html(
"<pre>{}</pre>".format(
escape(json.dumps(template_context, default=repr, indent=4))
)
)
return Response.html(await template.render_async(template_context))
if _format == "csv":
return await self.as_csv(request, database, hash, **kwargs)
if _format is None:
# HTML views default to expanding all foreign key labels
kwargs["default_labels"] = True
extra_template_data = {}
start = time.time()
status_code = 200
templates = []
try:
response_or_template_contexts = await self.data(
request, database, hash, **kwargs
)
if isinstance(response_or_template_contexts, Response):
return response_or_template_contexts
else:
data, extra_template_data, templates = response_or_template_contexts
except QueryInterrupted:
raise DatasetteError(
"""
SQL query took too long. The time limit is controlled by the
<a href="https://datasette.readthedocs.io/en/stable/config.html#sql-time-limit-ms">sql_time_limit_ms</a>
configuration option.
""",
title="SQL Interrupted",
status=400,
messagge_is_html=True,
)
except (sqlite3.OperationalError, InvalidSql) as e:
"tables_count": len(visible_tables),
"table_rows_sum": sum((t["count"] or 0) for t in visible_tables),
"show_table_row_counts": bool(table_counts),
"hidden_table_rows_sum": sum(
t["count"] for t in hidden_tables if t["count"] is not None
),
"hidden_tables_count": len(hidden_tables),
"views_count": len(views),
}
)
if as_format:
headers = {}
if self.ds.cors:
headers["Access-Control-Allow-Origin"] = "*"
return Response(
json.dumps({db["name"]: db for db in databases}, cls=CustomJSONEncoder),
content_type="application/json; charset=utf-8",
headers=headers,
)
else:
return await self.render(
["index.html"],
request=request,
context={
"databases": databases,
"metadata": self.ds.metadata(),
"datasette_version": __version__,
},
def redirect(self, request, path, forward_querystring=True, remove_args=None):
if request.query_string and "?" not in path and forward_querystring:
path = "{}?{}".format(path, request.query_string)
if remove_args:
path = path_with_removed_args(request, remove_args, path=path)
r = Response.redirect(path)
r.headers["Link"] = "<{}>; rel=preload".format(path)
if self.ds.cors:
r.headers["Access-Control-Allow-Origin"] = "*"
return r
def options(self, request, *args, **kwargs):
r = Response.text("ok")
if self.ds.cors:
r.headers["Access-Control-Allow-Origin"] = "*"
return r