Utilities

Common utilities

megforms.utils.itersubclasses(cls)

Generator over all subclasses of a given class, in depth first order.

>>> list(itersubclasses(int)) == [bool]
True
>>> class A(object): pass
>>> class B(A): pass
>>> class C(A): pass
>>> class D(B,C): pass
>>> class E(D): pass
>>>
>>> for cls in itersubclasses(A):
...     print(cls.__name__)
B
D
E
C
>>> # get ALL (new-style) classes currently defined
>>> [cls.__name__ for cls in itersubclasses(object)] 
['type', ...'tuple', ...]

http://code.activestate.com/recipes/576949-find-all-subclasses-of-a-given-class/

megforms.utils.get_observation_models() → Tuple[ObservationModel]

Returns tuple of all subclasses of BaseAuditModel

megforms.utils.get_subform_models() → Iterable[type[BaseSubObservation]]

Iterates over all subform models (models representing sub observations), including CustomSubObservationInstance, but excluding abstract models (base HIQA and IPC subobservation models)

megforms.utils.first(collection: Iterable[T]) → T | None

Returns first element of given iterable, even if its a set or an empty collection

megforms.utils.get_observation_model_paths() → set[str]

Returns an iterable of strings representing valid python paths for observation models. Use this to filter out invalid AuditForms from your queryset

megforms.utils.send_mail(subject, body_text, addr_from, addr_to, reply_to: list[str] = None, fail_silently=False, attachments=None, body_html=None, connection=None, headers=None, cc=None)

Sends a multipart email containing text and html versions with CC. Copy of Lib/site-packages/email_extras/utils.py:28 but with added CC and without encryption.

megforms.utils.get_dotted_path(cls: type | Callable) → str

Returns python dotted path for class

megforms.utils.get_func_name(func: type | Callable) → str

Based on a utility function from OpenCensus. Return a name which includes the module name and function name.

megforms.utils.has_subforms(observation_model: ObservationModel) → bool

Tells whether given observation model has hardcoded subforms

megforms.utils.common_public_report_widgets(method)

Prepend common widgets to return value of decorated method

megforms.utils.get_last_modified_date(qs: QuerySet, modified_field_name: str = 'modified') → datetime | None

Gets last modified date for the qs of objects Returns modification date of last modified item

Parameters:

• qs – the queryset of any model containing a date field

• modified_field_name – a datetime field representing when field was last modified

class megforms.utils.DefaultDateValue(*args, **kwargs)

A callable that can be used to set default date in a django field relative to date of model creation. For instance, use this to set date 2 weeks ahead: DefaultDateValue(timezone.timedelta(days=14))

Parameters:

• timedelta – timedelta added on top of from_date to create the default date. Use a negative timedelta to subtract instead of adding

• from_date – leave empty to use current date at the time this class is called, or hardcode a datetime

deconstruct()

Return a 3-tuple of class import path, positional arguments, and keyword arguments.

megforms.utils.autonomous_transaction(var_name='autonomuos')

Allows saving objects inside atomic transaction independently of the transaction rollback/commit. Wrapped function must be a view. Wrap in method_decorator to use with class based views

Parameters:

var_name – name of the variable storing the object(s) to be saved. Variable needs to be attached to the request object.

megforms.utils.FilteredRelatedListFilter(qs_or_callable)

Wrapper for RelatedFieldListFilter that shows a subset of items

megforms.utils.request_institution_getter(request: HttpRequest) → InstitutionQuerySet

Returns institution choices based on request returns empty QS if user is not an auditor or not logged in

class megforms.utils.PermissionValidator

Django validator that validates permission string whether it is a valid permission. Stores any “seen” permissions in its instance as cache to reduce db queries.

megforms.utils.get_permission_by_name(permission: str) → Permission

Gets permission object based on permission string.

Parameters:

permission – Permission string, for example ‘calendar.add_auditschedule’

Returns:

Permission object

Raises:

Permission.DoesNotExist if permission cannot be found

megforms.utils.permission_to_str(permission: Permission) → str

Given permission object, returns string representing the permission in str format. inverse of get_permission_by_name. Note that this function needs to query the related content_type object.

megforms.utils.filter_auditors_by_permission(auditors: QuerySet[Auditor], permission: Permission | str, *objs: AuditForm, include_global=True) → QuerySet[Auditor]

Filters given user queryset by permission and returns only users who have the given permission. Includes superusers who have all permissions implicitly as well as users who have the permission via group membership.

Warning

This utility does not filter by form. If user has the permission globally, but not access to the form, they will still be included in the result.

Do not rely on this function to filter users by form. Use for_audit_form() instead.

Parameters:

• auditors – the queryset to filter on

• permission – perm obj or its codename, for example qip.change_issue

• objs – optional list of forms. If passed, users who have the permission only for those forms will be included in the result query. If blank, only global permissions are taken into account

Include include_global:

whether to include global auditors and superusers

Returns:

queryset of users where each one has the requested permission globally or in any of the passed objs

megforms.utils.auditors_with_perm(permission: Permission | str, *objs: AuditForm, include_global=True) → Q

A Q object for Auditor queryset that filters objects based on permission.

Parameters:

• permission – perm obj or its codename, for example qip.change_issue

• objs – optional list of forms. If passed, users who have the permission only for those forms will be included in the result query. If blank, only global permissions are taken into account

Include include_global:

whether to include global auditors and superusers

Returns:

Q object applicable to Auditor queryset that results in filtering by permission

megforms.utils.get_permissions(permissions: Collection[str] = ('megforms.view_auditor', 'megforms.add_auditor', 'megforms.change_auditor', 'megforms.delete_auditor', 'megforms.view_team', 'megforms.view_auditform', 'megforms.delete_auditsession', 'megforms.view_department', 'megforms.add_ward', 'megforms.view_ward', 'megforms.change_ward', 'megforms.delete_ward', 'megforms.add_room', 'megforms.change_room', 'megforms.delete_room', 'calendar.view_auditschedule', 'calendar.add_auditschedule', 'calendar.change_auditschedule', 'calendar.delete_auditschedule', 'dashboard_widgets.view_dashboard', 'dashboard_widgets.view_publicdashboardviewconfig', 'information.view_information', 'qip.view_commonissuechoice', 'qip.export_data', 'emails.view_reportrule', 'calendar.view_timelineevent', 'audit_builder.export_data'), validate=False) → QuerySet

Gets permission objects based on permission strings. Optionally validates that number of permissions found matches input permissions

Parameters:

• permissions – Permission strings, for example [‘calendar.add_auditschedule’]

• validate – whether to validate that all permissions in input string exist (results in additional SQL query)

Raises:

ValueError – if validate is set to True and not all permissions are valid permission strings

Returns:

Permission QuerySet

megforms.utils.calculate_image_size(size: tuple[int, int], new_height: int) → tuple[int, int]

Calculates new image size to match expected height

Parameters:

• size – size tuple (width, height)

• height – new height

Returns:

size tuple (width, height)

megforms.utils.get_bmp_thumbnail_path(image_path: str, size_h=150) → str

Gives path to bmp file that works with Excel, and create one if needed

Parameters:

• image_path –

• size_h – expected width of new bitmap

Returns:

path to proper bmp thumbnail ready to put in excel sheet

megforms.utils.strict_password_enforced(user: User) → bool

Checks whether user account should be covered by strict password checks due to patient data access

megforms.utils.daterange_to_months(from_date: date, to_date: date) → str

Converts range of dates and returns month or multiple months containing selected dates

>>> utils.daterange_to_months( timezone.datetime(2018, 7, 1), timezone.datetime(2018, 7, 12))
'July 2018'
>>> utils.daterange_to_months( timezone.datetime(2017, 12, 1), timezone.datetime(2018, 1, 12))
'December 2017 - January 2018'
>>> utils.daterange_to_months( timezone.datetime(2018, 7, 1), timezone.datetime(2018, 8, 12))
'July - August 2018'

megforms.utils.get_compliance_level(compliance_levels: tuple, compliance: float | None) → int | None

Gets compliance level bucket based on compliance and min compliance level

Parameters:

• compliance_levels – tuple (first, second) containing compliance levels

• compliance – compliance percentage (float) between 0 and 1 or None

Returns:

Compliance level: 0 (lowest) to 2 (highest)

megforms.utils.get_compliance_level_name(compliance_levels: tuple, compliance: float | None, names: Sequence[T] = ('bad', 'fine', 'excellent'), default: T = '') → T

Picks the name of compliance level based on compliance and tuple of levels

Parameters:

• compliance_levels – tuple (first, second) containing compliance levels

• compliance – compliance percentage (float) between 0 and 1, or None

• names – tuple of custom names for compliance (ordered from worst to best)

• default – the default value, used if compliance is unknown (None)

Returns:

string name of compliance level or empty string if cannot be determined

megforms.utils.get_compliance_level_value(compliance_levels: tuple, compliance: float | None) → float

Gets the compliance level value for the provided compliance.

Parameters:

• compliance_levels – tuple (float) containing compliance levels

• compliance – compliance percentage (float) between 0 and 1 or None

Returns:

Compliance level value: (float) a value from compliance_levels

megforms.utils.clean_sheet_name(name: str) → str

Strip out unsupported characters from the spreadsheet name and truncate to supported length

megforms.utils.get_institution_label_variants_key(institution: Institution | None) → str

Return a key for constants.INSTITUTION_LABEL_VARIANTS. Default to ‘institution’. If an Institution is present, return that Institution’s institution_label choice.

megforms.utils.get_institution_label(institution: Institution | None, capitalize: bool = False, plural: bool = False, lazy: bool = True, plural_all: bool = False) → str

Gets a string representing how given institution group names their institutions. By default, it can be “institution” but other institution groups may have other preferences.

Parameters:

• institution – The institution.

• capitalize – Whether to capitalize the label or not.

• plural – Whether to pluralize the label or not.

• lazy – Whether to return a lazy string or not.

• plural_all – Whether to return a plural variant that includes the word “all”.

megforms.utils.get_department_label(institution: Institution | None, capitalize=False, plural=False, lazy=True) → str

Gets a string representing how given institution names their departments. By default it can be “department” but other institutions may have other preferences.

megforms.utils.get_ward_label(institution: Institution | None, capitalize=False, plural=False, lazy=True) → str

Gets a string representing how given institution names their wards. By default it can be “ward” but other institutions may have other preferences.

megforms.utils.get_user_label(institution: Institution | None, capitalize=False, plural=False, lazy=True) → str

Gets a string representing how given institution names their users. By default, it can be “user” but other institutions may have other preferences.

Parameters:

• institution – The institution.

• capitalize – Whether to capitalize the label or not.

• plural – Whether to pluralize the label or not.

• lazy – Whether to return a lazy string or not.

Returns:

user object label string

megforms.utils.get_issue_label(audit_form: 'AuditForm' | None, capitalize=False, plural=False, lazy=True) → str

Gets a string representing how given institution names their QIP issues. By default it can be “issue” but other institutions may have other preferences.

megforms.utils.get_sub_issue_label(audit_form: 'AuditForm' | None, capitalize=False, plural=False, lazy=True) → str

Gets a string representing how given institution names their QIP sub-issues. By default it can be “issue” but other institutions may have other preferences.

megforms.utils.get_audit_subforms(audit_form: AuditForm, subform_names: Sequence[str] = ()) → Iterator[Subform]

Gets all subforms for given audit (hardcoded and custom), limited by subform name

Parameters:

• audit_form – AuditForm instance

• subform_names –

  only subforms with those names will be returned:

  • For hardcoded subforms, model_name is used

  • For custom subforms, name is used for filtering

Returns:

iterator of subforms (mix of model class and instances)

class megforms.utils.LocaleWrapper(language: str | User | None)

Context manager to temporarily activate another language. After exiting the context manager will restore previously set locale.

>>> with LocaleWrapper('es'):
>>>     return gettext("Text that will be translated to Spanish")
"Texto que será traducido al español"

Parameters:

language – language code. Alternatively user instance can be passed. The context manager will determine language using accounts.utils.get_user_language().

class megforms.utils.TimezoneWrapper(tz: str | tzinfo)

Context manager to temporarily activate another timezone and return to the previous timezone after exiting

megforms.utils.iter_translations(lazy_string, skip_untranslated=False, languages=('en', 'pl', 'ro', 'de', 'fr', 'es', 'ar', 'it', 'pt-br', 'zh-hans')) → tuple[str, str]

Iterates over translations of given string

Parameters:

• lazy_string – string to be translated (must be a lazy string, for example from gettext_lazy())

• skip_untranslated – whether languages that don’t have a unique translation for this string should be skipped

• languages – list of language codes to get translations for

Yield:

Tuple of language code and translation in that language

megforms.utils.get_domain_url(request: WSGIRequest = None) → str

Gets current domain url with the correct http prefix. For example:

• http://localhost:8000

• https://audits.megsupporttools.com

megforms.utils.get_redirect_back_url(request: HttpRequest) → str

Gets referrer url, or navigates to landing page if referrer header is not present.

Use if you need a url to navigate the user away from current page

megforms.utils.make_absolute_url(url: str, request: HttpRequest = None) → str

Prepends domain name to url to make the address absolute. Returns url as-is if already prepended

>>> make_absolute_url('/login')
"http://localhost:8000/login"

megforms.utils.get_object_url(obj: Model) → str

Gets url for given model instance. The model should implement get_absolute_url method

Parameters:

obj – a model instance

Returns:

url to the object as defined in its get_absolute_url, or empty string

megforms.utils.read_as_base64(file: IO) → str

Returns file contents str encoded in base64

Parameters:

file – opened file object or other stream

Returns:

file contents as base64 str

megforms.utils.noop(*args, **kwargs)

A no-op function that does nothing

megforms.utils.empty_generator() → Iterable[None]

A generator that yields no items

megforms.utils.format_datetime(value: datetime | date | time, force_date: bool = False) → str

Formats given date/time int a human-friendly format ready to display to the user

Parameters:

• value – Value to be formatted

• force_date – Whether to force the string to always show date only format

Returns:

class megforms.utils.DateTimeRange(start: datetime.datetime, end: datetime.datetime)

Represents a datetime range with start and an end datetime

Create new instance of DateTimeRange(start, end)

start: datetime

start point of the range

end: datetime

end point of the range

static from_queryset(qs: QuerySet, field_name: str = 'created') → DateTimeRange | None

Gets datetime range from a model field based on a range in a given queryset

Parameters:

• qs – a queryset

• field_name – the datetime field in the model

Returns:

range of datetimes in the given field in the queryset

static from_observations(*observations: ObservationQueryset, date_getter: Callable[[Observation], datetime.datetime] | None = None, sessions: QuerySet[AuditSession] | None = None) → DateTimeRange | None

Creates date range from QuerySets of observations based on their submission date (session end date, or value returned from date_getter)

Parameters:

• observations – querysets of observations

• date_getter – optional function that takes observation and returns its date. If not provided, session end date is used

• sessions – optional session queryset to used. Filter down the queryset to optimize the query (for example filter by form ids). Unpublished sessions should already be excluded from the qs if necessary, as this utility does not apply further filters to the sessions queryset apart from filtering by observations. If this parameter is not present, all published sessions are used, which can be inefficient (Task #31445)

Returns:

The range starting at the earliest observation time, to the latest observation. Null if there are no observations

get_dates() → tuple[datetime.date, datetime.date]

Returns only date range for this range object. Strips out time and timezone information from the datetimes

class megforms.utils.ColorRGB(r, g, b)

Represents color consisting of Red, Green and Blue components where each component is an integer in range 0-255

Create new instance of ColorRGB(r, g, b)

megforms.utils.hex_to_rgb(hex: str) → ColorRGB

Parses colour from hex format (#FFFFFF) and returns it as a color tuple containing R, G, B components (int)

Parameters:

hex – hexadecimal representation of the colour, optionally starting with a #

Returns:

color named tuple containing each RGB component as int

class megforms.utils.ModelBackendHelper

Overrides django Model authentication backend by adding support for:

• permission validation

• case-insensitive login

megforms.utils.get_user_landing_page(user: User, next_url=None) → str

Returns url for user’s default landing page to redirect after login

class megforms.utils.DisableReadOnlyValidationFormMixin

A form mixin class that disables validation for disabled fields

megforms.utils.is_allowed_password_change(user: User) → bool

Tells whether given user is allowed to change their account password. Some shared account cannot change their password.

megforms.utils.fields_in_fieldsets(fieldsets: tuple) → Iterator[str]

Returns fields used in django admin fieldsets. For use in tests to show missing or addition fields

excluded - some fields inherited from the base class are excluded from fieldsets but need to be part of the count

megforms.utils.fields_in_specific_model(model, ignored: set[str] = frozenset({})) → list[str]

Returns fields from a model in list format.

excluded - fields inherited from BaseModel are excluded, alongside the id

megforms.utils.missing_fields_from_fieldsets(fields_from_fieldset: set[str], fields_from_model: list[str]) → set[str]

Returns fields that are missing from fieldsets in the AdminModel but present on the Model.

megforms.utils.validate_recaptcha_token(token: str | empty, platform_key: str, ip: str) → bool

Validates if user has passed recaptcha. platform_key is different for each platform: web, android and ios.

megforms.utils.make_cache_key(*args, **kwargs) → str

Given arbitrary values, generates cache key suitable for redis and django memory cache NOTE: Not suitable for memcached because key length and character set is not enforced. If you need to use cached_function with memcached, use a different key function that accounts for these limitations.

megforms.utils.cached_function(timeout=300, cache_name='default', key_fn=<function make_cache_key>)

Decorator - makes function cache-able using django cache mechanism. Notes:

• Return type of the function must be serializable/immutable object

• when applying to a class method, you must implement a custom key function (key_fn) and account for self if cached method uses any class member variables

megforms.utils.get_compliance_range_by_name(compliance_rate: Literal['incompliant', 'incompliant_and_all_comments', 'pass', 'compliant'], form: AuditForm) → tuple[float, float]

Given compliance range name (incompliant, incompliant_and_all_comments, pass, compliant) returns a range (tuple) representing its lowest compliance (inclusive) and highest compliance (exclusive).

Raises:

ValueError if compliance rate is not one of the valid values.

megforms.utils.field_with_translations(field_name: str, include_default=True) → Iterator[str]

Given a translatable field name (using django-modeltranslations lib) returns all language variants of the field name, including the default

megforms.utils.localize_time_range(start: date, end: date) → DateTimeRange

Converts dates to a datetime tuple and adds current timezone information

megforms.utils.filter_choices(choices: Sequence[Tuple[str, str] | tuple[str, Sequence[Tuple[str, str]]]], fn: Callable[[Tuple[str, str]], bool]) → Sequence[Tuple[str, str] | tuple[str, Sequence[Tuple[str, str]]]]

Given a (optionally nested) list of choices, filters the choices and returns a new set of choices filtered in the same shape, but with some items removed. Any empty groups are also removed from the list.

megforms.utils.filter_choice_values(choices: Sequence[Tuple[str, str] | tuple[str, Sequence[Tuple[str, str]]]], values: Iterable[str]) → Sequence[Tuple[str, str] | tuple[str, Sequence[Tuple[str, str]]]]

Given choices and a list of accepted values, filters out any choices whose value is not one of the given values. The input choice list can be either a flat or nested list of choices.

megforms.utils.validate_and_sort_multi_file_upload(field: Field, files: Sequence[UploadedFile], sorting: bool = True) → list[django.core.files.uploadedfile.UploadedFile]

Validates files in a multi-file field individually

Raises:

ValidationError – if any of the files does not pass validation

megforms.utils.get_subform_name(subform: Subform | SubObservation) → str

Gets name for a hardcoded or custom subform.

Note:

not to be confused with “display name”. This name is not meant to be displayed to the user.

Parameters:

subform – A subform or subobservation instance

Returns:

name of the subform

If subform is a hardcoded model, returns a model name (lowercase). For custom subforms, returns subform’s name.

Raises:

TypeError – if given object is not a subform

megforms.utils.get_subform_path(subform: Subform) → str

Similar to get_subform_name(), but for hardcoded subforms, returns dotted class path instead of model name

megforms.utils.get_set_value_placeholders_data(auditor: Auditor | None = None, ward: Ward | None = None) → dict

Get placeholder dict for “set_value” in custom field conditional logic

Parameters:

• auditor – Current Auditor

• ward – Current Ward

Returns:

dict with placeholder keys and values

megforms.utils.get_set_value_placeholders_dependant_data(wards: Iterable[Ward]) → dict

Get placeholder mapping dict for “set_value” dependant fields in custom field conditional logic

Parameters:

wards – Ward options

Returns:

dict with placeholder keys and values

megforms.utils.get_condition_set_value(session: AuditSession, field: CustomField, condition: Condition) → SerializedAnswer

Returns “set_value” value for a condition and calculates any placeholders that are used for it

Parameters:

• session – Related AuditSession

• field – CustomField

• condition – condition record

Returns:

Returns answer value after being calculated by placeholder

megforms.utils.check_edit_access(audit_form: AuditForm | None, auditor: Auditor | None) → bool | None

Checks if the user has the required level to edit the form. This function does not check user permission or whether user has access to the form.

Parameters:

• audit_form – The form to be audited.

• auditor – The user performing the audit.

Returns:

True if the user has the required level to access the form, False if the user does not have the required level, None if either audit_form or auditor is None.

megforms.utils.is_one_subform_selected(filter_form: OverviewFilterForm | None) → bool

Tells whether exactly one subform is selected in filter form. Returns True only if the form has subforms and user has selected exactly one. Returns False in every other scenario, including if filter form is null

megforms.utils.expand_aggregate_compliances(data: Iterable[dict]) → Iterator[tuple, tuple[Compliance, int]]

Parses data generated using aggregate_by_answer():

If there are any multichoice fields, the object is returned multiple times, once for each compination of answers

Parameters:

data – a sequence of dictionaries containing ‘count’, ‘compliance’ and other fields

Returns:

a list containing tuple with answers, and compliance/count tuple

megforms.utils.combine_aggregates_values(data: Iterable[tuple, tuple[Compliance, int]]) → dict[tuple, tuple[Compliance, int]]

From a stream of repeatable values assigned to compliance data (compliance, weight), builds a dictionary that maps unique data set to an average of its compliance

Parameters:

data – input data, an iterable where each item is a tuple containing a set of values, and its compliance/weight tuple

Returns:

dictionary aggregating unique values

megforms.utils.extract_url_id(request: HttpRequest, arg_name='object_id') → int | None

Extracts object’s id encoded into request url.

The function may raise an exception if value found at url does not parse into an integer.

Parameters:

• request – the incoming http request containing the object id in its request

• arg_name – name of the url argument containing the id. The default value is suitable to use with django admin.

Returns:

id extracted from the url, or none if url does not contain it.

megforms.utils.is_user_allowed_impersonate(user: User, user_cleaned_data: dict | None = None, auditor_cleaned_data: dict | None = None) → bool

Checks if the user is allowed to impersonate as per Task #29870. User must be a (global) staff member, or superuser.

Used in multiple contexts:

1. Django impersonation middleware via check_allow_impersonate()

2. Form validation in CustomUserChangeForm and AuditorAdminForm, hence optional clean data

3. Auditor model validation in clean() and save(), hence optional auditor

4. CustomUserAdmin to show / start impersonation

User attributes (is_superuser & is_staff):

1. Checks first user_cleaned_data for latest changes to user

2. If no changes checks user directly

Auditor data access (level):

1. Checks cleaned data for latest level

2. Otherwise checks user.auditor.level

3. Otherwise defaults to LEVEL_INSTITUTION

Parameters:

• user – The user requesting impersonation

• user_cleaned_data – An optional dict of new User values (from a form)

• auditor_cleaned_data – An optional dict of new Auditor values (from a form)

Returns:

True if the user is allowed to impersonate, False otherwise

megforms.utils.level_to_int(level: str) → int

given a user level, returns its integer representation so that levels can be compared

Parameters:

level – one of the valid level choices

Returns:

an integer representing the place of the level in hierarchy, starting at 0

Raises:

ValueError – if the input level is not valid

class megforms.utils.LazyTemplateValue(func: Callable, *args, **kwargs)

Allow any value or func to be lazily loaded inside template only when referenced in the template.

Lazy values can be passed with and without args and kwargs in context of a rendered template

template_html = '''
    <ul>
    {% if obj_index %}
        <li>{{obj_index}}</li>
    {% endif %}

    {% for obj in list %}
        <li>{{ obj }}</li>
    {% endfor %}
    </ul>
'''

def get_by_slug(code: str):
    return qs.filter(code=code).first()

def get_list():
    return list(qs.order_by('id'))

return render(
    request,
    template_name='template.html',
    context=dict(
        obj_index=LazyTemplateValue(get_by_slug, 'index')
        list=LazyTemplateValue(get_list)
    ),
)

Parameters:

• func – Function to evaluate when using the value

• args – function args

• kwargs – function kwargs

property value: Any

evaluate and return lazy value

megforms.utils.get_object_creator(obj: Any) → User | None

Get the user that has originally created provided object if exists

Parameters:

obj – object to be inspected

Returns:

User that has created the object

megforms.utils.safe_linebreaks(text: SafeString | str) → SafeString

Add lines breaks on html string then convert to SafeString, however if string is already a SafeString, only add lines break and keep html as is

Parameters:

text – SafeString or str of html

Returns:

megforms.utils.delete_issue_files(issue_qs: QuerySet) → None

Deletes all files associated with issues in the given queryset. Queries for all IssuePhoto, IssueDocument, and IssueAudio records associated with the issues and deletes their physical files.

Parameters:

issue_qs – QuerySet of Issue objects

megforms.utils.get_observation_ward_ringfencing_q(auditor: Auditor, *forms: AuditForm, readonly=True) → Q

Auditor ring-fence visible observations by ward for forms. Show observations where auditor is selected from a dropdown.

This method should not be used outside of implementations of for_auditor_forms. If you need to implement observation ringfencing, use for_auditor_form.

Parameters:

• forms – list of forms for the purpose of building the query

• readonly – whether the resulting query should show only items for read-only access. This can result in showing more items if user has readonly access to more observations due to ring-fencing settings.

Returns:

A Q object that can be used in an observation queryset to filter by wards

megforms.utils.update_usernames_to_saml_name_id(users_queryset: QuerySet, admin_user: User | None = None) → Dict[str, List[str]]

Updates usernames to match their SAML name_id for users with SAML identities

megforms.utils.compute_multi_ward_map(observations: Iterable['ObservationQuerySet']) → set[int]

Compute the set of session IDs whose observations span multiple wards.

For each provided ObservationQuerySet, we aggregate observations by session_id and count distinct wards. Any session that has more than one distinct ward is included in the resulting set.

Notes: - Accepts any iterable of ObservationQuerySets (generator, list, tuple, etc.). - Each queryset may refer to a different observation model; aggregation is performed per-QS.

Parameters:

observations – Iterable of ObservationQuerySets to inspect.

Returns:

Set of session IDs with more than one distinct ward among their observations.

megforms.utils.change_ward(user: User, observation: CustomObservation, target_ward: Ward, update_issues: bool, institution: Institution) → tuple[bool, Ward, Ward]

Changes the ward associated with an observation. Creates a changelog and handles edge cases like updating issues and the session’s ward. The session ward will only be updated if the observation is the only observation in the session.

Parameters:

• user – The user making the change.

• observation – The custom observation to be updated.

• target_ward – The ward to be added to the observation.

• update_issues – Whether to update issues or not.

• institution – The institution for generating the ward label in the log entry change message.

Returns:

A tuple with a boolean value representing if the ward was changed or not, the target ward and the old ward.

class utils.htmx.HTMXUtilsViewMixin

Provides view with utilities for handling htmx requests:

htmx:

provides a HTMX object for current request

htmx_redirect:

creates a HTTP response that triggers HTMX to redirect user to a url

htmx_redirect(url: str | None = None, *, preserve_arguments=True) → HttpResponse

Http response triggering htmx to refresh the page or redirect user to given url

Parameters:

• url – url to redirect to, or null to redirect back to the same page

• preserve_arguments – when redirecting to the same page, set this to False to clear any GET parameters

class utils.htmx.HTMXViewMixin

Mixin for views that redirects HTMX requests to a separate method

Important

if you’re using this mixin and PermissionRequiredMixin, in order to enforce the permission, this mixin should be listed after. For example:

class ViewWithHtmxMixin(PermissionRequiredMixin, HTMXViewMixin, View):
    permission_required = 'megforms.view_institution'

handle_htmx(request: HttpRequest, htmx: HtmxDetails, *args, **kwargs) → HttpResponse

Handle HTMX request and serve response

render_htmx_response(template_name: str, include_base_context=True, extra_context: dict | None = None) → TemplateResponse

Helper method to render template into htmx response and reuse the same kwargs as the view itself

Parameters:

• include_base_context – whether to pass view’s context into the template

• extra_context – context to pass to the template in addition to any view context

Returns:

Response containing rendered template

utils.html.strip_html_tags(html: str) → str

Strings html tags from input HTML text

Returns:

plain text from input HTML text

utils.html.append_class(classes: str, new_class: str) → str

Adds a class to HTML class property. Ensures space separation between classes.

Parameters:

• classes – string containing existing classes

• new_class – class name to add

Returns:

updated classes string

utils.diff.generate_diff(content1: str, content2: str) → Iterable[tuple[int, str]]

Generates the difference between two strings using diff_match_patch. The output can be passed to diff_to_html() to present it on a html page.

Parameters:

• content1 – First string to compare

• content2 – Second string to compare

Returns:

Difference between the two strings in form of an iterable (list) of tuples. Each tuple contains a difference information: 0 if unchanged, -1 or 1 to indicate whether the text was added or removed against content1.

Example: >>> generate_diff(“hello world”, “hello world”) [(0, ‘hello world’)] >>> generate_diff(“hello world”, “hello WORLD”) [ (0, ‘hello ‘), (-1, ‘world’), (1, ‘WORLD’) ]

utils.diff.diff_to_html(diffs: Iterable[tuple[int, str]]) → Iterable[str]

Turns diff generated from generate_diff() into a safe HTML string by replacing newlines with HTML br tags,, and individual differences into appropriate <ins/del/span> tags depending on the diff value.

Parameters:

diffs – Differences iterable

Yields:

parts of the html string, each containing a separate piece of the diff

utils.pandas.extract_json_data(dataframe: DataFrame, column: str, json_columns: tuple[str] | None = None) → DataFrame

Creates a dataframe representing columns from a json column in another dataframe. The resulting dataframe retains index from the source dataframe.

Parameters:

• dataframe – source dataframe

• column – name of the column containing json dicts

• json_columns – names of attributes in the json dicts to be used

Returns:

a new dataframe containing data from json_columns, using dataframe index

utils.pandas.expand_json_field(dataframe: DataFrame, json_field: str, *, json_columns: tuple[str] | None = None, rsuffix: str = '') → DataFrame

Takes a column containing json fields and expands it into a dataframe. Resulting dataframe is joined back into the dataframe

Parameters:

• dataframe – the df containing data and json field

• json_field – name of the column corresponding to the json field

• json_columns – keys from the json data dictionary to be extracted

• rsuffix – Suffix for overlapping custom answers column names

utils.pandas.weighted_mean(df: DataFrame, weight_column: str, group_by: str = None) → Callable[[Series], float]

Builds a callable that can be used by pandas to calculate weighted mean. Nulls and NaNs are excluded before calculating mean. Empty and null averages are represented as NaN.

Example:

df = df.agg({
    'compliance': weighted_mean(df, 'compliance_weight'),
})

Example:

df = df.groupby('ward').agg({
    'compliance': weighted_mean(df, 'compliance_weight', 'ward'),
})

Parameters:

• df – source dataframe the operation will be performed on

• weight_column – name of the df column containing weight

• group_by – column for grouping, if applied

Returns:

callable that can be passed to agg()

utils.misc.get_metadata() → dict

Get metadata about current cms instance

class utils.collections.parameteriseddefaultdict(factory: Callable[[_KT], _VT], *args, **kwargs)

like defaultdict, but take a default_factory function that runs on the missing key.

>>> squares = parameteriseddefaultdict(lambda x: x**2)
>>> squares
parameteriseddefaultdict(<function __main__.<lambda>(x)>, {})
>>> squares[2]
4
>>>squares
parameteriseddefaultdict(<function __main__.<lambda>(x)>, {2: 4})

exception utils.timeout.TimeoutError(value='Timed Out')

utils.timeout.timeout(seconds: int)

Function/method decorator that will raise TimeoutError if calling it exceeds the given execution threshold

class utils.filter_form_mixin.BaseFilterFormMixin

Base class for Mixins that provides filtering forms. Support caching Usage requires to set up filter_form_class variable to FilterForm class

get_filter_form_kwargs(**kwargs) → dict

updates kwargs with required data

get_filter_form() → Form

Returns object of initialized Form

property filter_form: Form

initialize filter_form_class with kwargs

get_queryset() → BaseQuerySet

Provides base queryset for required model

filter_objects(qs: BaseQuerySet) → BaseQuerySet

Applies filtering selected by the user to the queryset

get_context_data(**kwargs) → dict

updates context for template with form and cache_token for usage inside htmx

save_filter_state(request: HttpRequest) → str

Stores filter state in the session and returns session key

get_filter_redirect_url(request: HttpRequest, filter_state_key: str) → str

Builds url to reload this page using new filter state

utils.field_answers.get_display_value(observation: BaseAuditModel, field: Field) → str

Extracts value from observations field as display string

Parameters:

• observation – observation to fetch value from

• field – field that data we want to extract

Returns:

string value of observations field value

utils.field_answers.get_custom_value(observation: CustomObservation, custom_field: CustomField) → str

Extracts value from observations field as display string. for CustomFields

Parameters:

• observation – observation to fetch value from

• custom_field – field that data we want to extract

Returns:

string value of observations field value

class utils.pdf.WeasyPrintFileResultDict

utils.pdf.custom_url_fetcher(url: str, *args, **kwargs) → WeasyPrintFileResultDict

Custom URL fetcher for WeasyPrint that handles image paths consistently across Docker and local environments.

For relative media URLs, this fetcher handles both path conversion and file reading to avoid errors with WeasyPrint’s django_url_fetcher path parsing.

Parameters:

• url – The URL to fetch, typically beginning with ‘file://’ followed by a media URL

• args – Additional positional arguments passed to django_url_fetcher

• kwargs – Additional keyword arguments passed to django_url_fetcher

Returns:

Dictionary containing file data in WeasyPrint’s format (string, mime_type, encoding, filename, redirected_url, file_obj)

Path handling rules:

• Detects relative media URLs that WeasyPrint converts to file:// format

• Converts these to absolute filesystem paths using MEDIA_ROOT

• Reads file data directly instead of relying on URL fetcher path parsing

Example:

Original relative - ‘/media/images/photo.jpg’ WeasyPrint conversion - ‘file:///media/images/photo.jpg’ Fetcher conversion (Windows) - ‘C:/Users/user/Documents/meg-qms/media/images/photo.jpg’ Fetcher conversion (Linux) - ‘/opt/meg-qms/media/images/photo.jpg’

utils.pdf.html_to_pdf(html: str, output_file: Path | IO, *, stylesheet: str | Iterable[str] = ())

Converts HTML string to a PDF file, works with custom fonts implemented in ‘megdocs/static/css/fonts.css’

Parameters:

• html – HTML string

• output_file – Path to PDF file, or a file-like object

• stylesheet – absolute path to css file(s) used to style the output PDF

Example usage using a temporary file:

with tempfile.NamedTemporaryFile(prefix='html_conversion', suffix='.pdf') as pdf_file:
    html_to_pdf(draft.content, pdf_file, os.path.join(settings.ROOT_DIR, 'meg_forms/megdocs/static/css/document.css'))

    # Save pdf to django model
    object = ExampleModel()
    object.pdf_document.save("filename.pdf", pdf_file)

utils.pdf.pdf_to_txt(pdf: Path, **kwargs) → str

Extracts text from a PDF file Uses PDF Miner library to extract text from a PDF file.

Parameters:

• pdf – path to the PDF file

• kwargs – any additional kwargs for pdfminer

megforms.timezone_utils.get_database_tznames() → frozenset[str]

Queries database for all supported timezone names

Unpublishing & restoring

utils.unpublish.get_child_models(parent: type[megforms.base_model.BaseModel]) → Iterator[tuple[type[megforms.base_model.BaseModel], str]]

looks for models that have a fk relation to given parent model and yields them

Parameters:

parent – parent model class

Yield:

tuples containing model class, field name of the fk with relation to the parent, and the on_delete value of the foreign key

class utils.unpublish.Unpublisher(objs: QuerySet | BaseModel, user: User | None = None, force: bool = False)

Collects objects for un-publishing by visiting relations and storing object ids. The collected objects can be unpublished in bulk by invoking its unpublish() method.

Unpublishing objects results in creation of a UnpublishLogEntry object that serves as a log of objects affected by the action and can be used to re_publish() the affected objecs.

Note

this class operates exclusively on models that extend BaseModel. It is not suitable for django built-in models such as User, Group or models from third-party libraries as they do not have a publish field

Optionally user object can be designated for logging purposes.

Important

this class does not verify that user has the permission to unpublish objects, or that the user even has access to those objects or its children.

Example usage:

unpublisher = Unpublisher(Room.objects.filter(name='Test Room'))
if unpublisher:
    print(unpublisher.as_strings())
    unpublisher.unpublish()

Parameters:

• objs – objects to be unpublished. Can be single instance, or a queryset.

• user – optional user if action is being performed by a user rather than automated job.

• force – whether to include already unpublished objects when collecting

objs: dict[type[megforms.base_model.BaseModel], set[int]]

maps model to a set of ids of objects to be unpublished

user: User | None

user performing the action

force: bool

whether to include already unpublished objects when collecting

as_strings() → Iterable[tuple[str, Iterable[str]]]

Represents collected objects as a structure with model name (plural) to object names. Note that model names are localized into a non-lazy string. For example:

[
   ("Wards", ["Ward 1", "Ward 2"]),
   ("Rooms", ["Room 101", "Room 102", "Room 201"]),
]

collect_child_objects(parent_model: type[megforms.base_model.BaseModel], parent_ids: Iterable[int])

Collects child objects for parent objects in given model

collect_generic_relations(parent_model: type[megforms.base_model.BaseModel], parent_ids: Iterable[int])

Collects objects with a GenericRelation to given model instances.

The implementation hardcodes some rules as to which models will have a generic relation and may need to be updated if comments or QIP functionality is expanded to other models.

Parameters:

• parent_model – parent model class

• parent_ids – ids of parent objects being unpublished

collect_objs(objs: QuerySet)

Adds provided objects and their dependants to the collection.

Only published objects are

unpublish(*, log_action=False, log_comment: str | None = None) → int

Unpublish the collected items atomically, and logs an unpublish entry by creating a UnpublishLogEntry.

Optionally can also log user’s action to LogEntry.

Parameters:

• log_action – Whether action should be logged to LogEntry (requires that user is set)

• log_comment – optional comment to be added to log entry if log_action is set

Returns:

total number of unpublished objects

unpublish_logged(comment: str | None = None) → int

Un-publishes the object(s) and logs the action with a comment to Django’s LogEntry with action_flag=CHANGE.

Parameters:

comment – comment to be added to the action log

Returns:

number of affected objects

utils.unpublish.unpublish(user: User | None, objs: BaseQuerySet | BaseModel, log_comment: str | None = None) → int

Un-publishes object and cascades changes to its child objects.

Parameters:

• user – Usert whose to whom the action is attributed (logged). If passed, the action will be logged and reversible.

• objs – object(s) to be unpublished. Can be already un-published

• log_comment – comment to be added to log entry - requires that user is also passed

Returns:

Total number of updated objects, including child objects. Object already unpublished are included in this count.

Logging & Metrics

logger.get_logger() → Logger

Gets singleton logger instance

logger.log_debug(message: str, *args)

Log a debug message

logger.get_tracer() → Tracer

Gets OpenTelemetry Tracer instance

logger.trace_span(name: str, span_attrs: Mapping[str, str | bool | int | float | Sequence[str] | Sequence[bool] | Sequence[int] | Sequence[float]] | None = None) → Span

Helper context manager to trace code within the span

Parameters:

• name – trace span name

• span_attrs – any custom values to be included in trace. A Dictionary mapping string key to a value (str, int, bool, float, a Sequence or None)

Example:

>>> with trace_span('some operation') as span:
>>>     do_something()

logger.trace_decorator(name: str | None = None, *, trace_args: bool = False)

Decorator used to trace entire function

Parameters:

• name – Optional trace span name. Defaults to function path.

• trace_args – Wheter args and kwargs from the function calls should be included in trace WARNING: do not trace args in functions that can accept sensitive information or are expensive to compute

Usage:

>>> @trace_decorator(name='do something')
>>> def some_function():
>>>     do_something()

Can also be used without arguments

>>> @trace_decorator
>>> def some_function():
>>>     do_something()

class logger.catch_exception(*exceptions: type[Exception])

Catches exception within the context and reports them to Sentry and logger Similar to contextlib.suppress, but instead of completely ignoring the exception, it reports the exception to Sentry.

Use only if the piece of code within the context is allowed to fail without side effects, but you still want to track the exceptions.

Examples:

with catch_exception():
    ...
with catch_exception(TypeError, KeyError):
    ...

Equivalent to:

try:
    ...
except Exception:
    sentry_sdk.capture_exception(exc_value)

Parameters:

exceptions – types of exceptions to catch. Leave blank to catch all exceptions

class regions.metrics.RegionInsightsReporter

Reports server metrics to Azure Insights. All metrics are tagged with current site domain.

IMPORTANT:

Do not use this class if INSIGHTS_CONNECTION_STRING is not set

More information in Task #27621

log_metric_value(metric: Gauge, value: int | float | Callable[[], int | float], *, sample_rate: float = 0.01)

Report metric value to Azure Insights. Value can be callable, it will be evaluated only if INSIGHTS_SAMPLE_RATE determines that the value should be reported.

Parameters:

• metric – the metric being reported

• value – the value to be reported, or a callable returning the value (will be evaluated once)

• sample_rate – rate at which this metric should be sampled. Default value follows the global setting, but can be overridden to ensure that the metric is reported this time.

log_celery_queue_size(size: int | Callable[[], int])

Reports current celery task queue size

Celery

megforms.celery_utils.get_celery_task_path(task: str | Task) → str

Given a celery task reference, returns its path as a string. Does nothing if passed task is already a str.

Parameters:

task – task reference (class, function), or path

Returns:

task path

megforms.celery_utils.schedule_task(name: str, start_time: datetime | timedelta, *, task: str | Task, task_args: list | dict | None = None, overwrite=False, **kwargs) → PeriodicTask

Schedule a once-off celery task to be executed at specified time by creating a PeriodicTask object.

Parameters:

• name – unique name for this schedule

• task – the celery job to trigger at specified time

• start_time – time when job should be triggered. Either time interval (from now) or absolute datetime

• task_task_args – args or kwargs that will be passed to the task on execution

• overwrite – if task with this name already exists, delete it before scheduling new task. If set for False and task already exists the function will raise an error

Params kwargs:

Anny additional arguments for PeriodicTask model

megforms.celery_utils.schedule_repeating_task(name: str, start_time: datetime, interval: timedelta | relativedelta | IntervalSchedule, *, task: str | Task, task_args: list | dict | None = None, **kwargs) → tuple[django_celery_beat.models.PeriodicTask, bool]

Schedules a celery task to run under given name. Updates the schedule if name already exists.

Parameters:

• name – Unique name of this schedule

• start_time – First time when the schedule should be triggered

• interval – IntervalSchedule or timedelta or relativedelta object representing the repeat frequency. For once-off schedules, use timezone.timedelta(0).

• task – The task to be ran. Pass task’s path (string) or its reference (function or class). The value will be parsed using get_celery_task_path().

• task_args – Optional args (list) or kwargs (dict) to be passed into the task when executing.

• kwargs – Any additional arguments for PeriodicTask constructor

Returns:

tuple containing the PeriodicTask and bool whether it was created (True), or updated (False)

Date & Time utils

megforms.calendar.utils.iter_dates(start: date, end: date | None = None, step: timedelta | relativedelta = datetime.timedelta(days=1))

Iterates date range analogous to Python’s range()

Parameters:

• start – the first date to be yielded

• end – the last date to be yielded, or None to iterate indefinitely

• step – increment between dates; defaults to 1 day to yield each date, can be changed to skip days

Yield:

each date from start to end inclusive

megforms.calendar.utils.year_quarters(*years: int) → Iterable[Tuple[date, date]]

Generates first and last day of each quarter in the given year(s)

Parameters:

years – any number of years

Yield:

four date pair are yielded for each year passed in years, prepresenting each quarter of each year.

megforms.calendar.utils.get_quarter(date: datetime | date) → tuple[int, Tuple[datetime.date, datetime.date]]

Gets quarter number and its start/end dates for given date or time

Parameters:

date – the date or datetime. If datetime is passed, it is converted to local time first

Returns:

tuple containing the quarter number (0-3), and its first and last day

megforms.calendar.utils.get_year_half(date: datetime | date) → date

Returns a date representing the start date of the year half.

Parameters:

date – the date or datetime. If datetime is passed, it is converted to local time first

megforms.calendar.utils.get_quarter_string(date: datetime | date) → str

Gets string representation of quarter the given date belongs to

Parameters:

date – date or datetime

Returns:

Quarter string representation, for example “Q3 2021”

Example:

>>> get_quarter_string(datetime.date(2021, 6, 15))
'Q2 2021'

megforms.calendar.utils.get_half_year_string(date: datetime | date) → str

Gets string representation of the year half the given date belongs to

Parameters:

date – date or datetime

Returns:

Half year string representation, for example “H1 2023”

Example:

>>> get_half_year_string(datetime.date(2023, 1, 1))
'H1 2023'

megforms.calendar.utils.get_month_pair(date: datetime | date) → date

For month pairs jan/feb, mar/apr, for a given date, returns a date representing the start date of the month pair (first day of first month in the pair with the same year as the original date)

megforms.calendar.utils.generate_last_four_quarters_ranges(today: date, translatable: bool = False) → Iterable[Tuple[str, Tuple[date, date]]]

Generate predefined filters for the last four quarters relative to the given date.

Parameters:

• today – Today’s date

• translatable – Whether quarters are translatable or raw

Returns:

Iterable of (quarter label, tuple of quarter start date, quarter end date) tuples

megforms.calendar.utils.generate_date_range_choices(today: date, include_future: bool = False) → Iterable[Tuple[str, Tuple[date, date]]]

Generates various date ranges for filter dropdown presets

megforms.calendar.utils.date_range_choices(today: date, include_future=False) → Tuple[str, Tuple[str, str]]

Date range choices for given date, cached by date

Parameters:

• today – Today’s date

• include_future – Include future dates

Returns:

megforms.calendar.utils.number_passed_leap_years_for_range(start_date: date, end_date: date) → int

Calculates the number of leap years between two dates with one caveat. The number is the number of years that are leap and Feb 29th is within the range. If the start year is leap, and we passed Feb, then this year won’t be count in the range as leap year. Similarly, if the end year is leap, and Feb 29th is still not hit, then this year won’t be count in the range as leap year.’

class megforms.calendar.utils.TimelineEventTuple(name: str, date: date, color: str)

A simple tuple version of Timeline event, which allows its usage without object / db creation

Create new instance of TimelineEventTuple(name, date, color)

name: str

Alias for field number 0

date: date

Alias for field number 1

color: str

Alias for field number 2

megforms.time_tracking_utils.annotate_times(qs: ObservationQueryset, time_object: dict[str, Union[str, int, bool, list, float, NoneType]], annotation_name: str) → ObservationQueryset

Annotates queryset with a time value. Uses LogEntry and TrackerLogEntry table to extract the time when action was performed.

Example:

observations = CustomObservation.objects.all()
# Annotate start_time with a value of when the observation was created
observations = annotate_times(observations, {"time_created": ""}, 'start_time')
# Annotate review_time with the time when int_field was set to 20 for the first time.
observations = annotate_times(observations, {"int_field": 20}, 'review_time')

Parameters:

• qs – an observation queryset to be annotated

• time_object – a dict representing field or action (key) and optionally a value to filter by time given field was reviewed

• annotation_name – name to bind the value to in the queryset

Returns:

the same queryset, but with added annotation

megforms.time_tracking_utils.timedelta_to_string(timedelta: timedelta, verbose: bool = False) → str

Converts a timedelta object to a human-readable string.

Parameters:

timedelta – The timedelta object to convert.

Param:

verbose: when set to false, limits output to two time units, for example “1 year, 2 months”

Returns:

The human-readable time string.

megforms.time_tracking_utils.seconds_to_time_string(seconds: int, verbose: bool = False) → str

Converts number of seconds to a human-readable string. Negative delta is converted to positive. verbose - when set to false, limits output to two time units, for example “1 year, 2 months”

Examples:

>>> seconds_to_time_string(20000)
'5 hours, 33 minutes'
>>> seconds_to_time_string(200000, verbose=True)
'2 days, 7 hours, 33 minutes'
>>> seconds_to_time_string(200000, verbose=False)
'2 days, 7 hours'

megforms.time_tracking_utils.find_start_month(month: int, months: list[int]) → int

Get first allowed month for given months

Parameters:

• month – month

• months – Allowed month

Returns:

first allowed month

megforms.time_tracking_utils.months_between_two_dates(start_date: date, end_date: date, months: list[int] | None = None) → Iterator[date]

Get a list of datetime for each month between start and end dates

Parameters:

• start_date – Start Date

• end_date – End Date

• months – allowed months array (TWO_MONTHS, QUARTER_MONTHS)

Returns:

list of datetime for months

megforms.time_tracking_utils.get_first_day_of_quarter(time: datetime) → datetime

Helper method to get the first day of any given quarter.

Parameters:

time – a datetime used to find the quarter.

Returns:

first day of the quarter.

megforms.time_tracking_utils.get_datetime_of_last_day_of_last_month() → datetime

Shorthand for finding the last day of the previous month

Accounts

accounts.utils.get_user_institution(user: User) → Institution | None

Returns user’s institution. For auditors it simply returns their institution instance. For eGuide users, returns institution of the first eGuide they have access to. Returns None for users who are not auditors and don’t have access to any eGuides.

accounts.utils.get_user_institutions(user: User) → InstitutionQuerySet

Works like accounts.utils.get_user_institution() but returns multiple institutions for higher-level users like group auditors

accounts.utils.get_user_language(user: User) → str | None

Determines user’s language based on user’s preference. If user does not have a preference, return user institution’s language.

If null is returned, it is recommended to use server’s default language.

Parameters:

user – the user instance

Returns:

language code or null or language cannot be determined by user’s preference or institution setting.

accounts.utils.must_pass_captcha(user: User | RegionUserLink) → bool

Determines if the auditor must pass a captcha test on their next login. This is based on the institution’s password policy.

accounts.utils.get_is_auth_blocked(region_user_link: RegionUserLink, auth_restriction_type: str, request: HttpRequest | None = None) → bool

Checks whether account in another region is restricted from logging in to the account. Makes a request to user’s region to make the check

Parameters:

• region_user_link – the user account in another region

• auth_restriction_type – type of restriction to check for: AUTH_RESTRICTION_TYPE_BLOCKED or AUTH_RESTRICTION_TYPE_CAPTCHA

• request – current HTTP request

Returns:

True if the requested restriction is applied to the account (account blocked or captcha)

accounts.utils.check_login_blocked(user: User | RegionUserLink, request: HttpRequest | None = None) → bool

Determines if a user is blocked from logging in. If there hasn’t been a successful login within the limit, block user login until policy timeout has passed. Blocks the user if they are a global account that shouldn’t be a global account.

accounts.utils.build_app_deep_link(request: HttpRequest | None, auditor: Auditor | None, institution_id: int | None = None, form_id: int | None = None) → str

Builds pre-authenticated deep link to the client web app for the given institution and form id

accounts.utils.get_unique_username_for_usernames(username: str, usernames: Iterable[str], append: int = 1) → str

Raises:

ValidationError if given username is not a valid username as per django username rules

accounts.utils.generate_locally_unique_usernames(usernames: dict[str, str]) → dict[str, str]

Checks if usernames are unique locally. For any non-unique usernames, ints are appended to achieve uniqueness. Maintains a map between generated username and the original username.

accounts.utils.is_super_or_global(user: User) → bool

Checks if a user has a high level of access to data in MEG.

accounts.utils.is_super_global_or_staff(user: User) → bool

Checks if a user has a high level of access to data in MEG, including staff users.

accounts.utils.is_super_or_global_allowed(institution: Institution, email: str) → bool

Checks if a user with the provided properties is allowed to be a super or global user.

accounts.utils.is_two_factor_user(user: User) → bool

Determines if the user has two factor enabled.

accounts.utils.is_2fa_required(user: User) → bool

Determines whether 2FA is required for specified user account

accounts.utils.is_2fa_enabled(user: User) → bool

Determines whether 2FA is enabled for specified user account

accounts.utils.send_password_reset(request: HttpRequest, user: User, subject_template_name='registration/password_reset_subject.txt', email_template_name='password_reset_email')

Sends password reset e-mail to selected user if they have an e-mail address. Does nothing if there’s no e-mail address associated with the account.

accounts.utils.verify_ip(user_ip: str, auditor: Auditor) → bool

Verifies given IP address against whitelist associated with user account.

Parameters:

• user_ip – an IP address in string format (e.g. 192.168.1.1)

• auditor – user profile instance

Returns:

False if the ip address is not whitelisted. True if address is whitelisted, or there is no whitelist

accounts.utils.verify_request_ip(request: HttpRequest, auditor: Auditor) → bool

Calls verify_ip() using IP address from the request to verify whether the IP address is valid (allowed by whitelist) for this account.

Parameters:

• request – current HTTP request

• auditor – relevant user profile

Returns:

True if IP address is allowed

accounts.utils.password_complexity_is_valid(user: User, password: str) → bool

Validates password against the user’s current policy.

accounts.utils.can_user_set_password(user: User) → bool

whether user account is allowed to have a password. Some IDPs may prevent user from setting a password if user has logged in using SAML protocol.

accounts.utils.get_users_who_can_reset_password(**filters) → QuerySet

Gets users who can set their password. This excludes public accounts, accounts managed by some SAML IDPs etc

accounts.utils.get_user_for_username_case_insensitive(username: str, user_model: type[django.contrib.auth.models.User] | type[regions.models.RegionUserLink] = <class 'django.contrib.auth.models.User'>, select_related: ~typing.Iterable[str] | None = None) → User | RegionUserLink | None

Gets a user by username disregarding the case for a specified model. If there are case-insensitive duplicate usernames, a case-sensitive match is required.

Parameters:

• username – the username to find a user by.

• user_model – the user model to query.

• select_related – related field names to select using QuerySet.select_related.

Returns:

User, RegionUserLink or None

accounts.utils.string_is_email(string: str) → bool

Shorthand method for checking if a string matches the format of an email address. This doesn’t validate if the email address is a real email address.

Returns:

boolean if the string looks like an email address.

accounts.utils.login_as_demo(request: HttpRequest, demo_user: User, original_user: User | None = None, return_url: str | None = None)

Login as a demo account in a new session

accounts.utils.logout_from_demo(request: HttpRequest) → str

Logout from demo account and log back into original user

Parameters:

request –

Returns:

redirect url after login

accounts.utils.get_logged_in_as_demo(request: HttpRequest) → bool

Checks if current request is a logged in demo account

accounts.utils.allow_demo_login(request: HttpRequest) → bool

Demo Accounts function that returns True if this request is allowed to log in as demo account

accounts.utils.get_demo_accounts_queryset(request: HttpRequest) → AuditorQueryset

Demo Accounts function accessible demo users from the passed request

Action logs

action_audit.utils.iter_choices(choices: Sequence[Tuple[str, str] | tuple[str, Sequence[Tuple[str, str]]]]) → Iterator[Tuple[str, str]]

iterate nested/grouped choices as if it were a flat list

action_audit.utils.get_log_entry_objects_dict(qs: QuerySet) → dict[django.contrib.admin.models.LogEntry, Iterable[tuple[int, django.db.models.base.Model]]]

Dictionary mapping log entries to a set of objects referenced by them. Deleted objects will not yield a model instance

class action_audit.utils.FormChangesMixin

Supplies get_form_changes method to a view allowing it to extract data for a changelog

get_object_changes(observation: CustomObservation, old_observation: CustomObservation, changed_fields: Sequence[str] = None) → Iterator[str]

Iterates over changes from changes between old and new data of the observation for certain fields

Parameters:

• observation – current version of observation

• old_observation – old observation

• changed_fields –

Yield:

string representing each difference between the old and current objects custom field values

get_form_changes(form: BaseForm, ignore_fields: tuple[str, ...] = ()) → Iterator[str]

Iterates over changes in the form and yields string representing

Parameters:

• form – a django form

• ignore_fields – excludes fields from change message.

Yield:

string representing each change made by user in the form

get_field_diff(field: Field, initial: str | int | bool | list | float | date | time | datetime | None | Model | ObservationFile, value: str | int | bool | list | float | date | time | datetime | None | Model | ObservationFile) → str

Get the difference between initial and new value as str, for native values (str, int, float) python str() repr will be used, for choice values the label will be used as the repr, for list values comma separated array repr will be used

class action_audit.utils.BaseActionAuditMixin

Adds _log method to a view that logs user action

log_observation_change(request, obj: CustomObservation, old_obj: CustomObservation, fields: Sequence[str] = ()) → LogEntry | None

Logs change action for observation

Returns:

log entry

action_audit.utils.log_observation_change(user: User, obj: CustomObservation | CustomSubObservationInstance, old_obj: CustomObservation | CustomSubObservationInstance, fields: Sequence[str] = (), change_message: str = '') → LogEntry

Logs change action for observation. Intended for use in places where a request object isn’t available, such as a celery task.

Parameters:

• user – The user who made the change.

• obj – The observation whose change is being logged.

• old_obj – The old state of the observation before the changes.

• fields – The fields that were changed.

• change_message – A change message to attach to the log entry.

class action_audit.utils.BaseActionAuditViewMixin

Mixin for views that adds action logging methods

log_action(action: int, obj: Model | Sequence[Model], field_names: Sequence[str] = (), comment: str = '')

Logs an arbitrary action

log_view(obj: Model, comment: str = '')

Logs a view action of a single model instance

log_view_list(objs: Sequence[Model], comment: str = '')

Logs a view action of multiple model instances

log_change(obj: Model | Sequence[Model], field_names: Sequence[str] = (), comment: str = '')

Logs change to a model instance

log_creation(obj: Model | Sequence[Model], field_names: Sequence[str] = (), comment: str = '')

Logs creation of a model instance

log_deletion(obj: Model | Sequence[Model], comment: str = '')

Log deletion of a model instance. Can be used to log that object was marked as unpublished

class action_audit.utils.ActionAuditFormMixin

Mixin class for model form views (CreateView, ChangeView) which logs action when form is saved. Also supports form views with formset (multiple forms)

audit_change_message = 'Updated in Settings'

Change message for when object was edited

audit_create_message = 'Created in Settings'

Change message for when object was created

audit.utils.log_action(user: User, obj: Model, message: str | list, action_flag: int = 2) → LogEntry

Shortcut to creata a LogEntry instance

Parameters:

• user – The user performing hte action

• obj – the actioned object

• message – action description or a list of descriptions (list will be convertedto a json list before saving)

• action_flag – the action being taken: ADDITION, CHANGE, DELETION, or any of the custom flags

Returns:

Saved LogEntry instance

audit.utils.log_bulk_action(user: User, objs: QuerySet, message: str | list, action_flag: int = 2)

Shortcut to bulk create LogEntry instances

Parameters:

• user – The user performing hte action

• objs – the actioned objects

• message – action description or a list of descriptions (list will be converted to a json list before saving)

• action_flag – the action being taken: ADDITION, CHANGE, DELETION, or any of the custom flags

audit.utils.annotate_impersonation_reason(qs: QuerySet) → QuerySet

Annotates impersonation log queryset with reason field containing the reason for impersonation. The reason is extracted from LogEntry model within 1 minute of impersonation.

Returns:

a queryset whose objects have a new reason text field

Analytics

analytics.utils.log_event_lazy(obj: BaseModel | tuple[django.contrib.contenttypes.models.ContentType, int], request: HttpRequest | None = None, platforms: Sequence[str] = (), **kwargs)

Delegate event logging to a Celery job

analytics.utils.get_unique_tags() → set[str]

Unique set of tags in all analytics entries

API

api.utils.create_subobservation_serializers(model_class: Type[BaseAuditModel], audit_form: AuditForm, first_submission=True) → Iterator[Tuple[str, Serializer]]

Creates serializers for model’s subobservations yields tuple of name and serializer instance

api.utils.create_serializer(observation_model: Type[BaseAuditModel] | CustomSubform | Type[BaseSubObservation], audit_form: AuditForm | None = None, viewset: bool = False, first_submission: bool = True) → Type[Serializer]

Creates serializer class for specified model class model_class can be either class of the observaiton model or a custom subobservation instance viewset determines if the serializer is to be used as a viewset’s serializer_class

api.utils.create_viewset(model_class) → Type[BaseObservationViewSet]

Creates ViewSet and serializer for specified model class

class api.utils.SerializerValidatorAdapter

Helper object that holds a placeholder for serializer instance and its answers for the validator so that validator can be created before data is passed into the serializer.

class api.utils.UnpublishViewSetMixin

Mixin for DRF ViewSets implementing DestroyModelMixin that replaces delete action with unpublish.

class api.utils.ReversionAPIMixin

Mixin for views extending ModelViewSet that logs creation/update/deletion actions using the standard DRF endpoints.

This mixin should be included before any mixins that override the perform_create/update/destroy method

eGuides

eguides.utils.import_section_text(section: Section, directory, filename) → str

Imports section content from html file and creates related section images

eguides.utils.create_eguide_dict(eguide: Eguide) → dict

Creates a dict of eGuide content

eguides.utils.create_eguide_json_file(eguide: Eguide) → File

Creates a json file object containing a snapshot of eGuide content

eguides.utils.check_valid_user(user: User, eguide: Eguide) → bool

validates that given user has access to the eGuide

eguides.utils.flatten_sections(sections: Iterable[dict]) → Iterable[SectionDict]

flattens a list of nested sections

eguides.utils.download_file(link: str) → tuple[str, django.core.files.base.File]

Helper method to download a file and prepare for saving to a django model.

Parameters:

link – The file link to be downloaded.

E-mails

(celery task)emails.utils._send_mail(subject: str, body_text: str, addr_from: str | None, addr_to: str | Iterable[str], fail_silently=False, attachments=None, body_html=None, connection=None, headers=None, locale=None, bcc: str | Iterable[str] | NoneType = None, reply_to=None)

Sends a multipart email containing text and html versions which are encrypted for each recipient that has a valid gpg key installed.

emails.utils.send_mail_template(subject: str, template: str, addr_from: str, addr_to: str | Iterable[str], institution: Institution = None, fail_silently=False, attachments=None, context=None, connection=None, headers=None, locale=None, bcc=None, reply_to=None)

Send email rendering text and html versions for the specified template name using the context dictionary passed in.

emails.utils.send_simple_mail(addr_to: str | Iterable[str], subject: str, message: str | Iterable[str], *, bcc=None, addr_from='MEG <noreply@megit.com>', locale: str | None = None, links: Iterable[tuple[str, str]] = (), institution: Institution | None = None, template: str = 'default')

Uses a generic template to send the e-mail message in a plain text as well as html format. message: string message, or a list of lines of the message Links: add hyperlinks to the e-mail, will be rendered under the content

emails.utils.strip_message_thread(message: str) → str

Given an e-mail message thread including replies, extracts only the latest message (reply)

emails.utils.resolve_filter_obj(filter_obj: dict, custom_fields: CustomFieldQuerySet, flat_form: bool | None = None) → Q

Builds a django Q object representation of JSON answer filters.

Parameters:

• filter_obj – The object defining the filters.

• custom_fields – A queryset of custom fields which can be used in filter logic.

• flat_form – Whether to return a Q object for filtering just custom observations, sub observations or both.

emails.utils.build_q_object(filter_obj: list | dict, custom_fields: CustomFieldQuerySet, operator: str | None = 'and', flat_form: bool | None = None) → Q

Builds a django Q object representation of nested JSON answer filters.

Parameters:

• filter_obj – The object defining the filters.

• custom_fields – A queryset of custom fields which can be used in filter logic.

• operator – Operator for handling a list of filters.

• flat_form – Whether to return a Q object for filtering just custom observations, sub observations or both.

emails.utils.send_report_rule_sms_to_recipient_fields(rule: ReportRule, observations_iter: Iterable[CustomObservation], sms_content: str) → None

Sends sms content to phone number recipients defined in observation answer data.

Parameters:

• rule – The report rule defining recipient fields.

• observations_iter – An iterable of observations.

• sms_content – The sms text content to send.

Single sign-on

external_auth.utils.parse_saml_value(value: list | str | None) → str | None

Reads SAML value (normally stored as list). Handles empty list

external_auth.utils.get_or_create_user(idp: SamlIdentityProvider, saml_auth: OneLogin_Saml2_Auth) → User

Gets user account (or creates a new one) and updates their role based on provided SAML object

external_auth.utils.clean_username(username: str, idp: LDAPIdentityProvider) → str

Validates that username can be used

Parameters:

• username – the username

• idp – the relevant Identity Provider; used to append unique post-fix

Returns:

username if valid (unique), or validation error

external_auth.utils.parse_user_attr(user_attributes: dict, key: str) → str | None

attribute values are a list, parse first item if exists.

external_auth.utils.set_user_attr(user: User, auditor: Auditor, field: str, value: Any) → None

Shorthand method for updating either user or auditor. Adds a log message explaining how the object was updated. This method does not save either object. If updating a charfield and the string is longer than max length, it’s sliced to fit.

Parameters:

• user – the User.

• auditor – the Auditor.

• field – The field to update. The prefix ‘auditor.’ means it’s an auditor field.

• value – The value to set.

external_auth.utils.set_ldap_user_attributes(user: User, auditor: Auditor, user_attributes: dict, attribute_map: dict) → None

Updates a user model based on mapped LDAP remote user data. This method can update the user, or auditor, but doesn’t save either.

Parameters:

• user – the User.

• auditor – the Auditor.

• user_attributes – a dict of user data from LDAP server.

• attribute_map – a dict which maps LDAP attributes to user model properties.

external_auth.utils.set_user_attr_from_map(user: User, auditor: Auditor, field: dict, value: Any, idp: SamlIdentityProvider) → None

Shorthand method for updating either user or auditor from attribute map defined values. Adds a log message explaining how the object was updated. This method does not save either object. If updating a charfield and the string is longer than max length, it’s sliced to fit.

Parameters:

• user – the User.

• auditor – the Auditor.

• field – A dict describing the field and how it should be updated.

• value – The value to set.

• idp – The SAML identity provider.

external_auth.utils.get_user_attr_teams(field: dict, value: Any, idp: SamlIdentityProvider) → set[megforms.models.Team]

Gets the teams a user should have based on the attribute map

Parameters:

• field – A dict describing the field and how it should be updated.

• value – The value to set.

• idp – The SAML identity provider.

external_auth.utils.get_user_attr_patient_data_access(field: dict[str, Any], user_groups: list[str], sync_permissions: str) → bool | None

Checks whether a user should have patient data access based on attribute map values. To understand when each boolean is returned and when none is returned please check: #32510 note #28 and #37

Parameters:

• field – The field which may contain the group mapping.

• user_groups – groups assigned to the user.

• sync_permissions – identity provider sync settings overwrite/add/disabled.

Returns:

True when user belongs to any group that grants access, False when SAML should revoke access, or None when the mapping does not apply.

external_auth.utils.update_or_create_scim_user(idp: SamlIdentityProvider, name_id: str, scim_id: str | None, email: str | None, user_kwargs: dict, auditor_kwargs: dict, user: User | None = None) → tuple['SamlUser', 'Auditor']

Get or create a user and auditor from SCIM data. This function encapsulates the logic for finding an existing user (by scim_id, name_id, or email) or creating a new one if no existing user is found.

Parameters:

• idp – The saml identity provider.

• name_id – The SAML username of the user.

• scim_id – The SCIM identifier of the user.

• email – The email address of the user. Used to link SCIM accounts with existing accounts.

• user_kwargs – Keyword arguments used to create / update the user.

• auditor_kwargs – Keyword arguments used to create / update the auditor.

• user – The user that will be updated. This will be None when a new SCIM account is being provisioned.

Files

class files.utils.FileReuploadFormMixin

A mixin for forms containing a file field prompting user to re-add files if validation fails as files are cleared by the browser when page reloads. As per Task #25741, form cannot retain files from previous submission, so the user needs to re-attach the files again.

property errors

Return an ErrorDict for the data provided for the form.

class files.imagekit.ImagekitCacheStrategy

Solution to files being closed after saving. Copied from: https://github.com/matthewwithanm/django-imagekit/issues/391#issuecomment-283052935

class files.imagekit.CacheFileBackend

An extension of Simple backend, enables local customization.

files.imagekit.source_name_as_obfuscated_path(generator)

A legacy file namer that stores thumbnails in a directory per-file, but obfuscates original file name by hashing it

Messaging

messaging.utils.get_valid_recipients(sender: User, qs: QuerySet = None, include_sender=False, active: bool = True) → QuerySet

Given a sender user, returns a queryset of potential recipients they can send messages to.

Parameters:

• sender – the user

• qs – Optional queryset. If not given, all users will be considered

• include_sender – whether sender should also be considered, or excluded from potential recipients

• active – whether to show deactivated or active accounts.

Returns:

filtered qs or queryset of users sender can send messages to

Push messages

push_messages.utils.send_user_push_message(user: User, message: str, title='MEG') → PushResponse | None

Sends simple push notification to specified user’s device. The user is targeted by their id and username (using NamedUser).

User must be logged in to the app to receive messages.

Parameters:

• user – the recipient

• message – the message body

• title – push notification title

Returns:

UA push response instance

QIP

qip.utils.qip_field_choices() → Sequence[Tuple[str, str]]

Builds a list of QIP fields that can be included/excluded in QIP preferences

qip.utils.qip_app_field_choices() → Sequence[Tuple[str, str]]

Builds a list of QIP fields that can be included/excluded in QIP preferences Removes choices that are not relevant to the client app

qip.utils.can_edit_issue(issue: Issue, auditor: Auditor) → bool

Tells whether auditor has permission to edit given Issue. It is assumed that user already has read permission, so audit form access or institution membership is not verified. Returns True if user can edit the permission. False is returned if user has only read access.

qip.utils.is_status_disabled_for_auditor(issue: Issue, auditor: Auditor) → bool

Tells whether status field is disabled for given Issue and Auditor. Checks auditor’s form-level permissions against audit form config.

Regions

regions.utils.is_upstream_region(request: Request | None) → bool

Tells whether given request comes from upstream server. If authentication token is missing or invalid, it is assumed it didn’t come from upstream region.

Parameters:

request – an authenticated API request from another region

Returns:

True if request came from upstream region based on authentication token

regions.utils.is_username_unique(username: str) → bool

Checks if username isn’t already registered in this or another region. Returns True if username is not taken (can be used), or False if username is already taken

regions.utils.make_unique_username(username: str) → str

Given a username, update it to make unique globally. Adds and increments a number after username until it’s unique across all regions

Parameters:

username – a username string

Returns:

the same username if already unique, or updated username with appended number to make it unique

regions.utils.generate_upstream_unique_usernames(usernames: dict[str, str]) → dict[str, str]

Checks if usernames are unique on upstream servers. The upstream server generates unique usernames and returns a map of original -> unique username. usernames are updated with this data and returned.

Parameters:

usernames – an initial mapping of usernames mapped to themselves

Returns:

mutated usernames dict with updated values to be unique

regions.utils.remove_region_user_links(usernames: list[str]) → None

Removes users from upstream region. Use to free-up usernames after accounts have been deleted or their creation rolled back.

Parameters:

usernames – names to be deleted

Dashboards

exception dashboard_widgets.utils.InvalidFilterError

Exception raised when an invalid filter is passed (e.g. filter field does not exist)

exception dashboard_widgets.utils.DashboardConfigError(message: str, obj: BaseModel | None = None, *, url: str = '', config_field: str = '')

Exception raised when widgets config is in invalid due to user (configuration) error. For instance, configuration references a field that does not exist, or a required configuration field is not set.

This exception typically means the error can be remediated by the user by editing the WidgetConfig. However, if the widget is hardcoded, the configuration can not be edited by the user.

Parameters:

• message – the message explaining the error (mandatory); can be a lazy-translated string

• obj – the object whose configuration caused the error

• url – url at which the error occurred (if any)

• config_field – the name of the configuration field that caused the error (if any) to help troubleshooting

dashboard_widgets.utils.get_field_name_lookup(field_name: str) → tuple[str, Optional[str]]

Parses field name lookup and returns tuple of actual field name, lookup

Parameters:

field_name – field name to parse

Returns:

field name, lookup (if exists otherwise None)

dashboard_widgets.utils.get_field_names(filters: dict[str, Union[str, int, bool, list, float, NoneType]]) → list[str]

Get field names from FiltersDict, also removes lookup from filters key and returns actual field name

Parameters:

filters –

Returns:

dashboard_widgets.utils.get_field(form: AuditForm, field_name: str, include_hardcoded_fields: list[str] = None) → CustomField | Field

Gets Field or CustomField or related form CustomField instance by name. Looks for model fields first, then custom fields, then related form custom fields so passing any non-question field (Ward, auditor, date, etc) will result in that model field being returned. If custom field is unpublished, it will still be returned.

Parameters:

• form – Audit form that contains the field

• field_name – name of the field, or custom field

• include_hardcoded_fields – array of included hardcoded fields

Returns:

CustomField or model field instance

Raises:

KeyError – if field with given name does not exist in the form or any subform or related form.

dashboard_widgets.utils.get_nested_field(model: Model, field_path: str) → Field

Get Nested hardcoded field instance by key

Parameters:

• model – Base model to get nested field from

• field_path – nested field key with

Returns:

model field instance

Raises:

KeyError – if field with given name does not exist

dashboard_widgets.utils.get_field_choices(form: AuditForm | None = None, include_related: bool = False, include_hardcoded: bool = True, compliance_only: bool = False, additional_fields: list[str] = None, field_widgets: list[django.forms.widgets.Widget] | None = None, only_compliance: bool = False) → list[dashboard_widgets.widgets.types.ChoiceSchema]

For use in widget config schemas. Generates choices for form questions.

Parameters:

• form – A form to generate question choices for. If blank specific model fields will be returned.

• include_related – Whether to include fields from related forms or not.

• include_hardcoded – Whether to include hardcoded fields or not.

• compliance_only – Whether to include compliance fields only or not.

• additional_fields – List of additional custom fields, supported fields: “institution”

• field_widgets – Optionally specify the custom field widget classes to filter custom fields by. For widget config fields that only support specific data types, you can limit the custom fields to only those that accept the corresponding data type.

• only_compliance – Whether to only show compliance custom fields or not.

dashboard_widgets.utils.get_issue_field_choices(form: AuditForm | None = None, include_hardcoded: bool = True) → list[dashboard_widgets.widgets.types.ChoiceSchema]

For use in widget config schemas. Generates choices for issue fields.

Parameters:

• form – A form to generate issue field choices for. If blank specific model fields will be returned.

• include_hardcoded – Whether to include hardcoded fields or not.

dashboard_widgets.utils.get_model_choices(model: Model, dashboard: Dashboard, value_property: str = 'pk') → list[ChoiceSchema]

For use in widget config schemas. Generates choices for a model.

Parameters:

• model – The database model to get choices for.

• dashboard – The dashboard containing the widget.

• value_property – The property name of the value field.

dashboard_widgets.utils.get_tag_choices(dashboard: Dashboard) → list[ChoiceSchema]

Gets choices for tags related to a dashboard.

dashboard_widgets.utils.get_team_choices(dashboard: Dashboard, value_property: str = 'name') → list[ChoiceSchema]

For use in widget config schemas. Generates choices for teams.

Parameters:

• dashboard – The dashboard containing the widget.

• value_property – The property name of the value field.

dashboard_widgets.utils.get_subform_choices(form: AuditForm | None = None) → list[dashboard_widgets.widgets.types.ChoiceSchema]

For use in widget config schemas. Generates choices for subforms.

Parameters:

form – A form to generate subform choices for. If blank no choices are returned.

dashboard_widgets.utils.is_choice_field(schema: SchemaDict) → bool

For use in WidgetChangeForm to determine if a field schema defines a choice field.

dashboard_widgets.utils.get_custom_query_fields(form: AuditForm) → dict[str, Union[django.db.models.query_utils.Q, Callable[[Union[str, int, bool, list, float, NoneType]], django.db.models.query_utils.Q]]]

Returns custom query fields that are not related to any hardcoded question or CustomField

Parameters:

form – Audit form

Returns:

map of custom field annotation name and annotation query either as Q or as callable that returns Q

dashboard_widgets.utils.filter_observations_or_issues(form: AuditForm, qs: ObservationQuerySet | IssueQueryset, filters: FiltersDict, questions: dict[str, Question], ignore_names=None, query_fields: dict[str, CustomQueryField] = None) → ObservationQuerySet | IssueQueryset

Given a queryset of observations and a dictionary of filters, applies filtering to the queryset.

The input filter dictionary can map observation or subobservation field to a value. The filter key must be a valid name of either custom or hardcoded field. The value can be a single value or a list of values (observation having any of the values in the list will be included in the resulting queryset). Empty filters dict will result in the same queryset being returned.

Parameters:

• form – Audit form

• qs – observations or issues queryset

• filters – Filters map

• questions – questions map

• ignore_names – ignored fields names

• query_fields – custom fields queries map

Returns:

filtered queryset

Raises:

InvalidFilterError – if the passed filters are not valid and cannot be evaluated, e.g. the field named in the filter does not exist

dashboard_widgets.utils.get_dashboard_url(dashboard: Dashboard, *, widget: WidgetConfig | None | int = None, form_type: FormType | None = None, institution: Institution | None = None) → str

Generates url to the given dashboard or widget, given preferred form and institution parameters.

dashboard_widgets.utils.import_widgets()

Ensures that widget classes are imported so that they can be picked up by itersubclasses() when looking for dashboard widget classes. @cache decorator ensures that this function is used only once per python process as subsequent calls would be redundant. This allows widgets to be imported lazily.

dashboard_widgets.utils.get_user_editable_dashboard_ids(auditor: Auditor) → set[int]

For given user profile, returns a set of dashboard ids the user can edit by excluding dashboard they don’t have access to and global (shared) dashboards. Note: This utility does not check if user has edit dashboard permission unpublished dashboards are not excluded. Editable dashboard implies all widgets within the dashboard can be edited as well

dashboard_widgets.utils.make_date_getter(date_field: Question | None, *observations: ObservationQuerySet) → Callable[[Observation], datetime.datetime]

Builds a function that given observation returns its date/time given date field. It caches the results between observations from the same session.

Parameters:

• date_field – date field used to determine date/time. This defaults to session end date. If form uses a custom date field, it must be passed here.

• observations – list of observation querysets to pre-load and pre-cache end-date. This will help avoid per-observation queries when invoking the callable.

dashboard_widgets.utils.is_nan_value(value: Any) → bool

Safely check if a value is NaN, handling any input type

class megforms.calc_utils.Granularity(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Available granularity values copied from: https://www.chartjs.org/docs/latest/axes/cartesian/time.html#time-units commented-out values are currently not supported, but code is kept for future-proofing and to demonstrate the logic behind why specific values are chosen. Lower granularity = smaller time slices = smaller number representation.

get_truncator_class() → type[django.db.models.functions.datetime.TruncBase]

Django QS aggregation truncator class

get_relativedelta() → relativedelta

Get relative delta between keys for each Granularity

format_datetime(value: Timestamp | datetime) → str

Formats datetime object using this granularity and respecting current locale.

Parameters:

value – a pandas Timestamp, or datetime object. Either naive, or tz-aware is supported

get_keys_from_daterange(start: date, end: date) → Iterator[tuple[datetime.date, datetime.date]]

yields all possible granularity keys between given start and end dates

megforms.calc_utils.select_granularity(start_date: None | datetime | date, end_date: None | datetime | date, max_days: timedelta | int = datetime.timedelta(days=32)) → Granularity

Selects daily / monthly granularity for line chart based on time span between the two dates.

Parameters:

• start_date – start date of the timespan

• end_date – end date of the timespan

• max_days – max days to show daily granularity, else monthly granularity is used.

Returns:

GRANULARITY_MONTH or GRANULARITY_DAY depending on selection

megforms.calc_utils.get_count(observation: BaseAuditModel) → int

Simple function that returns count of passed in observations (1) used to implement calculation function for calculate_over_time

megforms.calc_utils.make_date_key(date: datetime | date, granularity: Granularity, always_return_date: bool = False) → date | int | str

Snaps given date to a start of daterange given granularity (day, month, etc)

Parameters:

• date – date to snap to the start of daterange given granularity

• granularity – GRANULARITY_MONTH, GRANULARITY_DAY, or any Granularity value

• always_return_date – instead of returning str or int for some granularity, always return a date

Returns:

first day of the date range where given date belongs

megforms.calc_utils.calculate_over_time(observations: Iterable[BaseAuditModel], calculation: Callable[[BaseAuditModel], int | float | None], date_getter: Callable[[BaseAuditModel], datetime.date | datetime.datetime] | None = None, granularity=Granularity.month, average: bool | str = True, include_none_compliance: bool = False) → ChartValues

Calculates monthly compliance / observation count.

Parameters:

• observations – collection of observations for which data will be calculated

• calculation – what type of built-in calculation should be carried out, or a function that accepts observation and returns value

• date_getter – function getting date given observation (leave blank to get date from session end time). Callable must return a date object.

• granularity – GRANULARITY_MONTH, GRANULARITY_DAY, or a callable returning either to select granularity used to plot days on line chart

• average – whether average of values should be calculated for when multiple values fall within one data point. Else, sum will be calculated

• include_none_compliance – whether to include null compliance values.

Returns:

list of tuples (value, month name) where value is a compliance or count depending on arguments passed

class megforms.calc_utils.BoxPlotValues

megforms.calc_utils.calculate_boxplot_values(values: list[Union[int, float]], decimal_places: int | None = None) → BoxPlotValues

Calculate and return min, first quantile, median, third quantile and max for box plot chart

Parameters:

• values –

• decimal_places –

Returns:

BoxPlotValues dictionary with min, first quantile, median, third quantile and max values

megforms.calc_utils.get_target_performance_for_observations(observations: tuple[ObservationQuerySet], filters: dict, target_action: int, forms: Iterable[AuditForm]) → int

Gets the performance for a given target type for a queryset of observations.

Parameters:

• observations – The observations used to calculate the performance.

• filters – A dict to filter the observations queryset.

• target_action – The type of target action being assessed.

• forms – Audit forms to filter the observation querysets by.

Returns:

An integer representing the target performance.

megforms.calc_utils.get_observation_targets(targets: TargetQueryset, start_date: datetime.date, end_date: datetime.date, target_breakdown: str, forms: Iterable[AuditForm], observations: ObservationQuerySet | Iterable[ObservationQuerySet], target_action: int | None = 1) → Iterator[TargetValue]

Gets an iterator of TargetValue objects for the given time range, target breakdown, forms, institutions, wards and observations.

Parameters:

• targets – The queryset of targets to compute the data for.

• start_date – The start date to get targets for.

• end_date – The end date to get targets for.

• target_breakdown – For grouping the targets. Can be one of TARGET_BREAKDOWN_WARD, TARGET_BREAKDOWN_USER or TARGET_BREAKDOWN_INSTITUTION.

• forms – An iterable of AuditForm objects to filter the targets by.

• observations – A QuerySet or an iterable of ObservationQuerySet objects to use when calculating the actual values for the targets.

• target_action – The target action.

Returns:

An iterator of TargetValue objects.

megforms.calc_utils.multiply_chart_values(multiplier: float | int, values: ChartValues) → ChartValues

Multiply each of the chart values by specified value

megforms.calc_utils.get_distinct_answers(field: Question, *observations: ObservationQuerySet) → set[Answer]

Extracts a set of answers given in a set of observations. Returns answer that was used at least once, excludes null values.

megforms.calc_utils.get_choices(field: CustomField | Field) → list[Tuple[str, str]]

Returns a list of choices for choice fields whether it’s a hardcoded or custom field

megforms.calc_utils.get_choices_or_distinct_answers(field: Question, *observations: ObservationQuerySet) → list[Choice | tuple[int, str]]

Distinct choices in a given set of observations. If fields is a choice field, returns all possible choices. Otherwise, returns a list of all answers used across all provided observations.

If field is a foreign key, returns value of the foreign key and label of the related object.

megforms.calc_utils.create_chart_data(field: Question, *observations: ObservationQuerySet, groupings: dict[str, list[str]] | None = None, compliance: bool = True, total_display: str | None = None, counter_logic: Literal['audits', 'observations'] = 'observations') → Iterable[ChartValue]

Generates values for chart widget for a given field. The observations are grouped by answer to the given field, and returned data shows average compliance/count of observations for each of the answers.

For choice fields, the generated dataset will include answers that have no values (None compliance, or 0 count) - get_choices_or_distinct_answers() is used.

Example result for Hand Hygiene profession field:

(100.0, "1.0 Any nurse or midwife")
(None, "1.1 Nurse")
...
(94.0, '4.15 Non-core allied health/other')

Parameters:

• field – the form field

• observations – querysets of observations to use as data source

• groupings – Optional dictionary of group name to a list of grouped values. When present, aggregate answers for grouped values into the group name.

• compliance – Whether to output compliance (True), or simply counts (False) of observations. NOTE: the returned compliance is in 0-100 range

• total_display – format for “total” display beside each item. Leave it null to not show the totals

• counter_logic – whether to count by audit sessions or observations.

Yields:

tuples containing values (compliance or count) and their field names (or date for over time widgets).

megforms.calc_utils.group_chart_values(values: ChartValues, groupings: dict[str, list[str]], reduce_func=<built-in function sum>) → Iterator[ChartValue]

Given a list of chart values and a dictionary, compresses the values by grouping them and summing their values. The dictionary should map a string to a list of values that should be converted to that string. reduce_func can be passed to change how values within the group are aggregated. The function needs to accept an iterable of values and output a single value.

megforms.calc_utils.get_form_questions(audit_form: AuditForm) → list[Union[audit_builder.models.CustomField, django.db.models.fields.Field]]

Gets all questions in given form

Legacy method that may duplicate AuditForm.get_questions()

Hand Hygiene

handhygiene.utils.start_date = datetime.datetime(2011, 1, 1, 0, 0, tzinfo=zoneinfo.ZoneInfo(key='Europe/Dublin'))

Collection period 10 starts in July 2015

handhygiene.utils.calculate_collection_period(date: date) → int

Calculate collection period for the selected date

There are two collection periods per year, in May and October Each collection period number is one higher than the previous one

Parameters:

date – the datetime object

Returns:

collection period

handhygiene.utils.calculate_collection_period_dates(period: int) → tuple[datetime.datetime, datetime.datetime]

calculate start and end date of the collection period

Parameters:

period – collection period number

Returns:

tuple containing start and end dates

handhygiene.utils.get_hcw_code_and_profession(hcw_string: str) → tuple[str, str]

Split hcw_string into (code, profession) tuple

Parameters:

hcw_string – str input string, like “1.1 Nurse”

Returns:

tuple containing hcw code and profession strings, like (“1.1”, “Nurse”) or tuple with empty string and whole input string

handhygiene.utils.parse_barriers(barriers: str | None) → Iterable[str]

Use this utility to parse barriers from text format to a list of selected barriers.

Barriers is a legacy field that stores multiple of elements in a text field. The data can be comma-separated, or encoded as a python literal list.

Parameters:

barriers – raw value from barriers field (can be null)

Returns:

a sequence containing the selected barriers, or an empty sequence if barriers is empty or null

handhygiene.utils.calculate_barriers_rate(observations: QuerySet) → Iterator[tuple[int, str]]

Generates chart data for barriers field. Uses parse_barriers() to extract individual barrier choices from queryset.

Yields:

tuples containing count and label of each barrier

Translation

class audit_builder.translation_utils.TranslationModelTabbedFormMetaclass(name, bases, attrs)

class audit_builder.translation_utils.TranslationTabbedModelForm(*args, data=None, languages: set[str] = None, **kwargs)

A model form that shows tabs on top allowing to switch between language-specific fields. .. highlight:: html+django * Target form must be rendered under tab template. Example:

{% include 'translation/language_tabs.html' %}
<div class="card language-card">
    <form />
</div>

• view class should also extend TranslationFormViewMixin to hide irrelevant languages from the tab

property media

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

class audit_builder.translation_utils.TranslationFormViewMixin

Mixin for views using TranslationTabbedModelForm. Passes the relevant languages to the form

get_form_languages() → set[str]

List of languages supported by current institution or null if no preference

Custom forms

class audit_builder.utils.CustomValueSerializer(serializer, deserializer, serializer_field_class)

Create new instance of CustomValueSerializer(serializer, deserializer, serializer_field_class)

deserializer

Alias for field number 1

serializer

Alias for field number 0

serializer_field_class

Alias for field number 2

class audit_builder.utils.WeighedValue(value, weight)

Create new instance of WeighedValue(value, weight)

value: float | None

Alias for field number 0

weight: float

Alias for field number 1

get_percent() → float | None

Compliance expressed as a value between 0 and 100 (percent), or null if compliance is null

audit_builder.utils.WeightedValue

alias of WeighedValue

audit_builder.utils.get_model_field_models() → Tuple[type, ...]

Models available to model select and model multiselect fields

audit_builder.utils.get_observation_url(observation: ObservationOrSubObservation, audit_form_id: int, institution_slug: str) → str

url of the review page for this issue’s observation. The url will work only for custom observations

audit_builder.utils.get_model_content_type_choices() → QuerySet

Queryset of ContentType instances for models available for model fields

audit_builder.utils.get_remote_form_model_content_type_choices() → QuerySet

Queryset of ContentType instances for models available for remote form select fields

exception audit_builder.utils.Incompliant

Thrown when the whole observation is considered incompliant

audit_builder.utils.to_boolean(value: str | bool | None) → bool | None

Interprets string value as a boolean

audit_builder.utils.deserialize_phone_number(value: str | int | bool | list | float | None, region: str | None = None) → PhoneNumber | None

Converts a phone number string to a phone number python class. Allows you to specify the two-letter country code indicating how to interpret regional phone numbers.

Parameters:

• value – The value to parse.

• region – The country code used to interpret regional phone numbers

class audit_builder.utils.RemoteFormModelChoiceSerializer(*args, **kwargs)

This field will accept and validate an annotated value field for a customobservation model, but it will return the value instead of returning a model instance

When a field is instantiated, we store the arguments that were used, so that we can present a helpful representation of the object.

to_internal_value(data)

Transform the incoming primitive data into a native value.

to_representation(obj)

Transform the outgoing native value into primitive data.

audit_builder.utils.serialize_value(field_type: type, value: Answer) → SerializedAnswer

Turns python value into database text representation

audit_builder.utils.deserialize_value(field_type: type, value, **kwargs)

Turns database text representation into python value

audit_builder.utils.create_serializer_field(field: CustomField, serializer: AnswerContainerOrCallable) → Field

Creates DRF serializer field instance for specified CustomField

Parameters:

• field – CustomField instance

• serializer – the serializer object

audit_builder.utils.make_model(model: ModelType, form: AuditForm, **kwargs) → Model

Creates a dummy instance of model, related to the given form. This ensures that:

• dummy answers are valid and contain valid auditor/room etc choice

• no additional dummy institutions or forms are created as a result of creating a dummy object

Parameters:

• model – model class

• form – form instance to query related choices from.

• kwargs – any additional arguments for model baker factory

Returns:

model instance, or a list of models if _quantity was set

audit_builder.utils.gen_multichoice(choices: Iterable[Tuple[str, str]], required: bool) → Callable[[], List[str]]

Generates a multichoice selection from a repeated list of choices Choices can be repeated to increase their probability. The number of choices returned will be random between 1 (or 0 if required=False) and the number of unique choices

Parameters:

• choices – an iterable of choices

• required – whether the result must be a non-empty list

Returns:

a callable that can be invoked to generate a random set of choices

audit_builder.utils.generate_value(custom_field: CustomField, serialize=False) → Answer | SerializedAnswer

Used to generate values for model baker

audit_builder.utils.is_choice_field(field_type)

Tells whether specified form field class accepts or requires ‘choices’

audit_builder.utils.is_model_field(field_type)

Checks whether selected form field class uses model instances as choices

class audit_builder.utils.AverageCalculator(sum=0.0, count=0.0)

Class used to calculate average value from a number of values with varying weights

add(value: float | None, weight: float = 1.0)

Add item to the calculator

Parameters:

• value – value (None values will be ignored)

• weight – weight of the value

get_average() → float | None

Calculates average value from given values

Returns:

average value or None if no entries were added

get_average_percentage() → float | None

Average value, multipled by 100

property weighed_value: WeighedValue

Represents result of this calculation as a weighted average used for further calculations and comparison

audit_builder.utils.avg_or_none(values: Iterable[float | None]) → float | None

Returns average of values in the list, or None if list is empty. Any null values are excluded

NOTE: If you know that your input does not contain nulls, use the more efficient statistics.fmean() directly, but you will also have to handle statistics.StatisticsError if your dataset is empty.

Parameters:

values – any iterable of numbers, may contain nulls

Returns:

average of passed values

audit_builder.utils.avg_or_none_weighted(values: Iterable[tuple[float | None, float]]) → float | None

Computes a weighted average of values in the iterable.

Parameters:

values – the iterable containing tuple (value, weight) where value can be a number of null, and weight is a number.

Returns:

average of values, taking into account weight of each item

audit_builder.utils.average_dict(dicts: Iterable[dict[str, float | None]], keys: Sequence[str] | None = None) → dict[str, audit_builder.utils.WeighedValue]

Given multiple dictionaries mapping string to a value, returns a single dictionary mapping each key to average value from provided dicts.

Mapped average value is an average of values, weighed by number of non-null values.

Parameters:

• dicts – any iterable containing dictionaries mapping string to a floating point or None

• keys – an optional list of tuple containing keys to extract. Leave null to extract all keys. Empty keylist will return an empty dict.

Returns:

dictionary mapping keys to average value from all provided dicts. The result is not guaranteed to contain a mapping for each of the specified keys - empty values are not included

audit_builder.utils.average_tuples(values: Iterable[tuple[str, float | None]]) → dict[str, audit_builder.utils.WeighedValue]

Calculates average value grouped by a key from an iterable of tuples.

Example:

>>> average_tuples([('key1', 1.0), ('key1', 0.0), ('key2', 1.0)])
{
    'key1': WeightedValue(0.5, 2.0),
    'key2': WeightedValue(1.0, 1.0),
}

Parameters:

values – tuples containing a repeatable key and a value

Returns:

dictionary mapping each encountered key to average value of its values and weight representing number of non-null items

audit_builder.utils.count_answers(observations: ObservationQuerySet, field: Question, counter_logic: Literal['audits', 'observations'] = 'observations') → int

If counting observations counts how many observations and subobservations have an answer to the given field. If counting audits counts the number of audits which have a observations or subobservations with the question answered.

Exclude is used here so we find observations & sub observations with the question answered by filtering out those that don’t.

This function is used for custom fields in custom_answers and also database fields for the audit form like auditor, ward etc. If the field is not a custom field and its in the CustomObservation its a db field so we query it directly.

Parameters:

• observations – A QuerySet of audit observations.

• field – The field to check for an answer. Can be a custom field or a hardcoded question or hardcoded field.

• counter_logic – Determines the counting strategy (‘audits’ or ‘observations’).

Returns:

An integer representing either the count of unique audits or the total count of observations with an answer, based on the counter_logic.

audit_builder.utils.parse_choices(raw_choices: str) → Iterable[Tuple[str, str]]

Parse raw choices and yield tuples of (value, display_value)

Parameters:

raw_choices – string of raw choices separated by newline where display value is separated from db value by comma “,”

audit_builder.utils.get_subobservations(observation: Observation, subform_or_id: int | 'CustomSubform') → Iterable[SubObservation]

Gets queryset of sub_observations of specific type for the observation

Parameters:

• observation – an observation instance (must be a CustomObservation to get a custom sub-observation instance).

• subform_or_id – an instance of SubObservation (or it’s id)

Returns:

QuerySet of sub-observations. Ordinarily it would return a sub-observation instance, but because some subforms are allowed multiple instances per observation, a collection needs to be returned. This covers all 3 possible edge cases: No SubObservation, 1 SubObservation and multiple SubObservations.

audit_builder.utils.get_answers_with_observation(custom_field: CustomField, observation: Observation) → Iterator[tuple[ObservationOrSubObservation, SerializedAnswer]]

Gets answer value from observation for specified field. Handles sub-observation traversal if custom field belongs to a subobservation.

If the observation is annotated with subform_answers uses the annotation instead of querying sub-observations. Use annotate_subform_answers() to annotate the queryset.

Yields:

tuples containing the sub/observation with the value and the value itself

audit_builder.utils.get_answers(custom_field: CustomField, observation: Observation) → tuple[SerializedAnswer, ...]

Gets answer value from observation for specified field. Handles sub-observation traversal if custom field belongs to a subobservation.

Returns:

a tuple containing the answer(s)

audit_builder.utils.choices_to_string(choices: Sequence[Tuple[str, str] | str]) → str

Converts list of choices - (value, display) tuple - to a raw_choices multiline text

Parameters:

choices – sequence of tuples (value, display)

Returns:

Choices encoded into multiline text string

audit_builder.utils.create_name(label: str, reserved_names: Iterable[str], form_field_name: str | None = None)

Turns human-readable name to a name usable as field or subform name.

Parameters:

• label – Field or subform label

• reserved_names – sequence of names that are already used

• form_field_name – name of the form field this function is used on, used for displaying an accurate error message

Returns:

string that represents base_name in lower-case, without special characters (except for _) and does not collide with reserved names.

audit_builder.utils.is_compliance_field(field: Question) → bool

Tells whether field has compliance data

audit_builder.utils.get_fields(form: AuditForm, field_names: Collection[str]) → Iterable[Question]

Given audit form and field names, generates field instances (custom or hardcoded)

audit_builder.utils.get_issue_fields(form: AuditForm, field_names: Collection[str]) → Iterable[Field | 'CustomIssueField']

Given audit form and field names, generates issue field instances (custom or hardcoded)

audit_builder.utils.get_subform_label(subform: Subform) → str

Given a subform (hardcoded or custom), returns its human-friendly name

audit_builder.utils.compress_observations(session: AuditSession) → Dict[int, Set[int]]

Moves all sub-observations to a single observation if possible without creating duplicate subforms where not allowed. Used to convert a default audit to accordion audit while keeping structure consistent with how accordion audits structure observations.

Returns:

map of moved subobservations

audit_builder.utils.make_observation_file_link(observation: ObservationOrSubObservation, field_name: str, file_index: int, file_path: str, as_thumbnail: bool = False) → str

Create absolute url link for given media file with its filename as the label.

Parameters:

path – relative path to media file

Returns:

<a> tag fith filename in its title, linking to the selected file

audit_builder.utils.sanitize_filename(path: str) → partial[str]

Returns a callable which is used to create a clean path in FileField.upload_to

audit_builder.utils.humanize_number(number: float | int, decimal_places: int | None = None) → str

Converts a number into a human-readable format with commas every three digits.

Parameters:

• number – the number to be converted.

• decimal_places – If None, number is converted to int.

Returns:

string of the humanized number.

audit_builder.utils.get_subform_fields(*subforms: Subform) → Iterator[Question]

Gets questions for all provided subforms. Filters out unpublished subforms. Does not account for subforms excluded in audit form config Supports subforms across different audit forms.

audit_builder.utils.is_iexact(value: str | int | None, expected_value: str | int | None) → bool

Case-insensitive exact match between value and expected_value

Parameters:

• value – value to check

• expected_value – the expected value to check for

Returns:

audit_builder.utils.is_icontains(container: str | int | None, member: str | int | None) → bool

Case-insensitive check if member is contained by container

Parameters:

• member – value to check

• container – container that should contain the value

Returns:

audit_builder.utils.evaluate_session_date_condition(condition: Condition, custom_answers: dict[str, Answer], session: AuditSession) → bool

Checks if the condition’s field has the same date as session.endtime.

Parameters:

• condition – The condition dictionary.

• custom_answers – The dictionary containing custom answers.

• session – The audit session object.

Returns:

True if the condition field’s answer is the same date as session.endtime.

class audit_builder.utils.RelatedObservationDataWrapper(config: FormRelatedData, observations: Iterable[CustomObservation])

Wrapper for related observation data

audit_builder.utils.populate_filter_placeholders(filters: FiltersDict, auditor: 'Auditor' | None = None) → dict[str, SerializedAnswer]

Modifies provided filters dictionary by populating placeholders with concrete data. Note that input dictionary is modified and the same dictionary instance is returned.

audit_builder.utils.filter_condition_for_field_names(condition: Condition, field_names: list[str]) → bool

Recursive helper method for finding a field name in a list of conditions. Supports checking nested conditions.

Parameters:

• condition – The condition object to check.

• field_names – The list of field names to check for in the condition.

audit_builder.utils.link_related_observations(instance: 'CustomObservation' | 'CustomSubObservationInstance', created: bool)

Links observations via common linkable field values. Removes model links when the linkable field value changes.

Parameters:

• instance – The observation or sub observation to update the model links for.

• created – Is true if the instance was just created.

LFPSE

audit_builder.lfpse.utils.clean_property_id(property_id: str) → str

Strips property id of useless meta data which has no meaning outside of the API: ‘Extension.extension:InvolvedAgents.valueCode’ to ‘InvolvedAgents’. ‘AdverseEvent.description’ to ‘description’. This allows it to map to the LFPSE taxonomy spreadsheet.

class audit_builder.lfpse.utils.CustomFieldCache(field, question_schema, parent_accordion_conditional_show_logic, required)

Create new instance of CustomFieldCache(field, question_schema, parent_accordion_conditional_show_logic, required)

field

Alias for field number 0

parent_accordion_conditional_show_logic

Alias for field number 2

question_schema

Alias for field number 1

required

Alias for field number 3

audit_builder.lfpse.utils.convert_calc_logic(calc_logic: dict, custom_fields_map: dict[str, audit_builder.lfpse.utils.CustomFieldCache]) → dict

Updates calc logic defined in question_flow.py to contain the correct field names.

audit_builder.lfpse.utils.convert_conditional_logic(field_cache: CustomFieldCache, organization: str, custom_fields_map: dict[str, audit_builder.lfpse.utils.CustomFieldCache]) → list[dict]

Converts question flow schema conditional logic: - swaps the property id for the newly generated custom field names in the fields map. - update set value placeholder with actual value from the database. - set make required for hidden required fields.

Returns:

list of conditions

audit_builder.lfpse.utils.get_help_text_from_taxonomy(field_taxonomy: Series) → str

Helper method for getting the full help text for a field.

Parameters:

field_taxonomy – A pandas series of the field’s taxonomy.

Returns:

The help text as a str.

audit_builder.lfpse.utils.get_taxonomy() → DataFrame

Loads LFPSE field data into a dataframe

audit_builder.lfpse.utils.build_field_links(form_link: LFPSEFormLink) → dict[str, audit_builder.lfpse.models.LFPSEFieldLink]

Uses LFPSEFieldBuilder to create form links, and deletes any outdated links.

Parameters:

form_link – The form whose links should be built

Returns:

mapping of each field’s property_id to its LFPSEFieldLink instance.

audit_builder.lfpse.utils.build_form_from_schema(form_link: LFPSEFormLink, fields_map: dict[str, audit_builder.lfpse.models.LFPSEFieldLink]) → tuple[int, int]

Updates form questions based on LFPSE resources

audit_builder.lfpse.utils.get_organization_choices(backend: LFPSEBackend) → list[Tuple[str, str]]

Downloads ODS codes for all organizations from LFPSE and caches them for subsequent calls. Each client (API key) has a separate cache.

Parameters:

backend – the back-end config containing a valid token

Returns:

json object containing all available organizations

Raises:

• HTTPError – if a HTTP error occurred.

• JSONDecodeError – If the response body does not contain valid json.

audit_builder.lfpse.utils.load_default_organization_choices() → list[Tuple[str, str]]

Loads organizations from a hardcoded file. The file is good for initial setup of institutions and choices can be further narrowed-down in set-up

Returns:

list of organization choices

Query utils

class utils.query.SubqueryCount(queryset, output_field=None, **extra)

Custom Count function to just perform simple count on any queryset without grouping using OuterRef and other subquery specific function.

Usage:

>>> # Counting CustomField for each AuditForm
>>> forms_qs = AuditForm.objects.annotate(
>>>     fields_count=SubqueryCount(
>>>         CustomField.objects.filter(audit_form_id=OuterRef('pk'))
>>>     )
>>> )

output_field = <django.db.models.fields.PositiveIntegerField>

class utils.query.ArraySubquery(queryset, output_field=None, **extra)

Wraps subquery in an array so that multiple values can be returned

For example, to add an array of field names to an AuditForm queryset:

AuditForm.objects.annotate(
    field_names=ArraySubquery(CustomField.objects.filter(audit_form_id=OuterRef('pk')).values('field_name')),
)

For some data types it may be necessary to override output_field:

 comments: CommentQuerySet = Comment.objects.published().annotate(
        field_comments=JSONObject(field_name=F('field_name'), comment=F('comment'))
    ).values('field_comments')
sub_observations.annotate(comments=ArraySubquery(comments.subquery(observations), output_field=ArrayField(JSONField())))

Important

The subquery must return exactly one field

Note

Order of the elements in subquery is not reflected in the resulting array

See also

Equivalent function is available in newer version of Django: https://docs.djangoproject.com/en/4.2/ref/contrib/postgres/expressions/#django.contrib.postgres.expressions.ArraySubquery

utils.query.bulk_delete(objs: QuerySet, chunk_size: int = 1000) → int

Deletes large amounts of objects from database in smaller chunks to avoid memory errors

Note

This operation is intentionally non-atomic. If deletion is interrupted, it can be re-started with the same queryset to continue deletion. If this behaviour is not desired, wrap the function call in transaction.atomic.

Parameters:

• objs – QuerySet of objects to delete

• chunk_size – Number of objects to delete at a time

Returns:

total number of deleted objects in the queryset (including related objects with on_delete=CASCADE that have been deleted as a result)

class utils.query.UnnestJson(*args, **kwargs)

Expands a JSON list [1, 2, 3] into separate rows: 1, 2, 3.

Uses PostgreSQL’s jsonb_array_elements function to transform a JSONB array into a set of rows, where each element becomes a separate row.

https://www.postgresql.org/docs/15/functions-json.html

output_field = <django.db.models.fields.json.JSONField>

class utils.query.JsonPathExists(*args, **kwargs)

Checks if a JSON path returns any items for the given JSONB field. Returns True if the path matches, False otherwise.

https://www.postgresql.org/docs/15/functions-json.html

Wraps: jsonb_path_exists(target, path)

output_field = <django.db.models.fields.BooleanField>

arity = 2

Takes two arguments (field_name, path_string). increment to enable passing more arguments as per postgres function spec

Permission utils

accounts.perm_update.get_user_forms(auditor: Auditor) → AuditFormQueryset

Gets forms accessible by given auditor in their institutions

Important

this utility ignores global/superusers

accounts.perm_update.get_user_wards(auditor: Auditor) → WardQuerySet

Gets wards available to the user in their institution.

Important

this utility ignores global/superusers

accounts.perm_update.get_base_user_folder_rules(auditor: Auditor) → FolderPermissionRuleQueryset

Get base user folder rules for building cache based of

accounts.perm_update.get_user_folders(auditor: Auditor) → FolderQuerySet

Get folders that user has access to based on folder rules and permissions

accounts.perm_update.update_user_permission_cache(auditor: Auditor, forms: Sequence[AuditForm] | None = None, update_global: bool | None = None)

Re-generates user’s permission cache and stores them in the AuditorPermissionCache model.

Parameters:

• auditor – Auditor whose form access should be updated

• forms – limit scope of the update to just the selected forms. Leave None to update all forms. Blank means no forms will be updated.

• update_global – whether to also update user’s global (non-form specific) permissions. Leave None to only update global when no forms are passed (empty, but non-null).

accounts.perm_update.update_users_perm_cache(auditors: Iterable[Auditor])

Utility to update permissions for multiple user accounts

accounts.perm_update.add_folder_to_permission_cache(auditor: Auditor, folder: Folder)

Creates or updates a cache entry for a newly created folder with permissions from its parent folder.

This is a lightweight alternative to update_folder_permission_cache() for when a single folder is created and needs to be immediately accessible without rebuilding the entire cache. This prevents 404 errors when users are redirected to newly created folders before the permission cache catches up.

For nested folders, permissions are inherited from the parent folder’s cache. For top-level folders (where parent_id is null), permissions are inherited from the global cache entry. If no parent/global cache entry exists, the cache entry won’t be created - this is expected for users without folder access.

Parameters:

• auditor – Auditor who owns the newly created folder

• folder – Folder to add to the cache

Workflow utils

megworkflows.utils.replace_placeholders_with_original_types(template: str, data: dict) → str | int | float | None

Replaces placeholders in the template with values from the data dictionary, preserving original types and supporting dot notation for nested dictionaries. this function maintains data types and allows nested key access.

megworkflows.utils.process_custom_answers(config: CreateObservationConfig, trigger_observation: CustomObservation, choice_placeholder_value: str | None = None) → dict

Process custom answers by replacing placeholders in the template strings with values from the trigger_observation custom answers.

Parameters:

• config – Operation configuration containing custom_answers templates.

• trigger_observation – Observation that triggered the workflow.

• choice_placeholder_value – Optional value to replace {choice_value} placeholders.

Returns:

Processed custom_answers dictionary with resolved values.

megworkflows.utils.create_observation(workflow_id: int, operation: Operation, trigger_observation: CustomObservation, change_time: datetime | None = None, choice_placeholder_value: str | None = None, audit_form_id: str | None = None) → CustomObservation | None

Create an observation using the provided configuration and return it.

Parameters:

• workflow_id – ID of the Workflow initiating the observation creation.

• operation – The Operation object that defines the type of observation to create.

• trigger_observation – The CustomObservation object that triggered this observation creation.

• change_time – Optional datetime representing the time at which the observation change is performed.

• choice_placeholder_value – Optional value to replace {choice_value} placeholders.

• audit_form_id – Optional audit form id whose observation should be created, skips fetching from operation.config

Returns:

A new CustomObservation object if creation is successful; otherwise, None.

megworkflows.utils.update_observation(workflow_id: int, operation: Operation, trigger_observation: CustomObservation, observation: CustomObservation) → CustomObservation

Update an observation’s answers using the provided configuration and return it. Runs custom field calculation on the updated answers, then saves observation.

Parameters:

• workflow_id – The pk of the workflow.

• operation – The operation object.

• trigger_observation – The observation that triggered the workflow.

• observation – The observation to be updated.

megworkflows.utils.query_exists_for_conditions(conditions: list, form_id: int, observation: CustomObservation, operator: str) → bool

Check if a record exists in the database that matches the given conditions. It uses the gte operator to include submissions created on the same day as the trigger observation. This ensures that matching records from the same day are considered valid.

Parameters:

• conditions – The list of conditions to evaluate.

• form_id – The ID of the form to query against.

• observation – The observation object to use in the query.

• operator – The operator to use in the query.

Returns:

True if a matching record exists, False otherwise.

megworkflows.utils.evaluate_answer_equal(conditions: dict[str, Any], observation_id: int) → bool

Evaluates if the custom answer for a given observation matches the expected answer. This function is used to check whether a specific field, identified by field_name in the observation-related data, matches the provided answer.

Parameters:

• conditions – A dictionary containing the conditions for evaluation.

• observation_id – An integer ID of the observation to be evaluated.

Returns:

Boolean value indicating whether the custom answer matches the expected one.

megworkflows.utils.evaluate_answer_changed(filters: dict[str, Any], observation_id: int) → bool

Evaluate whether the answer in question has changed to the desired value in order to process the workflow

Parameters:

• filters – the filter conditions to evaluate

• observation_id – id of the trigger observation

Returns:

False if the answer has changed to the desired answer; else True

megworkflows.utils.evaluate_query_filter_conditions(filters: Dict[str, Any], observation_id: int) → bool

Evaluate the filter conditions against the observation.

Parameters:

• filters – The filter conditions to evaluate.

• observation_id – The observation object to check against.

Returns:

True if all conditions are met, False otherwise.

megworkflows.utils.render_template_with_placeholders(template: str, data: dict, default_values: dict = None) → str

Renders a template string by replacing placeholders with values from the data dictionary. This function uses the existing replace_placeholders_with_original_types function to resolve values. Converts non strings to strings before adding the value to the template.

Parameters:

• template – The template string containing placeholders.

• data – The data dictionary containing values to replace placeholders.

• default_values – Dictionary of default values for specific placeholders.

Returns:

The rendered string with placeholders replaced.

megworkflows.utils.create_trigger_object_data(trigger_object: CustomObservation) → dict

Converts the trigger object to a dict for use in workflows.

Parameters:

trigger_object – The object to be converted into a dict.

megworkflows.utils.is_empty_or_none(value: str) → bool

Helper function to check if a value is None, empty, or whitespace-only or null_placeholder.

megworkflows.utils.trigger_email_notification(config: EmailConfig, observation: CustomObservation) → None

Trigger email notification using send_simple_mail with templated content. This function retrieves an observation, prepares a data context, and sends an email with dynamically rendered content using placeholders.

Parameters:

• config – Configuration dictionary for email composition.

• observation – The CustomObservation object that triggered this observation creation.

megworkflows.utils.trigger_sms_notification(config: SMSConfig, observation: CustomObservation) → None

Trigger SMS notification using send_sms_to_number with templated content.

Parameters:

• config – Configuration dictionary for SMS composition.

• observation – The CustomObservation object that triggered this observation creation.

megworkflows.utils.handle_create_observation_for_each_choice(workflow_id: int, operation: Operation, trigger_observation: CustomObservation, change_time: datetime | None = None) → list['CustomObservation']

Creates multiple observations, one for each selected choice in the specified multi-choice field.

If creating observations for different forms (audit_forms is set) uses audit_forms to get form IDs rather than using default (audit_form_id in config)

Parameters:

• workflow_id – ID of the Workflow that triggered the process.

• operation – The operation object containing configuration with choice_field_name.

• trigger_observation – The observation that triggered the workflow.

• change_time – Time at which this operation is performed.

Returns:

A list of new CustomObservation objects.

megworkflows.utils.trigger_web_request(config: RequestConfig, observation: CustomObservation) → None

Trigger web requests that will use the url, auth and request body from the config object after the creation/updating of the relevant observation Does not support related data from forms with nested subforms

Parameters:

• config – Configuration dictionary for request objects

• observation – The CustomObservation object that triggered this observation creation.

Returns:

megworkflows.utils.collect_subform_answers(observation: CustomObservation, custom_answers: dict, linked_choice_fields: dict[str, str] | None = None) → None

Helper function to collect all subform answers for an observation. Validates that each custom field is published and actually valid for the current subform before adding to the custom answers.

Parameters:

• observation – The observation to collect subform answers from

• custom_answers – Dictionary to update with collected answers

• linked_choice_fields – Optional dict with ‘single_choice_field_name’ and ‘multi_choice_field_name’ to link fields for pairing

megworkflows.utils.append_object_mapping(trigger_object: CustomObservation, object_mapping: dict) → dict

Extracts values from trigger_object based on dot notation paths

Parameters:

• trigger_object – The source object to extract values from

• object_mapping – Dictionary mapping dot notation paths to target keys

Returns:

Dictionary with resolved values

megworkflows.utils.convert_time(time_str: str, time_format: str) → str

Converts time from HH:MM:SS to another format.

megworkflows.utils.prepare_xml_request(config: dict, custom_answers: dict, resolved_mapping: dict = None) → str

Parses XML template and applies field mapping to create the final XML payload. It supports simple tag names or XPath expressions in the field mapping to handle nested or non-unique tags. For simple tags, it maintains backward compatibility by updating all matching tags. For XPath expressions, it allows precise targeting of elements.

Parameters:

• config – The Operation configuration object.

• custom_answers – Values from the observation to be inserted into the template

• resolved_mapping – Optional pre-resolved values to apply directly

Returns:

Formatted XML string

Document editor

editor.revision_utils.update_revision_history(instance: Model, revisions_data: list[editor.types.RevisionData])

Parses list of revisions from CKEditor and creates/updates revision history accordingly.

Objects are created if their ID does not already exist. Or updated if object with that ID already exists and is associated with the same instance. Subset of fields can be updated (for example only name if user renamed the revision).

Parameters:

• instance – object whose revision is being updated

• revisions_data – revision data received from CKEditor