geonode.utils

Attributes

MAX_EXTENT
FULL_ROTATION_DEG
HALF_ROTATION_DEG
DEFAULT_TITLE
DEFAULT_ABSTRACT
INVALID_PERMISSION_MESSAGE
ALPHABET
ALPHABET_REVERSE
BASE
SIGN_CHARACTER
SQL_PARAMS_RE
FORWARDED_HEADERS
XML_PARSER
signalnames
signals_store
id_none
logger
http_client
json_serializer_k_map

Exceptions

ServerDoesNotExist

Common base class for all non-exit exceptions.

Classes

OGC_Server	OGC Server object.
OGC_Servers_Handler	OGC Server Settings Convenience dict.
DisableDjangoSignals	Python3 class temporarily disabling django signals on model creation.
HttpClient

Functions

mkdtemp([dir])
unzip_file(upload_file[, extension, tempdir])	Unzips a zipfile into a temporary directory and returns the full path of the .shp file inside (if any)
extract_tarfile(upload_file[, extension, tempdir])	Extracts a tarfile into a temporary directory and returns the full path of the .shp file inside (if any)
get_dataset_name(dataset)	Get the workspace where the input layer belongs
get_dataset_workspace(dataset)	Get the workspace where the input layer belongs
get_headers(request, url, raw_url[, allowed_hosts])
_get_basic_auth_info(request)	grab basic auth info
batch_delete(request)
_split_query(query)	split and strip keywords, preserve space
bbox_swap(bbox)
bbox_to_wkt(x0, x1, y0, y1[, srid, include_srid])
_v(coord, x[, source_srid, target_srid])
bbox_to_projection(native_bbox[, target_srid])	native_bbox must be in the form
bounds_to_zoom_level(bounds, width, height)
llbbox_to_mercator(llbbox)
mercator_to_llbbox(bbox)
forward_mercator(lonlat)	Given geographic coordinates, return a x,y tuple in spherical mercator.
inverse_mercator(xy)	Given coordinates in spherical mercator, return a lon,lat tuple.
resolve_object(request, model, query[, permission, ...])	Resolve an object using the provided query and check the optional
json_response([body, errors, url, redirect_to, ...])	Create a proper JSON response. If body is provided, this is the response.
num_encode(n)
num_decode(s)
format_urls(a, values)
build_abstract(resourcebase[, url, includeURL])
build_caveats(resourcebase)
build_social_links(request, resourcebase)
check_shp_columnnames(layer)	Check if shapefile for a given layer has valid column names.
clone_shp_field_defn(srcFieldDefn, name)	Clone an existing ogr.FieldDefn with a new name
rename_shp_columnnames(inLayer, fieldnames)	Rename columns in a layer to those specified in the given mapping
fixup_shp_columnnames(inShapefile, charset[, tempdir])	Try to fix column names and warn the user
id_to_obj(id_)
printsignals()
run_subprocess(*cmd, **kwargs)
parse_datetime(value)
_convert_sql_params(cur, query)
raw_sql(query[, params, ret])	Execute raw query
get_client_ip(request)
get_client_host(request)
check_ogc_backend(backend_package)	Check that geonode use a particular OGC Backend integration
get_dir_time_suffix()	Returns the name of a folder with the 'now' time as suffix
zip_dir(basedir, archivename)
copy_tree(src, dst[, symlinks, ignore])
extract_archive(zip_file, dst)
chmod_tree(dst[, permissions])
slugify_zh(text[, separator])	Make a slug from the given text, which is simplified from slugify.
get_legend_url(instance, style_name, /[, service_url, ...])
set_resource_default_links(instance, layer[, prune])
add_url_params(url, params)	Add GET params to provided URL being aware of existing.
json_serializer_producer(dictionary)
is_monochromatic_image(image_url[, image_data])
find_by_attr(lst, val[, attr])	Returns an object if the id matches in any list of objects
build_absolute_uri(url)
remove_credentials_from_url(url)
extract_ip_or_domain(url)
get_xpath_value(→ Optional[str])
get_geonode_app_types()
get_supported_datasets_file_types()
get_allowed_extensions()
safe_path_leaf(path)	A view that is not vulnerable to malicious file access.

Module Contents

geonode.utils.MAX_EXTENT = 20037508.34

geonode.utils.FULL_ROTATION_DEG = 360.0

geonode.utils.HALF_ROTATION_DEG = 180.0

geonode.utils.DEFAULT_TITLE = ''

geonode.utils.DEFAULT_ABSTRACT = ''

geonode.utils.INVALID_PERMISSION_MESSAGE

geonode.utils.ALPHABET

geonode.utils.ALPHABET_REVERSE

geonode.utils.BASE

geonode.utils.SIGN_CHARACTER = '$'

geonode.utils.SQL_PARAMS_RE

geonode.utils.FORWARDED_HEADERS = ['content-type', 'content-disposition']

geonode.utils.XML_PARSER: Final

geonode.utils.signalnames = ['class_prepared', 'm2m_changed', 'post_delete', 'post_init', 'post_save', 'post_syncdb',...

geonode.utils.signals_store

geonode.utils.id_none

geonode.utils.logger

exception geonode.utils.ServerDoesNotExist

Bases: Exception

Common base class for all non-exit exceptions.

class geonode.utils.OGC_Server(ogc_server, alias)

Bases: object

OGC Server object.

alias

server

__getattr__(item)

property credentials

Returns a tuple of the server's credentials.

property datastore_db

Returns the server's datastore dict or None.

property ows

The Open Web Service url for the server.

property rest

The REST endpoint for the server.

property public_url

The global public endpoint for the server.

property internal_ows

The Open Web Service url for the server used by GeoNode internally.

property hostname

property netloc

__str__()

class geonode.utils.OGC_Servers_Handler(ogc_server_dict)

OGC Server Settings Convenience dict.

servers

_servers

ensure_valid_configuration(alias)

Ensures the settings are valid.

ensure_defaults(alias)

Puts the defaults into the settings dictionary for a given connection where no settings is provided.

__getitem__(alias)

__setitem__(key, value)

__iter__()

all()

geonode.utils.mkdtemp(dir=settings.MEDIA_ROOT)

geonode.utils.unzip_file(upload_file, extension='.shp', tempdir=None)

Unzips a zipfile into a temporary directory and returns the full path of the .shp file inside (if any)

geonode.utils.extract_tarfile(upload_file, extension='.shp', tempdir=None)

Extracts a tarfile into a temporary directory and returns the full path of the .shp file inside (if any)

geonode.utils.get_dataset_name(dataset)

Get the workspace where the input layer belongs

geonode.utils.get_dataset_workspace(dataset)

Get the workspace where the input layer belongs

geonode.utils.get_headers(request, url, raw_url, allowed_hosts=[])

geonode.utils._get_basic_auth_info(request)

grab basic auth info

geonode.utils.batch_delete(request)

geonode.utils._split_query(query)

split and strip keywords, preserve space separated quoted blocks.

geonode.utils.bbox_swap(bbox)

geonode.utils.bbox_to_wkt(x0, x1, y0, y1, srid='4326', include_srid=True)

geonode.utils._v(coord, x, source_srid=4326, target_srid=3857)

geonode.utils.bbox_to_projection(native_bbox, target_srid=4326)

native_bbox must be in the form

(‘-81.3962935’, ‘-81.3490249’, ‘13.3202891’, ‘13.3859614’, ‘EPSG:4326’)

geonode.utils.bounds_to_zoom_level(bounds, width, height)

geonode.utils.llbbox_to_mercator(llbbox)

geonode.utils.mercator_to_llbbox(bbox)

geonode.utils.forward_mercator(lonlat)

Given geographic coordinates, return a x,y tuple in spherical mercator.

If the lat value is out of range, -inf will be returned as the y value

geonode.utils.inverse_mercator(xy)

Given coordinates in spherical mercator, return a lon,lat tuple.

geonode.utils.resolve_object(request, model, query, permission='base.view_resourcebase', user=None, permission_required=True, permission_msg=None)

Resolve an object using the provided query and check the optional permission. Model views should wrap this function as a shortcut.

query - a dict to use for querying the model permission - an optional permission to check permission_required - if False, allow get methods to proceed permission_msg - optional message to use in 403

geonode.utils.json_response(body=None, errors=None, url=None, redirect_to=None, exception=None, content_type=None, status=None)

Create a proper JSON response. If body is provided, this is the response. If errors is not None, the response is a success/errors json object. If redirect_to is not None, the response is a success=True, redirect_to object If the exception is provided, it will be logged. If body is a string, the exception message will be used as a format option to that string and the result will be a success=False, errors = body % exception

geonode.utils.num_encode(n)

geonode.utils.num_decode(s)

geonode.utils.format_urls(a, values)

geonode.utils.build_abstract(resourcebase, url=None, includeURL=True)

geonode.utils.build_caveats(resourcebase)

geonode.utils.build_social_links(request, resourcebase)

geonode.utils.check_shp_columnnames(layer)

Check if shapefile for a given layer has valid column names. If not, try to fix column names and warn the user

geonode.utils.clone_shp_field_defn(srcFieldDefn, name)

Clone an existing ogr.FieldDefn with a new name

geonode.utils.rename_shp_columnnames(inLayer, fieldnames)

Rename columns in a layer to those specified in the given mapping

geonode.utils.fixup_shp_columnnames(inShapefile, charset, tempdir=None)

Try to fix column names and warn the user

geonode.utils.id_to_obj(id_)

geonode.utils.printsignals()

class geonode.utils.DisableDjangoSignals(disabled_signals=None, skip=False)

Python3 class temporarily disabling django signals on model creation.

usage: with DisableDjangoSignals():

skip

stashed_signals

disabled_signals

__enter__()

__exit__(exc_type, exc_val, exc_tb)

disconnect(signal)

reconnect(signal)

geonode.utils.run_subprocess(*cmd, **kwargs)

geonode.utils.parse_datetime(value)

geonode.utils._convert_sql_params(cur, query)

geonode.utils.raw_sql(query, params=None, ret=True)

Execute raw query param ret=True returns data from cursor as iterator

geonode.utils.get_client_ip(request)

geonode.utils.get_client_host(request)

geonode.utils.check_ogc_backend(backend_package)

Check that geonode use a particular OGC Backend integration

Parameters:

backend_package (str) – django app of backend to use

Returns:

bool

Return type:

bool

class geonode.utils.HttpClient

timeout = 5

retries = 1

pool_maxsize = 10

backoff_factor = 0.3

pool_connections = 10

status_forcelist = (500, 502, 503, 504)

username = 'admin'

password = 'admin'

request(url, method='GET', data=None, headers={}, stream=False, timeout=None, retries=None, user=None, verify=False)

get(url, data=None, headers={}, stream=False, timeout=None, user=None, verify=False)

post(url, data=None, headers={}, stream=False, timeout=None, user=None, verify=False)

geonode.utils.http_client

geonode.utils.get_dir_time_suffix()

Returns the name of a folder with the ‘now’ time as suffix

geonode.utils.zip_dir(basedir, archivename)

geonode.utils.copy_tree(src, dst, symlinks=False, ignore=None)

geonode.utils.extract_archive(zip_file, dst)

geonode.utils.chmod_tree(dst, permissions=511)

geonode.utils.slugify_zh(text, separator='_')

Make a slug from the given text, which is simplified from slugify. It remove the other args and do not convert Chinese into Pinyin :param text (str): initial text :param separator (str): separator between words :return (str):

geonode.utils.get_legend_url(instance, style_name, /, service_url=None, dataset_name=None, version='1.3.0', sld_version='1.1.0', width=20, height=20, params=None)

geonode.utils.set_resource_default_links(instance, layer, prune=False, **kwargs)

geonode.utils.add_url_params(url, params)

Add GET params to provided URL being aware of existing.

Parameters:

• url – string of target URL

• params – dict containing requested params to be added

Returns:

string with updated URL

>> url = ‘http://stackoverflow.com/test?answers=true’ >> new_params = {‘answers’: False, ‘data’: [‘some’,’values’]} >> add_url_params(url, new_params) ‘http://stackoverflow.com/test?data=some&data=values&answers=false’

geonode.utils.json_serializer_k_map

geonode.utils.json_serializer_producer(dictionary)

• usage:

  serialized_obj =

  json_serializer_producer(model_to_dict(instance))

• dump to file:

  with open(‘data.json’, ‘w’) as outfile:

  json.dump(serialized_obj, outfile)

• read from file:

  with open(‘data.json’, ‘r’) as infile:

  serialized_obj = json.load(infile)

geonode.utils.is_monochromatic_image(image_url, image_data=None)

geonode.utils.find_by_attr(lst, val, attr='id')

Returns an object if the id matches in any list of objects

geonode.utils.build_absolute_uri(url)

geonode.utils.remove_credentials_from_url(url)

geonode.utils.extract_ip_or_domain(url)

geonode.utils.get_xpath_value(element: lxml.etree.Element, xpath_expression: str, nsmap: dict | None = None) → str | None

geonode.utils.get_geonode_app_types()

geonode.utils.get_supported_datasets_file_types()

geonode.utils.get_allowed_extensions()

geonode.utils.safe_path_leaf(path)

A view that is not vulnerable to malicious file access.