Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_path_with_added_args(path, added_args, expected):
request = Request.from_path_with_query_string(path)
actual = utils.path_with_added_args(request, added_args)
assert expected == actual
except (sqlite3.OperationalError) as e:
raise DatasetteError(str(e))
except DatasetteError:
raise
end = time.time()
data['query_ms'] = (end - start) * 1000
for key in ('source', 'source_url', 'license', 'license_url'):
value = self.ds.metadata.get(key)
if value:
data[key] = value
if as_json:
# Special case for .jsono extension - redirect to _shape=objects
if as_json == '.jsono':
return self.redirect(
request,
path_with_added_args(
request,
{'_shape': 'objects'},
path=request.path.rsplit('.jsono', 1)[0] + '.json'
),
forward_querystring=False
)
# Deal with the _shape option
shape = request.args.get('_shape', 'lists')
if shape in ('objects', 'object'):
columns = data.get('columns')
rows = data.get('rows')
if rows and columns:
data['rows'] = [
dict(zip(columns, row))
for row in rows
]
except DatasetteError:
raise
end = time.time()
data["query_ms"] = (end - start) * 1000
for key in ("source", "source_url", "license", "license_url"):
value = self.ds.metadata(key)
if value:
data[key] = value
# Special case for .jsono extension - redirect to _shape=objects
if _format == "jsono":
return self.redirect(
request,
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"),
special_args = {}
special_args_lists = {}
other_args = {}
for key, value in args.items():
if key.startswith('_') and '__' not in key:
special_args[key] = value[0]
special_args_lists[key] = value
else:
other_args[key] = value[0]
# Handle ?_filter_column and redirect, if present
redirect_params = filters_should_redirect(special_args)
if redirect_params:
return self.redirect(
request,
path_with_added_args(request, redirect_params),
forward_querystring=False
)
# Spot ?_sort_by_desc and redirect to _sort_desc=(_sort)
if '_sort_by_desc' in special_args:
return self.redirect(
request,
path_with_added_args(request, {
'_sort_desc': special_args.get('_sort'),
'_sort_by_desc': None,
'_sort': None,
}),
forward_querystring=False
)
units = self.table_metadata(name, table).get('units', {})
prefix = str(rows[-2][sort or sort_desc])
next_value = '{},{}'.format(
urllib.parse.quote_plus(prefix), next_value
)
added_args = {
'_next': next_value,
}
if sort:
added_args['_sort'] = sort
else:
added_args['_sort_desc'] = sort_desc
else:
added_args = {
'_next': next_value,
}
next_url = urllib.parse.urljoin(request.url, path_with_added_args(
request, added_args
))
rows = rows[:self.page_size]
# Number of filtered rows in whole set:
filtered_table_rows_count = None
if count_sql:
try:
count_rows = list(await self.execute(name, count_sql, params))
filtered_table_rows_count = count_rows[0][0]
except sqlite3.OperationalError:
# Almost certainly hit the timeout
pass
# human_description_en combines filters AND search, if provided
human_description_en = filters.human_description_en(extra=search_description)
other_args[key] = value[0]
# Handle ?_filter_column and redirect, if present
redirect_params = filters_should_redirect(special_args)
if redirect_params:
return self.redirect(
request,
path_with_added_args(request, redirect_params),
forward_querystring=False
)
# Spot ?_sort_by_desc and redirect to _sort_desc=(_sort)
if '_sort_by_desc' in special_args:
return self.redirect(
request,
path_with_added_args(request, {
'_sort_desc': special_args.get('_sort'),
'_sort_by_desc': None,
'_sort': None,
}),
forward_querystring=False
)
units = self.table_metadata(name, table).get('units', {})
filters = Filters(sorted(other_args.items()), units, ureg)
where_clauses, params = filters.build_where_clauses()
# _search support:
fts_table = None
fts_sql = detect_fts_sql(table)
fts_rows = list(await self.execute(name, fts_sql))
other_args.append((key, v))
# Handle ?_filter_column and redirect, if present
redirect_params = filters_should_redirect(special_args)
if redirect_params:
return self.redirect(
request,
path_with_added_args(request, redirect_params),
forward_querystring=False,
)
# Spot ?_sort_by_desc and redirect to _sort_desc=(_sort)
if "_sort_by_desc" in special_args:
return self.redirect(
request,
path_with_added_args(
request,
{
"_sort_desc": special_args.get("_sort"),
"_sort_by_desc": None,
"_sort": None,
},
),
forward_querystring=False,
)
table_metadata = self.ds.table_metadata(database, table)
units = table_metadata.get("units", {})
filters = Filters(sorted(other_args), units, ureg)
where_clauses, params = filters.build_where_clauses(table)
extra_wheres_for_ui = []
# it can still be queried using ?_col__exact=blah
special_args = {}
other_args = []
for key, value in args.items():
if key.startswith("_") and "__" not in key:
special_args[key] = value[0]
else:
for v in value:
other_args.append((key, v))
# Handle ?_filter_column and redirect, if present
redirect_params = filters_should_redirect(special_args)
if redirect_params:
return self.redirect(
request,
path_with_added_args(request, redirect_params),
forward_querystring=False,
)
# Spot ?_sort_by_desc and redirect to _sort_desc=(_sort)
if "_sort_by_desc" in special_args:
return self.redirect(
request,
path_with_added_args(
request,
{
"_sort_desc": special_args.get("_sort"),
"_sort_by_desc": None,
"_sort": None,
},
),
forward_querystring=False,
'<a href="{url}">{url}</a>'.format(
url=jinja2.escape(value.strip())
)
)
display_row.append(display_value)
display_rows.append(display_row)
return {
"display_rows": display_rows,
"custom_sql": True,
"named_parameter_values": named_parameter_values,
"editable": editable,
"canned_query": canned_query,
"metadata": metadata,
"config": self.ds.config_dict(),
"request": request,
"path_with_added_args": path_with_added_args,
"path_with_removed_args": path_with_removed_args,
"hide_sql": "_hide_sql" in params,
}