onadata.libs.utils package

Submodules

onadata.libs.utils.analytics module

Analytics package for tracking and measuring with services like Segment.

class onadata.libs.utils.analytics.TrackObjectEvent(user_field: str, properties: Dict[str, str], event_name: str = '', event_label: str = '', additional_context: Dict[str, str] | None = None)

Bases: object

Decorator that helps track create and update actions for model objects.

This decorator should only be used on functions that return either an object or a list of objects that you would like to track. For more precise control of what is tracked utilize the track() function

get_event_label() → str

Returns an event label.

get_event_name() → str

Returns an event name.

get_request_origin(request, tracking_properties)

Returns the request origin

get_tracking_properties(label: str | None = None) → dict

Returns tracking properties

set_user() → User | None

Set’s the user attribute.

onadata.libs.utils.analytics.get_user_id(user)

Return a user username or the string ‘anonymous’

onadata.libs.utils.analytics.init_analytics()

Initialize the analytics agents with write credentials.

onadata.libs.utils.analytics.track(user, event_name, properties=None, context=None, request=None)

Record actions with the track() API call to the analytics agents.

onadata.libs.utils.api_export_tools module

API export utility functions.

onadata.libs.utils.api_export_tools.custom_response_handler(request, xform, query, export_type, token=None, meta=None, dataview=False, filename=None, metadata=None)

Returns a HTTP response with export file for download.

onadata.libs.utils.api_export_tools.export_async_export_response(request, export)

Checks the export status and generates the reponse :param request: :param export: :return: response dict example {“job_status”: “Success”, “export_url”: …}

onadata.libs.utils.api_export_tools.external_export_response(export)

Redirects to export_url of XLSReports successful export. In case of a failure, returns a 400 HTTP JSON response with the error message.

onadata.libs.utils.api_export_tools.generate_google_web_flow(request)

Returns a OAuth2WebServerFlow object from the request redirect_uri.

onadata.libs.utils.api_export_tools.get_async_response(job_uuid, request, xform, count=0)

Returns the status of an async task for the given job_uuid.

onadata.libs.utils.api_export_tools.get_entity_list_export_response(request, entity_list, filename)

Returns an EntityList dataset export

onadata.libs.utils.api_export_tools.get_existing_file_format(data, format)

Util function to extract the existing form extension

onadata.libs.utils.api_export_tools.get_metadata_format(data_value)

Returns metadata format/extension

onadata.libs.utils.api_export_tools.include_hxl_row(dv_columns, hxl_columns)

This function returns a boolean value. If the dataview’s columns are not part of the hxl columns, we return False. Returning False would mean that we don’t have to add the hxl column row if there aren’t any hxl columns in the dataview. :param dv_columns - dataview columns :param hxl_columns - hxl columns from the dataview’s xform

:return True or False

onadata.libs.utils.api_export_tools.log_export(request, xform, export_type)

Logs audit logs of export requests.

onadata.libs.utils.api_export_tools.process_async_export(request, xform, export_type, options=None)

Check if should generate export or just return the latest export. Rules for regenerating an export are:

1. Filter included on the exports.

2. New submission done.

3. Always regenerate external exports.

   (External exports uses templates and the template might have

   changed)

Parameters:

• request

• xform

• export_type

• options – additional export params that may include query: export filter token: template url for xls external reports meta: metadataid that contains the external xls report template url remove_group_name: Flag to determine if group names should appear

Returns:

response dictionary

onadata.libs.utils.api_export_tools.response_for_format(data, format=None)

Return appropriately formatted data in Response().

onadata.libs.utils.async_status module

async_status - helper functions to return the status string for celery processes.

onadata.libs.utils.async_status.async_status(status, error=None)

Takes a numeric celery task status and returns equivalent status dictionary with the string representation of the state. If error is passed in the error message is added to the status dictionary.

onadata.libs.utils.async_status.celery_state_to_status(state)

Takes a numeric celery task status and returns equivalent string representation of the state.

onadata.libs.utils.backup_tools module

Backup utilities.

onadata.libs.utils.backup_tools.create_zip_backup(zip_output_file, user, xform=None)

Create a ZIP file with a user’s XForms and submissions.

onadata.libs.utils.backup_tools.restore_backup_from_path(dir_path, username)

Only restores xml submissions, media files are assumed to still be in storage and will be retrieved by the filename stored within the submission

onadata.libs.utils.backup_tools.restore_backup_from_xml_file(xml_instance_path, username)

Creates submission instances in the DB from a submission XML file.

onadata.libs.utils.backup_tools.restore_backup_from_zip(zip_file_path, username)

Restores XForms and submission instances from a ZIP backup file.

onadata.libs.utils.briefcase_client module

ODK BriefcaseClient utils module

class onadata.libs.utils.briefcase_client.BriefcaseClient(url, username, password, user)

Bases: object

ODK BriefcaseClient class

download_instances(form_id, cursor=0, num_entries=100)

Download the XML submissions.

download_manifest(manifest_url, id_string)

Downloads the XForm manifest from an ODK server.

download_media_files(xml_doc, media_path)

Downloads media files from an ODK server.

download_xforms(include_instances=False)

Downloads the XForm XML form an ODK server.

push()

Publishes XForms and XForm submissions.

onadata.libs.utils.briefcase_client.django_file(file_obj, field_name, content_type)

Return an InMemoryUploadedFile file object.

onadata.libs.utils.briefcase_client.node_value(node, tag_name)

Returns the first nodeValue of of an elementin the node with the matching tag_name otherwise returns empty list [].

onadata.libs.utils.cache_tools module

Cache utilities.

exception onadata.libs.utils.cache_tools.CacheLockError

Bases: Exception

Custom exception raised when a cache lock cannot be acquired.

onadata.libs.utils.cache_tools.reset_project_cache(project, request, project_serializer_class)

Clears and sets project cache

onadata.libs.utils.cache_tools.safe_cache_add(key, value, timeout=None)

Safely add a value to the cache.

If the cache is not reachable, the operation silently fails.

Args:

key (str): The cache key to add. value (Any): The value to store in the cache. timeout (int, optional): The cache timeout in seconds. If None,

the default cache timeout will be used.

Returns:

bool: True if the value was added to the cache, False otherwise.

onadata.libs.utils.cache_tools.safe_cache_get(key, default=None)

Safely get a value from the cache.

If the cache is not reachable, the operation silently fails.

Args:

key (str): The cache key to retrieve. default (Any): The default value to return if the cache is inaccessible

or the key does not exist.

Returns:

Any: The value from the cache if accessible, otherwise the default value.

onadata.libs.utils.cache_tools.safe_cache_set(key, value, timeout=None)

Safely set a value in the cache.

If the cache is not reachable, the operation silently fails.

Args:

key (str): The cache key to set. value (Any): The value to store in the cache. timeout (int, optional): The cache timeout in seconds. If None,

the default cache timeout will be used.

Returns:

None

onadata.libs.utils.cache_tools.safe_delete(key)

Safely deletes a given key from the cache.

onadata.libs.utils.cache_tools.safe_key(key)

Return a hashed key.

onadata.libs.utils.cache_tools.set_cache_with_lock(cache_key, modify_callback, cache_timeout=None, lock_expire=30, lock_timeout=10)

Set a cache value with a lock, using a callback function to modify the value.

Use of lock ensures that race conditions are avoided, even when multiple processes or threads attempt to modifiy the cache concurrently.

Args:

cache_key (str): The key under which the value is stored in the cache. modify_callback (callable): A callback function that takes the current cache

value and returns the modified value.

cache_timeout (int, optional): The expiration time for the cached value

in seconds. If None, the default cache timeout is used.

lock_expire (int): The expiration time for the lock in seconds. lock_timeout (int): The maximum time to wait for the lock in seconds.

Raises:

CacheLockError: If the lock cannot be acquired within the specified

lock_timeout.

Returns:

None

onadata.libs.utils.cache_tools.with_cache_lock(cache_key, lock_expire=30, lock_timeout=10)

Context manager for safely setting a cache value with a lock.

Args:

cache_key (str): The key under which the value is stored in the cache. lock_expire (int): The expiration time for the lock in seconds. lock_timeout (int): The maximum time to wait for the lock in seconds.

Raises:

CacheLockError: If the lock cannot be acquired within

the specified lock_timeout.

Yields:

None

onadata.libs.utils.chart_tools module

Chart utility functions.

onadata.libs.utils.chart_tools.build_chart_data(xform, language_index=0, page=0)

Returns chart data for all the fields in the xform.

onadata.libs.utils.chart_tools.build_chart_data_for_field(xform, field, language_index=0, choices=None, group_by=None, data_view=None)

Returns the chart data for a given field.

onadata.libs.utils.chart_tools.build_chart_data_from_widget(widget, language_index=0)

Returns chart data from a widget.

onadata.libs.utils.chart_tools.calculate_ranges(page, items_per_page, total_items)

Return the offset and end indices for a slice.

onadata.libs.utils.chart_tools.find_choice_label(choices, string)

Returns the choice label of the given string.

onadata.libs.utils.chart_tools.get_chart_data_for_field(field_name, xform, accepted_format, group_by, field_xpath=None, data_view=None)

Get chart data for a given xlsform field.

onadata.libs.utils.chart_tools.get_choice_label(choices, string)

string is the name value found in the choices sheet.

Select one names should not contain spaces but some do and this conflicts with Select Multiple fields which use spaces to distinguish multiple choices.

A temporal fix to this is to search the choices list for both the full-string and the split keys.

onadata.libs.utils.chart_tools.get_field_choices(field, xform)

Retrieve field choices from a form survey element :param field: :param xform: :return: Form field choices

onadata.libs.utils.chart_tools.get_field_from_field_name(field_name, xform)

Returns the field if the field_name is in the xform.

onadata.libs.utils.chart_tools.get_field_from_field_xpath(field_xpath, xform)

Returns the field if the field_xpath is in the xform.

onadata.libs.utils.chart_tools.get_field_label(field, language_index=0)

Returns the field’s label or name based on selected language_index.’

onadata.libs.utils.chart_tools.utc_time_string_for_javascript(date_string)

Convert 2014-01-16T12:07:23.322+03 to 2014-01-16T12:07:23.322+03:00

Cant use datetime.str[fp]time here since python 2.7’s %z is platform dependant - http://stackoverflow.com/questions/2609259/converting-string-t o-datetime-object-in-python

onadata.libs.utils.common_tags module

Common tags.

onadata.libs.utils.common_tools module

Common helper functions

exception onadata.libs.utils.common_tools.FilenameMissing

Bases: Exception

Custom Exception for a missing filename.

onadata.libs.utils.common_tools.cmp_to_key(mycmp)

Convert a cmp= function into a key= function

onadata.libs.utils.common_tools.current_site_url(path, host)

Returns fully qualified URL (no trailing slash) for the current site. :param path :return: complete url

onadata.libs.utils.common_tools.filename_from_disposition(content_disposition)

Gets a filename from the given content disposition header.

onadata.libs.utils.common_tools.get_boolean_value(str_var, default=None)

Converts a string into boolean

onadata.libs.utils.common_tools.get_choice_label(label, data_dictionary, language=None)

Return the label matching selected language or simply just the label.

onadata.libs.utils.common_tools.get_choice_label_value(key, value, data_dictionary, language=None)

Return the label of a choice matching the value if the key xpath is a SELECT_ONE otherwise it returns the value unchanged.

onadata.libs.utils.common_tools.get_response_content(response, decode=True)

Gets HTTP content for the given a HTTP response object.

Handles the case where a streaming_content is in the response.

Parameters:

• response – The response to extract content from.

• decode – If true decode as utf-8, default True.

onadata.libs.utils.common_tools.get_uuid(hex_only: bool = True)

Return UUID4 hex value

onadata.libs.utils.common_tools.get_value_or_attachment_uri(key, value, row, data_dictionary, media_xpaths, attachment_list=None, show_choice_labels=False, language=None, host=None)

Gets either the attachment value or the attachment url :param key: used to retrieve survey element :param value: filename :param row: current records row :param data_dictionary: form structure :param include_images: boolean value to either inlcude images or not :param attachment_list: to be used incase row doesn’t have ATTACHMENTS key :return: value

onadata.libs.utils.common_tools.json_stream(data, json_string)

Generator function to stream JSON data

onadata.libs.utils.common_tools.merge_dicts(*dict_args)

Given any number of dicts, shallow copy and merge into a new dict, precedence goes to key value pairs in latter dicts.

onadata.libs.utils.common_tools.report_exception(subject, info, exc_info=None)

Formats an exception then posts it to sentry and if not in debug or testing sends email to mail_admins.

onadata.libs.utils.common_tools.retry(tries, delay=3, backoff=2)

Adapted from code found here:

http://wiki.python.org/moin/PythonDecoratorLibrary#Retry

Retries a function or method until it returns True.

Parameters:

delay – sets the initial delay in seconds, and backoff sets the

factor by which the delay should lengthen after each failure. :param backoff: must be greater than 1, or else it isn’t really a backoff. :param tries: must be at least 0, and delay greater than 0.

onadata.libs.utils.common_tools.str_to_bool(str_var)

Return boolean True or False if string s represents a boolean value

onadata.libs.utils.common_tools.track_task_progress(additions, total=None)

Updates the current export task with number of submission processed. Updates in batches of settings EXPORT_TASK_PROGRESS_UPDATE_BATCH defaults to 100. :param additions: :param total: :return:

onadata.libs.utils.country_field module

CountryField - provides a list of countries.

class onadata.libs.utils.country_field.CountryField(*args, **kwargs)

Bases: CharField

A CharField that limits the choices to country codes.

get_internal_type()

onadata.libs.utils.csv_builder module

CSV export utility functions.

class onadata.libs.utils.csv_builder.AbstractDataFrameBuilder(username, id_string, filter_query=None, group_delimiter='/', split_select_multiples=True, binary_select_multiples=False, start=None, end=None, remove_group_name=False, xform=None, include_labels=False, include_labels_only=False, include_images=True, include_hxl=False, win_excel_utf8=False, total_records=None, index_tags=('[', ']'), value_select_multiples=False, show_choice_labels=True, include_reviews=False, language=None, host=None, entity_list: EntityList | None = None)

Bases: object

Abstract Data Frame Builder class

ADDITIONAL_COLUMNS = ['_id', '_uuid', '_submission_time', '_date_modified', '_tags', '_notes', '_version', '_duration', '_submitted_by', '_total_media', '_media_count', '_media_all_received']

BINARY_SELECT_MULTIPLES = False

IGNORED_COLUMNS = ['_xform_id_string', '_status', '_attachments', '_geolocation', '_bamboo_dataset_id', '_deleted_at', '_edited']

VALUE_SELECT_MULTIPLES = False

Group functionality used by any DataFrameBuilder i.e. XLS, CSV and KML

class onadata.libs.utils.csv_builder.CSVDataFrameBuilder(username, id_string, filter_query=None, group_delimiter='/', split_select_multiples=True, binary_select_multiples=False, start=None, end=None, remove_group_name=False, xform=None, include_labels=False, include_labels_only=False, include_images=False, include_hxl=False, win_excel_utf8=False, total_records=None, index_tags=('[', ']'), value_select_multiples=False, show_choice_labels=False, include_reviews=False, language=None, host=None, entity_list: EntityList | None = None)

Bases: AbstractDataFrameBuilder

CSV data frame builder

export_to(path, cursor, dataview=None)

Export a CSV formated to the given path.

onadata.libs.utils.csv_builder.get_column_names_only(columns, data_dictionary, group_delimiter)

Return column names as a list.

onadata.libs.utils.csv_builder.get_labels_from_columns(columns, data_dictionary, group_delimiter, language=None)

Return column labels

onadata.libs.utils.csv_builder.get_prefix_from_xpath(xpath)

Returns xpath prefix.

onadata.libs.utils.csv_builder.remove_dups_from_list_maintain_order(lst)

Removes duplicates from a list and still maintains the order.

onadata.libs.utils.csv_builder.write_to_csv(path, rows, columns, columns_with_hxl=None, remove_group_name=False, data_dictionary=None, group_delimiter='/', include_labels=False, include_labels_only=False, include_hxl=False, win_excel_utf8=False, total_records=None, index_tags=('[', ']'), language=None)

Writes rows to a file in CSV format.

onadata.libs.utils.csv_import module

CSV data import module.

onadata.libs.utils.csv_import.dict2xmlsubmission(submission_dict, xform, instance_id, submission_date)

Creates and xml submission from an appropriate dict (& other data)

Parameters:

• submission_dict (dict) – A dict containing form submission data.

• xfrom (onadata.apps.logger.models.XForm) – The submission’s XForm.

• instance_id (string) – The submission/instance uuid.

• submission_date (string) – An isoformatted datetime string.

Returns:

An xml submission string

Return type:

string

onadata.libs.utils.csv_import.dict_merge(part_a, part_b)

Returns a merger of two dicts a and b

credits: https://www.xormedia.com/recursively-merge-dictionaries-in-python

Parameters:

• a (dict) – The “Part A” dict

• b (dict) – The “Part B” dict

Returns:

The merger

Return type:

dict

onadata.libs.utils.csv_import.dict_pathkeys_to_nested_dicts(dictionary)

Turns a flat dict to a nested dict

Takes a dict with pathkeys or “slash-namespaced” keys and inflates them into nested dictionaries i.e:- d[‘/path/key/123’] -> d[‘path’][‘key’][‘123’]

Parameters:

dictionary (dict) – A dict with one or more “slash-namespaced” keys

Returns:

A nested dict

Return type:

dict

onadata.libs.utils.csv_import.failed_import(rollback_uuids, xform, exception, status_message)

Report a failed import. :param rollback_uuids: The rollback UUIDs :param xform: The XForm that failed to import to :param exception: The exception object :return: The async_status result

onadata.libs.utils.csv_import.flatten_split_select_multiples(row: Dict[str, Any], select_multiples: List[str]) → dict

Flattens a select_multiple question that was previously split into different choice columns into one column

onadata.libs.utils.csv_import.get_async_csv_submission_status(job_uuid)

Gets CSV Submision progress or result Can be used to pol long running submissions :param str job_uuid: The submission job uuid returned by _submit_csv.delay :return: Dict with import progress info (insertions & total) :rtype: Dict

onadata.libs.utils.csv_import.get_columns_by_type(type_list, form_json)

Returns a column that match types passed within the field_list

Parameters:

type_list (list) – A list containing strings that represent

XLS field types :param dict form_json: A dict representing the contents of a form :return: A list containing the column names that store values passed within the type_list :rtype: list

onadata.libs.utils.csv_import.get_submission_meta_dict(xform, instance_id)

Generates metadata for our submission

Checks if instance_id belongs to an existing submission. If it does, it’s considered an edit and its uuid gets deprecated. In either case, a new one is generated and assigned.

Parameters:

• xform (onadata.apps.logger.models.XForm) – The submission’s XForm.

• instance_id (string) – The submission/instance uuid.

Returns:

The metadata dict

Return type:

dict

onadata.libs.utils.csv_import.submission_xls_to_csv(xls_file)

Convert a submission xls file to submissions csv file

Parameters:

xls_file – submissions xls file

Returns:

csv_file

onadata.libs.utils.csv_import.submit_csv(username, xform, csv_file, overwrite=False)

Imports CSV data to an existing form

Takes a csv formatted file or string containing rows of submission/instance and converts those to xml submissions and finally submits them by calling onadata.libs.utils.logger_tools.safe_create_instance()

Parameters:

• username (str) – the submission user

• xform (onadata.apps.logger.models.XForm) – The submission’s XForm.

• csv_file ((str or file)) – A CSV formatted file with submission rows.

Returns:

If sucessful, a dict with import summary else dict with error str.

Return type:

Dict

onadata.libs.utils.csv_import.validate_column(row, columns, constraint_check)

Validates columns within a row making sure data constraints are not broken

Takes a list of column headers to validate and a function of which is used to validate the data is valid.

Parameters:

• row (Dict) – The row of which the columns values should be validated

• columns (List) – A list of headers to be validated within the row.

• constraint_check (func) – A function used to validate column data

Returns:

Returns a tuple containing the validity status of the rows

data and the validated_data if successful else it contains the invalid_data

onadata.libs.utils.csv_import.validate_csv_file(csv_file, xform)

Validates a CSV File

Takes a CSV Formatted file or sring containing rows of submission data and validates that the file is valid enough to be processed.

Parameters:

csv_file ((str or file)) – A CSV formatted file or string with

submission rows :param onadata.apps.logger.models.XForm xform: The submission’s XForm :rtype: Dict :returns: A dict containing the validity of the CSV file as well as additional columns(additional_col) if any when successful else it returns an error message(error_msg)

onadata.libs.utils.csv_import.validate_row(row, columns)

Validate row of data making sure column constraints are enforced

Takes a csv row containing data from a submission and validates the data making sure data types are enforced.

Parameters:

• row (Dict) – The csv row

• columns (Dict) – A dict containing the column headers to be validated

and the function that should check that the columns constraint is not broken :return: Returns a tuple containing the validated row and errors found within the row if any :rtype: Tuple

onadata.libs.utils.csv_reader module

CsvReader class module.

class onadata.libs.utils.csv_reader.CsvReader(path)

Bases: object

Typical usage:

csv_reader = CsvReader(path)
for d in csv_reader.iter_dicts():
Variable.objects.create(**d)

close()

Closes the file handle.

iter_dicts()

Iterate over CSV rows as dict items.

next()

A CsvReader object is iterable (since we have defined __iter__ and next methods. Each iteration of this object returns a row of data.

open(path)

Opens a file handle sets a CSV reader.

onadata.libs.utils.decorators module

decorators module

onadata.libs.utils.decorators.check_obj(func)

Checks if the first argument is truthy and then calls the underlying function.

onadata.libs.utils.decorators.is_owner(view_func)

Redirects to login if not owner.

onadata.libs.utils.dict_tools module

Dict utility functions module.

onadata.libs.utils.dict_tools.csv_dict_to_nested_dict(csv_dict, select_multiples=None)

Converts a CSV dict to nested dicts.

onadata.libs.utils.dict_tools.dict_lists2strings(adict)

Convert lists in a dict to joined strings.

Parameters:

d – The dict to convert.

Returns:

The converted dict.

onadata.libs.utils.dict_tools.dict_paths2dict(adict)

Turns a dict with ‘/’ in keys to a nested dict.

onadata.libs.utils.dict_tools.floip_response_headers_dict(data, xform_headers)

Returns a dict from matching xform headers and floip responses.

onadata.libs.utils.dict_tools.get_values_matching_key(doc, key)

Returns iterator of values in ‘doc’ with the matching ‘key’.

onadata.libs.utils.dict_tools.list_to_dict(items, value)

Converts a list into a dict.

onadata.libs.utils.dict_tools.merge_list_of_dicts(list_of_dicts, override_keys: list | None = None)

Merges a list of dicts to return one dict.

onadata.libs.utils.dict_tools.query_list_to_dict(query_list_str)

Returns a ‘label’ and ‘text’ from a Rapidpro values JSON string as a dict.

onadata.libs.utils.dict_tools.remove_indices_from_dict(obj)

Removes indices from a obj dict.

onadata.libs.utils.email module

Email utility functions.

class onadata.libs.utils.email.ProjectInvitationEmail(invitation: ProjectInvitation, url: str)

Bases: object

A class to send a project invitation email

get_email_data() → dict[str, str]

Get the email data to be sent

get_template_data() → dict[str, str]

Get context data for the templates

send() → None

Send project invitation email

onadata.libs.utils.email.get_account_lockout_email_data(username, ip_address, end=False)

Generates both the email upon start and end of account lockout

onadata.libs.utils.email.get_project_invitation_url(request: HttpRequest)

Get project invitation url

onadata.libs.utils.email.get_verification_email_data(email, username, verification_url, request)

Returns the verification email content

onadata.libs.utils.email.get_verification_url(redirect_url, request, verification_key)

Returns the verification_url

onadata.libs.utils.email.send_generic_email(email, message_txt, subject)

Sends an email.

onadata.libs.utils.export_builder module

ExportBuilder

class onadata.libs.utils.export_builder.ExportBuilder

Bases: object

Utility class for generating multiple formats of data export to file.

BINARY_SELECT_MULTIPLES = False

CONVERT_FUNCS = {'date': <function string_to_date_with_xls_validation>, 'dateTime': <function ExportBuilder.<lambda>>, 'decimal': <class 'float'>, 'int': <class 'int'>}

EXTRA_FIELDS = ['_id', '_uuid', '_submission_time', '_index', '_parent_table_name', '_parent_index', '_tags', '_notes', '_version', '_duration', '_submitted_by']

GROUP_DELIMITER = '/'

GROUP_DELIMITERS = ['/', '.']

GROUP_DELIMITER_DOT = '.'

GROUP_DELIMITER_SLASH = '/'

IGNORED_COLUMNS = ['_xform_id_string', '_status', '_attachments', '_geolocation', '_bamboo_dataset_id', '_deleted_at']

INCLUDE_HXL = False

INCLUDE_IMAGES = True

INCLUDE_LABELS = False

INCLUDE_LABELS_ONLY = False

INCLUDE_REVIEWS = False

REPEAT_INDEX_TAGS = ('[', ']')

SHOW_CHOICE_LABELS = False

SPLIT_SELECT_MULTIPLES = True

TRUNCATE_GROUP_TITLE = False

TYPES_TO_CONVERT = ['int', 'decimal', 'date']

VALUE_SELECT_MULTIPLES = False

XLS_SHEET_NAME_MAX_CHARS = 31

classmethod convert_type(value, data_type)

Convert data to its native type e.g. string ‘1’ to int 1 @param value: the string value to convert @param data_type: the native data type to convert to @return: the converted value

classmethod decode_mongo_encoded_fields(row, encoded_fields)

Update encoded fields with their corresponding xpath

classmethod format_field_title(abbreviated_xpath, field_delimiter, data_dictionary, remove_group_name=False, gps_extra_field=False)

Format the field title.

get_choice_label_from_dict(label)

Returns the choice label for the default language

If a label for the target language is blank then the default language is used

get_default_language(languages)

Return the default languange of the XForm.

get_fields(dataview, section, key)

Return list of element value with the key in section[‘elements’].

classmethod get_valid_sheet_name(desired_name, existing_names)

Returns a valid sheet_name based on the desired names

language = None

pre_process_row(row, section)

Split select multiples, gps and decode . and $

section_by_name(name)

Return section by the given name.

set_survey(survey, xform=None, include_reviews=False)

Set’s the XForm XML survey instance.

classmethod split_gps_components(row, gps_fields)

Splits GPS components into their own fields.

classmethod split_select_multiples(row, select_multiples, select_values=False, binary_select_multiples=False, show_choice_labels=False, data_dictionary=None, language=None)

Split select multiple choices in a submission to individual columns.

Parameters:

• row – single submission dict

• select_multiples – list of XPATHs and choices of select multiple questions.

• binary_select_multiples – if True the value of the split columns will be 1 when the choice has been selected otherwise it will be 0.

• select_values – the value of the split columns will be the name/value of the choice when selected otherwise blank/None.

• show_choice_labels – Show a choice label instead of the value/True/False/1/0.

• data_dictionary – A DataDictionary/XForm object

• language – specific language as defined in the XLSForm.

Returns:

the row dict with select multiples choice as fields in the row

to_flat_csv_export(path, data, username, id_string, filter_query, **kwargs)

Generates a flattened CSV file for submitted data.

to_xlsx_export(path, data, *args, **kwargs)

Export data to a spreadsheet document.

to_zipped_csv(path, data, *args, **kwargs)

Export CSV formatted files from data and zip the files.

to_zipped_sav(path, data, *args, **kwargs)

Generates the SPSS zipped file format export.

url = None

onadata.libs.utils.export_builder.decode_mongo_encoded_section_names(data)

Recursively decode mongo keys.

Parameters:

data – A dictionary to decode.

onadata.libs.utils.export_builder.dict_to_joined_export(data, index, indices, name, survey, row, host, media_xpaths=None)

Converts a dict into one or more tabular datasets :param data: current record which can be changed or updated :param index: keeps count of record number :param indices: a dictionary storing list values if data is a dict :param name: the name of the survey :param survey: the survey :param row: current record that remains unchanged on this function’s recall

onadata.libs.utils.export_builder.encode_if_str(row, key, encode_dates=False, sav_writer=None)

Encode a string value in row[key].

onadata.libs.utils.export_builder.get_data_dictionary_from_survey(survey)

Creates a DataDictionary instance from an XML survey instance.

onadata.libs.utils.export_builder.is_all_numeric(items)

Check if all items on the list are numeric, zero padded numbers will not be considered as numeric.

Parameters:

items – list of values to be checked

Return boolean:

onadata.libs.utils.export_builder.string_to_date_with_xls_validation(date_str)

Try to convert a string to a date object.

Parameters:

date_str – string to convert

Returns:

object if converted, otherwise date string

onadata.libs.utils.export_tools module

Export tools

onadata.libs.utils.export_tools.check_pending_export(xform, export_type, options, minutes=5)

Check for pending export done within a specific period of time and returns the export :param xform: :param export_type: :param options: :param minutes :return:

onadata.libs.utils.export_tools.clean_keys_of_slashes(record)

Replaces the slashes found in a dataset keys with underscores :param record: list containing a couple of dictionaries :return: record with keys without slashes

onadata.libs.utils.export_tools.create_export_object(xform, export_type, options)

Return an export object that has not been saved to the database.

onadata.libs.utils.export_tools.generate_attachments_zip_export(export_type, username, id_string, export_id=None, options=None, xform=None)

Generates zip export of attachments.

param: export_type params: username: logged in username params: id_string: xform id_string params: export_id: ID of export object associated with the request param: options: additional parameters required for the lookup.

ext: File extension of the generated export

onadata.libs.utils.export_tools.generate_entity_list_export(entity_list: EntityList) → GenericExport

Generates a CSV for an EntityList dataset

onadata.libs.utils.export_tools.generate_external_export(export_type, username, id_string, export_id=None, options=None, xform=None)

Generates external export using ONA data through an external service.

param: export_type params: username: logged in username params: id_string: xform id_string params: export_id: ID of export object associated with the request param: options: additional parameters required for the lookup.

data_id: instance id query: filter_query for custom queries meta: metadata associated with external export token: authentication key required by external service

onadata.libs.utils.export_tools.generate_geojson_export(export_type, username, id_string, metadata=None, export_id=None, options=None, xform=None)

Generates Linked Geojson export

Parameters:

• export_type – type of export

• username – logged in username

• id_string – xform id_string

• export_id – ID of export object associated with the request

• options – additional parameters required for the lookup.

• ext – File extension of the generated export

onadata.libs.utils.export_tools.generate_kml_export(export_type, username, id_string, export_id=None, options=None, xform=None)

Generates kml export for geographical data

Parameters:

• export_type – type of export

• username – logged in username

• id_string – xform id_string

• export_id – ID of export object associated with the request

• options – additional parameters required for the lookup.

• extension – File extension of the generated export

onadata.libs.utils.export_tools.generate_osm_export(export_type, username, id_string, export_id=None, options=None, xform=None)

Generates osm export for OpenStreetMap data

Parameters:

• export_type – type of export

• username – logged in username

• id_string – xform id_string

• export_id – ID of export object associated with the request

• options – additional parameters required for the lookup.

• ext – File extension of the generated export

onadata.libs.utils.export_tools.get_entity_list_dataset(entity_list: EntityList) → Iterator[dict]

Get entity data for a an EntityList dataset

Args:

entity_list (EntityList): The EntityList whose data will be returned

Returns:

An iterator of dicts which represent the json data for Entities belonging to the dataset

onadata.libs.utils.export_tools.get_export_options(options)

Returns expirt options as defined in Export.EXPORT_OPTION_FIELDS from a list of provided options to be saved with each Export object.

onadata.libs.utils.export_tools.get_latest_generic_export(instance, export_type, options=None) → GenericExport | None

Retrieve the latest GenericExport

onadata.libs.utils.export_tools.get_or_create_export(export_id, xform, export_type, options)

Returns an existing export object or creates a new one with the given options.

onadata.libs.utils.export_tools.get_or_create_export_object(export_id, options, xform, export_type)

Get or create export object.

Parameters:

• export_id – Export ID

• options – Options to convert to export options

• xform – XForm to export

• export_type – The type of export

Returns:

A new or found export object

onadata.libs.utils.export_tools.get_osm_data_kwargs(xform)

Return kwargs for OsmData queryset for given xform

onadata.libs.utils.export_tools.get_repeat_index_tags(index_tags)

Gets a comma separated string index_tags

Retuns a tuple of two strings with SUPPORTED_INDEX_TAGS,

onadata.libs.utils.export_tools.increment_index_in_filename(filename)

filename should be in the form file.ext or file-2.ext - we check for the dash and index and increment appropriately

onadata.libs.utils.export_tools.kml_export_data(id_string, user, xform=None)

KML export data from form submissions.

onadata.libs.utils.export_tools.md5hash(string)

Return the MD5 hex digest of the given string.

onadata.libs.utils.export_tools.newest_export_for(xform, export_type, options)

Retrieve the latest export given the following arguments:

param: xform param: export_type param: options: additional parameters required for the lookup.

remove_group_name: boolean flag group_delimiter: “/” or “.” with “/” as the default split_select_multiples: boolean flag binary_select_multiples: boolean flag index_tag: (‘[’, ‘]’) or (‘_’, ‘_’)

onadata.libs.utils.export_tools.parse_request_export_options(params)

Parse export options in the request object into values returned in a list. The list represents a boolean for whether the group name should be removed, the group delimiter, and a boolean for whether select multiples should be split.

onadata.libs.utils.export_tools.should_create_new_export(instance, export_type, options, request=None, is_generic=False)

Function that determines whether to create a new export. param: xform param: export_type param: options: additional parameters required for the lookup.

remove_group_name: boolean flag group_delimiter: “/” or “.” with “/” as the default split_select_multiples: boolean flag binary_select_multiples: boolean flag index_tag: (‘[’, ‘]’) or (‘_’, ‘_’)

params: request: Get params are used to determine if new export is required

onadata.libs.utils.export_tools.upload_template_for_external_export(server, file_obj)

Uploads an Excel template to the XLSReport server.

Returns the status code with the server response.

onadata.libs.utils.export_tools.write_temp_file_to_path(suffix, content, file_path)

Write a temp file and return the name of the file. :param suffix: The file suffix :param content: The content to write :param file_path: The path to write the temp file to :return: The filename written to

onadata.libs.utils.google module

Google utility functions.

onadata.libs.utils.google.create_flow(redirect_uri: str | None = None) → Flow

Returns a Google Flow from client configuration.

onadata.libs.utils.gravatar module

Gravatar utils module

onadata.libs.utils.gravatar.email_md5(user)

Returns the hash of an email for the user

onadata.libs.utils.gravatar.get_gravatar_img_link(user)

Returns the Gravatar image URL

onadata.libs.utils.gravatar.gravatar_exists(user)

Checks if the Gravatar URL exists

onadata.libs.utils.image_tools module

Image utility functions module.

onadata.libs.utils.image_tools.flat(*nums)

Build a tuple of ints from float or integer arguments. Useful because PIL crop and resize require integer points. source: https://gist.github.com/16a01455

onadata.libs.utils.image_tools.generate_media_download_url(obj, expiration: int = 3600)

Returns a HTTP response of a media object or a redirect to the image URL for S3 and Azure storage objects.

onadata.libs.utils.image_tools.get_dimensions(size, longest_side)

Return integer tuple of width and height given size and longest_side length.

onadata.libs.utils.image_tools.image_url(attachment, suffix)

Return url of an image given size(@param suffix) e.g large, medium, small, or generate required thumbnail

onadata.libs.utils.image_tools.is_azure_storage()

Checks if azure storage is in use

onadata.libs.utils.image_tools.resize(filename, extension)

Resize an image into multiple sizes.

onadata.libs.utils.image_tools.resize_local_env(filename, extension)

Resize images in a local environment.

onadata.libs.utils.log module

Log utility functions and classes.

class onadata.libs.utils.log.AuditLogHandler(model='')

Bases: Handler

Audit logging handler class.

emit(record)

Do whatever it takes to actually log the specified logging record.

This version is intended to be implemented by subclasses and so raises a NotImplementedError.

get_model(name)

Import and return the model under the given name.

class onadata.libs.utils.log.Enum(**enums)

Bases: object

Enum class - dict-like class

onadata.libs.utils.log.audit_log(action, request_user, account_user, message, audit, request, level=10)

Create a log message based on these params

@param action: Action performed e.g. form-deleted @param request_username: User performing the action @param account_username: The formhub account the action was performed on @param message: The message to be displayed on the log @param level: log level @param audit: a dict of key/values of other info pertaining to the action e.g. form’s id_string, submission uuid @return: None

onadata.libs.utils.logger_tools module

logger_tools - Logger app utility functions.

class onadata.libs.utils.logger_tools.BaseOpenRosaResponse(*args, **kwargs)

Bases: HttpResponse

The base HTTP response class with OpenRosa headers.

status_code = 201

class onadata.libs.utils.logger_tools.OpenRosaNotAuthenticated(*args, **kwargs)

Bases: Response

An HTTP response class with OpenRosa headers for the Not Authenticated response.

status_code = 401

class onadata.libs.utils.logger_tools.OpenRosaResponse(*args, **kwargs)

Bases: BaseOpenRosaResponse

An HTTP response class with OpenRosa headers for the created response.

status_code = 201

class onadata.libs.utils.logger_tools.OpenRosaResponseBadRequest(*args, **kwargs)

Bases: OpenRosaResponse

An HTTP response class with OpenRosa headers for the Bad Request response.

status_code = 400

class onadata.libs.utils.logger_tools.OpenRosaResponseForbidden(*args, **kwargs)

Bases: OpenRosaResponse

An HTTP response class with OpenRosa headers for the Forbidden response.

status_code = 403

class onadata.libs.utils.logger_tools.OpenRosaResponseNotAllowed(*args, **kwargs)

Bases: OpenRosaResponse

An HTTP response class with OpenRosa headers for the Not Allowed response.

status_code = 405

class onadata.libs.utils.logger_tools.OpenRosaResponseNotFound(*args, **kwargs)

Bases: OpenRosaResponse

An HTTP response class with OpenRosa headers for the Not Found response.

status_code = 404

class onadata.libs.utils.logger_tools.PublishXForm(xml_file, user)

Bases: object

A class to publish an XML XForm file.

publish_xform()

Publish an XForm XML file.

onadata.libs.utils.logger_tools.check_edit_submission_permissions(request_user, xform)

Checks edit submission permissions.

onadata.libs.utils.logger_tools.check_submission_encryption(xform: XForm, xml: bytes) → NoReturn

Check that the submission is encrypted or unencrypted depending on the encryption status of an XForm.

The submission is invalid if the XForm’s encryption status is different from the submissions

onadata.libs.utils.logger_tools.check_submission_permissions(request, xform)

Check that permission is required and the request user has permission.

The user does no have permissions iff:

• the user is authed,

• either the profile or the form require auth,

• the xform user is not submitting.

Since we have a username, the Instance creation logic will handle checking for the forms existence by its id_string.

Returns:

None.

Raises:

PermissionDenied based on the above criteria.

onadata.libs.utils.logger_tools.commit_cached_elist_num_entities() → None

Commit cached EntityList num_entities counter to the database

Commit is successful if no other process holds the lock

onadata.libs.utils.logger_tools.create_entity_from_instance(instance: Instance, registration_form: RegistrationForm) → Entity

Create an Entity

Args:

instance (Instance): Submission from which the Entity is created from registration_form (RegistrationForm): RegistrationForm creating the Entity

Returns:

Entity: A newly created Entity

onadata.libs.utils.logger_tools.create_instance(username, xml_file, media_files, status='submitted_via_web', uuid=None, date_created_override=None, request=None)

I used to check if this file had been submitted already, I’ve taken this out because it was too slow. Now we’re going to create a way for an admin to mark duplicate instances. This should simplify things a bit. Submission cases: * If there is a username and no uuid, submitting an old ODK form. * If there is a username and a uuid, submitting a new ODK form.

onadata.libs.utils.logger_tools.create_or_update_entity_from_instance(instance: Instance) → None

Create or Update Entity from Instance

Args:

instance (Instance): Instance to create/update Entity from

onadata.libs.utils.logger_tools.create_xform_version(xform: XForm, user: User) → XFormVersion

Creates an XFormVersion object for the passed in XForm

onadata.libs.utils.logger_tools.dec_elist_num_entities(pk: int) → None

Decrement EntityList num_entities counter

Updates cached counter if cache is not locked. Else, the database counter is updated.

Args:

pk (int): Primary key for EntityList

onadata.libs.utils.logger_tools.delete_xform_submissions(xform: XForm, deleted_by: User, instance_ids: list[int] | None = None, soft_delete: bool = True) → None

“Delete subset or all submissions of an XForm

Parameters:

• xform – XForm object

• deleted_by – User initiating the delete

• instance_ids – List of instance ids to delete, None to delete all

• soft_delete – Flag to soft delete or hard delete

Returns:

None

onadata.libs.utils.logger_tools.dict2xform(jsform, form_id, root=None, username=None, gen_uuid=False)

Converts a dictionary containing submission data into an XML Submission for the appropriate form.

Parameters:

• (dict) (jsform) – A python dictionary object containing the submission data

• XForm) (form_id (str or) – An XForm object or a string value representing the forms id_string

• (str) (root) – An optional string that should be used as the root nodes name. Defaults to None

Param:

username (str): An optional string representing a users username. Used alongside the form_id to locate the XForm object the user is trying to submit data too. Defaults to None

Returns:

Returns a string containing the Submission XML

Return type:

str

onadata.libs.utils.logger_tools.generate_aws_media_url(file_path: str, content_disposition: str, expiration: int = 3600)

Generate S3 URL.

onadata.libs.utils.logger_tools.generate_content_disposition_header(name, extension, show_date=True)

Returns the a Content-Description header formatting string,

onadata.libs.utils.logger_tools.generate_media_url_with_sas(file_path: str, expiration: int = 3600)

Generate Azure storage URL.

onadata.libs.utils.logger_tools.get_entity_json_from_instance(instance: Instance, registration_form: RegistrationForm) → dict

Parses Instance json and returns Entity json

Args:

instance (Instance): Submission to create Entity

Returns:

dict: Entity properties

onadata.libs.utils.logger_tools.get_filtered_instances(*args, **kwargs)

Get filtered instances - mainly to allow mocking in tests

onadata.libs.utils.logger_tools.get_first_record(queryset)

Returns the first item in a queryset sorted by id.

onadata.libs.utils.logger_tools.get_storages_media_download_url(file_path: str, content_disposition: str, expires_in=3600) → str | None

Get the media download URL for the storages backend.

Parameters:

• file_path – The path to the media file.

• content_disposition – The content disposition header.

• expires_in – The expiration time in seconds.

Returns:

The media download URL.

onadata.libs.utils.logger_tools.get_uuid_from_submission(xml)

Extracts and returns the UUID from a submission XML.

onadata.libs.utils.logger_tools.get_xform_from_submission(xml, username, uuid=None, request=None)

Gets the submissions target XForm.

Retrieves the target XForm by either utilizing the uuid param or the uuid retrievable from the xml or the id_string retrievable from the XML. Only returns form if request_user has permission to submit.

Parameters:

• xml ((str)) – The submission in XML form

• username ((str)) – The owner of the target XForm

• uuid ((str)) – The target XForms universally unique identifier.

Default: None :param (django.http.request) request: Request object. Default: None

onadata.libs.utils.logger_tools.inc_elist_num_entities(pk: int) → None

Increment EntityList num_entities counter

Updates cached counter if cache is not locked. Else, the database counter is updated

Args:

pk (int): Primary key for EntityList

onadata.libs.utils.logger_tools.inject_instanceid(xml_str, uuid)

Adds the uuid as the <instanceID/> to an XML string xml_str.

onadata.libs.utils.logger_tools.publish_form(callback)

Calls the callback function to publish a XLSForm and returns appropriate message depending on exception throw during publishing of a XLSForm.

onadata.libs.utils.logger_tools.remove_metadata_fields(data)

Clean out unneccessary metadata fields

onadata.libs.utils.logger_tools.response_with_mimetype_and_name(mimetype, name, extension=None, show_date=True, file_path=None, use_local_filesystem=False, full_mime=False, expires_in=3600)

Returns a HttpResponse with Content-Disposition header set

Triggers a download on the browser.

onadata.libs.utils.logger_tools.safe_create_instance(username, xml_file, media_files, uuid, request, instance_status: str = 'submitted_via_web')

Create an instance and catch exceptions.

Returns:

A list [error, instance] where error is None if there was no error.

onadata.libs.utils.logger_tools.save_attachments(xform, instance, media_files, remove_deleted_media=False)

Saves attachments for the given instance/submission.

onadata.libs.utils.logger_tools.save_submission(xform, xml, media_files, new_uuid, submitted_by, status, date_created_override, checksum, request=None)

Persist a submission into the ParsedInstance model.

onadata.libs.utils.logger_tools.set_default_openrosa_headers(response)

Sets the default OpenRosa headers into a response object.

onadata.libs.utils.logger_tools.soft_delete_entities_bulk(entity_qs: QuerySet, deleted_by=None) → None

Soft delete Entities in bulk

Args:

entity_qs QuerySet(Entity): Entity queryset deleted_by (User): User initiating the delete

onadata.libs.utils.logger_tools.store_temp_file(data)

Creates a temporary file with the data and returns it.

onadata.libs.utils.logger_tools.update_attachment_tracking(instance)

Takes an Instance object and updates attachment tracking fields

onadata.libs.utils.logger_tools.update_entity_from_instance(uuid: str, instance: Instance, registration_form: RegistrationForm) → Entity | None

Updates Entity

Args:

uuid (str): uuid of the Entity to be updated instance (Instance): Submission that updates an Entity

Returns:

Entity | None: updated Entity if uuid valid, else None

onadata.libs.utils.middleware module

Custom middleware classes.

class onadata.libs.utils.middleware.BaseMiddleware(get_response)

Bases: object

BaseMiddleware - The base middleware class.

class onadata.libs.utils.middleware.ExceptionLoggingMiddleware(get_response)

Bases: object

The exception logging middleware class - prints the exception traceback.

process_exception(request, exception)

Prints the exception traceback.

class onadata.libs.utils.middleware.HTTPResponseNotAllowedMiddleware(get_response)

Bases: object

The HTTP Not Allowed middleware class - renders the 405.html template.

class onadata.libs.utils.middleware.LocaleMiddlewareWithTweaks(get_response)

Bases: LocaleMiddleware

Overrides LocaleMiddleware from django with:

Khmer km language code in Accept-Language is rewritten to km-kh

process_request(request)

class onadata.libs.utils.middleware.OperationalErrorMiddleware(get_response)

Bases: BaseMiddleware

Captures requests returning 500 status code. Then retry it against master database.

process_exception(request, exception)

Handle retrying OperatuonalError exceptions.

class onadata.libs.utils.middleware.SqlLogging(get_response)

Bases: object

SQL logging middleware.

onadata.libs.utils.model_tools module

Model utility functions.

onadata.libs.utils.model_tools.get_columns_with_hxl(survey_elements)

Returns a dictionary whose keys are xform field names and values are instance::hxl values set on the xform :param include_hxl - boolean value :param survey_elements - survey elements of an xform return dictionary or None

onadata.libs.utils.model_tools.queryset_iterator(queryset, chunksize=100)

Iterate over a Django Queryset.

This method loads a maximum of chunksize (default: 100) rows in its memory at the same time while django normally would load all rows in its memory. Using the iterator() method only causes it to not preload all the classes.

See https://docs.djangoproject.com/en/2.1/ref/models/querysets/#iterator

onadata.libs.utils.model_tools.set_uuid(obj)

Only give an object a new UUID if it does not have one.

onadata.libs.utils.mongo module

Utility functions for MongoDB operations.

onadata.libs.utils.numeric module

The int_or_parse_error utility function.

onadata.libs.utils.numeric.int_or_parse_error(value, error_string)

If value is not an int raise a parse error with error_string, which is a format string that takes one argument, the value.

onadata.libs.utils.openid_connect_tools module

OpenID Connect Tools

class onadata.libs.utils.openid_connect_tools.OpenIDHandler(provider_configuration: dict)

Bases: object

Base OpenID Connect Handler

Implements functions neccessary to implement the OpenID Connect ‘code’ or ‘id_token’ authorization flow

end_openid_provider_session()

Clears the SSO cookie set at authentication and redirects the User to the end_session endpoint provided by the provider configuration

get_claim_values(claim_list: list, decoded_token: dict)

Retrieves claim values from a decoded_token based on the claim name either configured in the provider configuration or the passed in claim

:params claim_list: A list of strings containing the name of claim decoded_token: A dict containing the decoded values of an ID Token

make_login_request(nonce: int, state=None)

Makes a login request to the “authorization_endpoint” listed in the provider_configuration

obtain_id_token_from_code(code: str, openid_provider: str = '')

Obtain an ID Token using the Authorization Code flow

verify_and_decode_id_token(id_token: str, cached_nonce: bool = False, openid_provider: str = '')

Verifies that the ID Token passed was signed and sent by the Open ID Connect Provider and that the client is one of the audiences then decodes the token and returns the decoded information

onadata.libs.utils.organization_utils module

OrganizationProfile utility functions

onadata.libs.utils.organization_utils.get_organization_members(organization)

Returns a dictionary of organizations members

onadata.libs.utils.osm module

OSM utility module.

onadata.libs.utils.osm.get_combined_osm(osm_list)

Combine osm xml form list of OsmData objects

onadata.libs.utils.osm.osm_flat_dict(instance_id)

Flat dict of OSM tags for the specified submission.

Each key starts with ‘osm_*’.

onadata.libs.utils.osm.parse_osm(osm_xml, include_osm_id=False)

Parses OSM XML and return a list of ways or nodes.

onadata.libs.utils.osm.parse_osm_nodes(osm_xml, include_osm_id=False)

Converts an OSM XMl to a list of GEOSGeometry objects

onadata.libs.utils.osm.parse_osm_tags(node, include_osm_id=False)

Retrieves all the tags from a osm xml node

onadata.libs.utils.osm.parse_osm_ways(osm_xml, include_osm_id=False)

Converts an OSM XMl to a list of GEOSGeometry objects

onadata.libs.utils.osm.save_osm_data(instance_id)

Includes the OSM data in the specified submission json data.

onadata.libs.utils.project_utils module

project_utils module - apply project permissions to a form.

exception onadata.libs.utils.project_utils.ExternalServiceRequestError

Bases: Exception

Custom Exception class for External service requests i.e Formbuilder

onadata.libs.utils.project_utils.assign_change_asset_permission(service_url: str, asset_id: str, usernames: List[str], session: Session) → Response

Bulk assigns the change_asset permission to a group of users on KPI

onadata.libs.utils.project_utils.get_project_users(project)

Return project users with the role assigned to them.

onadata.libs.utils.project_utils.propagate_project_permissions(project: Project, headers: dict | None = None, use_asset_owner_auth: bool = True) → None

Propagates Project permissions to external services(KPI)

onadata.libs.utils.project_utils.retrieve_asset_permissions(service_url: str, asset_id: str, session: Session) → Dict[str, List[str]]

Retrieves the currently assigned asset permissions for each user on KPI

onadata.libs.utils.project_utils.set_project_perms_to_object(obj: Any, project: Project) → None

Apply project permissions to an object

Args:

obj: Object to set permissions for project: Project under which the object belongs to

onadata.libs.utils.project_utils.set_project_perms_to_xform(xform, project)

Apply project permissions to a form, this usually happens when a new form is being published or it is being moved to a new project.

onadata.libs.utils.qrcode module

QR code utility function.

onadata.libs.utils.qrcode.generate_qrcode(message)

Generate a QRCode, settings options and output.

onadata.libs.utils.quick_converter module

The QuickConverter form class - publishes XLSForms.

class onadata.libs.utils.quick_converter.QuickConverter(data=None, files=None, auto_id='id_%s', prefix=None, initial=None, error_class=<class 'django.forms.utils.ErrorList'>, label_suffix=None, empty_permitted=False, field_order=None, use_required_attribute=None, renderer=None)

Bases: Form

The QuickConverter form - publishes XLSForms.

base_fields = {'xls_file': <django.forms.fields.FileField object>}

declared_fields = {'xls_file': <django.forms.fields.FileField object>}

property media

Return all media required to render the widgets on this form.

publish(user)

Create and return a DataDictionary object.

onadata.libs.utils.string module

String utility function str2bool - converts yes, true, t, 1 to True else returns the argument value v.

onadata.libs.utils.string.str2bool(value)

String utility function str2bool - converts “yes”, “true”, “t”, “1” to True else returns the argument value v.

onadata.libs.utils.timing module

Custom date utility functions.

onadata.libs.utils.timing.calculate_duration(start_time, end_time)

This function calculates duration when given start and end times. An empty string is returned if either of the time formats does not match ‘_format’ format else, the duration is returned

onadata.libs.utils.timing.get_date(_object=None)

Returns a date formatted string of the date_modified of the given _object.

onadata.libs.utils.timing.get_header_date_format(date_modified)

Returns date_modified as a string with the format %a, %d %b %Y %H:%M:%S GMT

onadata.libs.utils.timing.last_modified_header(last_modified_date)

Returns a dictionary with the ‘Last-Modified’ key and value.

onadata.libs.utils.user_auth module

User authentication utility functions.

class onadata.libs.utils.user_auth.HttpResponseNotAuthorized(*args, **kwargs)

Bases: HttpResponse

HttpResponse that sets basic authentication prompt.

status_code = 401

onadata.libs.utils.user_auth.add_cors_headers(response)

Add CORS headers to the HttpResponse response instance.

onadata.libs.utils.user_auth.basic_http_auth(func)

A basic authentication decorator.

onadata.libs.utils.user_auth.check_and_set_form_by_id(pk, request)

Checks xform by pk and returns an xform if request.user has permission.

onadata.libs.utils.user_auth.check_and_set_form_by_id_string(username, id_string, request)

Checks xform by id_string and returns an xform if request.user has permission.

onadata.libs.utils.user_auth.check_and_set_user(request, username)

Returns a User object or a string to redirect.

onadata.libs.utils.user_auth.check_and_set_user_and_form(username, id_string, request)

Checks and returns an xform and owner if request.user has permission.

onadata.libs.utils.user_auth.get_user_default_project(user)

Return’s the user’s default project, creates it if it does not exist.’

onadata.libs.utils.user_auth.get_xform_and_perms(username, id_string, request)

Returns the xform with the matching id_string, and the permissions the request.user has.

onadata.libs.utils.user_auth.get_xform_users_with_perms(xform)

Similar to django-guadian’s get_users_with_perms here the query makes use of the xformuserobjectpermission_set to return a dictionary of users with a list of permissions to the XForm object. The query in use is not as expensive as the one in use with get_users_with_perms

onadata.libs.utils.user_auth.has_edit_permission(xform, owner, request, shared=False)

Checks if the request.user has edit permissions to the xform.

onadata.libs.utils.user_auth.has_permission(xform, owner, request, shared=False)

Checks if the request.user has the necessary permissions to an xform.

onadata.libs.utils.user_auth.helper_auth_helper(request)

Authenticates a user via basic authentication.

onadata.libs.utils.user_auth.http_auth_string(username, password)

Return a basic authentication string with username and password.

onadata.libs.utils.user_auth.invalidate_and_regen_tokens(user)

Invalidates a users Access and Temp tokens and generates new ones

onadata.libs.utils.user_auth.set_api_permissions_for_user(user)

Sets the permissions to allow a user to access the APU.

onadata.libs.utils.user_auth.set_profile_data(data, content_user)

Set user profile data.

onadata.libs.utils.validators module

Module containing custom validator classes for the User Model

class onadata.libs.utils.validators.PreviousPasswordValidator(history_limt=5)

Bases: object

Class validates password was not previously recorded.

get_help_text()

Returns the help text.

validate(password, user=None)

Checks password was not used previously.

onadata.libs.utils.viewer_tools module

Utility functions for data views.

onadata.libs.utils.viewer_tools.create_attachments_zipfile(attachments, zip_file)

Return a zip file with submission attachments.

Parameters:

• attachments – an Attachments queryset.

• zip_file – a file object, more likely a NamedTemporaryFile() object.

onadata.libs.utils.viewer_tools.export_def_from_filename(filename)

Return file extension and mimetype from filename.

onadata.libs.utils.viewer_tools.generate_enketo_form_defaults(xform, **kwargs)

Return Enketo default options for preloading data into a web form.

onadata.libs.utils.viewer_tools.get_client_ip(request)

Return an IP from HTTP_X_FORWARDED_FOR or REMOTE_ADDR request headers.

arguments: request – HttpRequest object.

onadata.libs.utils.viewer_tools.get_enketo_urls(form_url, id_string, instance_xml=None, instance_id=None, return_url=None, **kwargs) → Dict[str, str]

Return Enketo URLs.

onadata.libs.utils.viewer_tools.get_form(kwargs)

Return XForm object by applying kwargs on an XForm queryset.

onadata.libs.utils.viewer_tools.get_form_url(request, username=None, protocol='https', preview=False, xform_pk=None, generate_consistent_urls=False)

Return a form list url endpoint to be used to make a request to Enketo.

For example, it will return https://example.com and Enketo will know to look for the form list at https://example.com/formList. If a username is provided then Enketo will request the form list from https://example.com/[username]/formList. Same applies for preview if preview is True and also to a single form when xform_pk is provided.

onadata.libs.utils.viewer_tools.get_path(path, suffix)

Apply the suffix to the path.

onadata.libs.utils.viewer_tools.handle_enketo_error(response)

Handle enketo error response.

onadata.libs.utils.viewer_tools.image_urls(instance)

Return image urls of all image attachments of the submission instance.

arguments: instance – Instance submission object.

onadata.libs.utils.viewer_tools.image_urls_for_form(xform)

Return image urls of all image attachments of the xform.

onadata.libs.utils.viewer_tools.parse_xform_instance(xml_str)

‘xml_str’ is a str object holding the XML of an XForm instance.

Return a python object representation of this XML file.

Module contents