Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
srsName = query.get('srsName', DEFAULT_SRS).strip()
if srsName:
if ':' in srsName:
srsName = srsName.split(':')[-1]
srs = int(srsName)
lower_x, lower_y = self._parse_coordinate(envelope, 'lowerCorner')
upper_x, upper_y = self._parse_coordinate(envelope, 'upperCorner')
# bbox must have lower corner first
bbox = {'lower_x': lower_x,
'lower_y': lower_y,
'upper_x': upper_x,
'upper_y': upper_y}
log.info('WFS Query typeName: %s parsed: SRS %s BBOX %s',
typeName, srs, bbox.values() if bbox else None)
return (typeName, srs, bbox)
abort(400, 'No suitable WFS Query found')
as a tuple of two floats. Any error, it calls abort (exception).
'''
if not corner_name in ('lowerCorner', 'upperCorner'):
abort(500, 'Bad param for _parse_coordinate')
corner = None
for corner_xml in envelope.iter('{http://www.opengis.net/gml}%s' % corner_name):
corner = corner_xml.text
error_message_base = 'WFS Query/Filter/BBOX/Envelope/%s ' % corner_name
if not corner:
abort(400, error_message_base + 'element missing')
corner = corner.strip()
if not corner:
abort(400, error_message_base + 'element blank')
coordinates = corner.split(' ')
if len(coordinates) != 2:
abort(400, error_message_base + 'has %i coordinates but should be 2' \
% len(coordinates))
try:
coordinates = [float(coord) for coord in coordinates]
except ValueError:
abort(400, error_message_base + 'coordinates not floats: %s' \
% coordinates)
return coordinates
import ckan.logic as logic
import ckan.plugins as p
from ckan.common import _
import ckan.lib.base as base
import logging
from ckan.lib import helpers as h
import re
from ckanext.nhm.views.default import DefaultView
from ckanext.nhm.lib.dwc import dwc_terms
log = logging.getLogger(__name__)
abort = base.abort
get_action = logic.get_action
class DarwinCoreView(DefaultView):
"""
View for displaying DwC resources
"""
format = 'dwc'
grid_default_columns = [
'_id',
'gbifIssue',
'scientificName',
'scientificNameAuthorship',
context = {
'model': model,
'session': model.Session,
'user': c.user or c.author,
'auth_user_obj': c.userobj
}
try:
resource = logic.get_action('resource_show')(
context,
{
'id': resource_id
}
)
except logic.NotFound:
base.abort(404, _('Resource not found'))
except logic.NotAuthorized:
base.abort(401, _('Unauthorized to read resource {0}'.format(id)))
# This isn't a file upload, so either redirect to the source
# (if available) or error out.
if resource.get('url_type') != 'upload':
url = resource.get('url')
if not url:
base.abort(404, _('No download is available'))
h.redirect_to(url)
if filename is None:
# No filename was provided so we'll try to get one from the url.
filename = os.path.basename(resource['url'])
upload = uploader.get_resource_uploader(resource)
def add(self, id):
if not id:
abort(409, 'Dataset not identified')
preview_list = pylons_session.get('preview_list', [])
pkg = model.Package.get(id)
if not pkg:
abort(404, 'Dataset not found')
if not self._get(pkg.id):
extent = (pkg.extras.get('bbox-north-lat'),
pkg.extras.get('bbox-west-long'),
pkg.extras.get('bbox-east-long'),
pkg.extras.get('bbox-south-lat'))
preview_list.append({
'id': pkg.id,
'querystring': self._querystring(pkg),
'name': pkg.name,
'extent': extent,
context, only_privatized=True)
interesting_activity_html = fetch_recent_package_activity_list_html(
context, only_resourceful=True)
# Render template
vars = {'invalid_resources': invalid_resources,
'package_activity_html': package_activity_html,
'harvest_activity_html': harvest_activity_html,
'privatized_activity_html': privatized_activity_html,
'interesting_activity_html': interesting_activity_html,
'stats': statistics
}
template = 'admin/dashboard.html'
return base.render(template, extra_vars=vars)
except NotAuthorized:
base.abort(403)
import json
from ckan.controllers.revision import RevisionController
from ckan.controllers.user import UserController
from ckan.controllers.organization import OrganizationController
from ckan.common import c, _, request, response
import ckan.model as model
import ckan.lib.navl.dictization_functions as dictization_functions
import ckan.authz as authz
import ckan.logic as logic
import ckan.lib.helpers as h
import ckan.lib.authenticator as authenticator
import ckan.lib.base as base
import ckan.lib.csrf_token as csrf_token
import ckan.lib.mailer as mailer
abort = base.abort
render = base.render
check_access = ckan.logic.check_access
NotAuthorized = ckan.logic.NotAuthorized
NotFound = ckan.logic.NotFound
get_action = ckan.logic.get_action
unflatten = dictization_functions.unflatten
DataError = dictization_functions.DataError
UsernamePasswordError = logic.UsernamePasswordError
ValidationError = logic.ValidationError
import logging
log = logging.getLogger(__name__)
pformat(harvest_source)))
result = False
elif not last_job_status.get('finished'):
harvest_job_created = last_job_status.get('created')
created = datetime.datetime.strptime(harvest_job_created, HARVEST_JOB_TIMESTAMP_FORMAT)
now = datetime.datetime.now()
if now - created > HARVEST_JOB_TIMEOUT:
log.warn(HARVEST_TIMEOUT_LOGMESSAGE % (
harvest_source.get('title', ''),
pformat(harvest_source)))
result = False
if result:
base.abort(200, SUCCESS_MESSAGE)
else:
base.abort(503, FAILURE_MESSAGE)
def index(self, id):
resource = model.Resource.get(id)
if not resource or resource.state != 'active':
abort(404, "Resource not found")
context = {'model': model,
'session': model.Session,
'user': c.user}
try:
check_access("resource_show", context, {'id': resource.id})
except NotAuthorized, e:
abort(403, "You are not permitted access to this resource")
size_limit = config.get('ckan.datapreview.limit', 5242880)
qa = QA.get_for_resource(resource.id)
format_ = qa.format if qa else None
log.debug('File format (according to QA): %r' % format_)
if not format_:
format_ = resource.format.lower() if resource.format else ''
def _before_org(self, org_id):
'''Returns the organization dict and checks issues are enabled for it.'''
self.context = {'for_view': True}
try:
org = logic.get_action('organization_show')(self.context,
{'id': org_id})
# we should pass org to the template as an extra_var
# directly that's returned from this function
if not issues_helpers.issues_enabled_for_organization(org):
abort(404, _('Issues have not been enabled for this organization'))
return org
except logic.NotFound:
abort(404, _('Dataset not found'))
except p.toolkit.NotAuthorized:
p.toolkit.abort(401,
_('Unauthorized to view issues for this organization'))