Portal.Models

Address module

Address data lives in the ‘addresses’ table. Several entities link to address via foreign keys.

class portal.models.address.Address(**kwargs)

SQLAlchemy class for addresses table

as_fhir()
city
country
district
classmethod from_fhir(data)
id
line1
line2
line3
lines
postalCode
state
type
use

Audit Module

class portal.models.audit.Audit(**kwargs)

ORM class for audit data

Holds meta info about changes in other tables, such as when and by whom the data was added. Several other tables maintain foreign keys to audit rows, such as Observation and Procedure.

as_fhir()

Typically included as meta data in containing FHIR resource

comment
context
classmethod from_logentry(entry)

Parse and create an Audit instance from audit log entry

Prior to version v16.5.12, audit entries only landed in log. This may be used to convert old entries, but newer ones should already be there.

id
subject_id
timestamp
user_id
version
class portal.models.audit.Context

An enumeration.

account = 5
assessment = 2
authentication = 3
consent = 6
group = 10
intervention = 4
login = 1
observation = 8
organization = 9
other = 0
procedure = 11
relationship = 12
role = 13
tou = 14
user = 7
portal.models.audit.lookup_version()

Auth related model classes

class portal.models.auth.AuthProvider(**kwargs)
as_fhir()
created_at
id
provider
provider_id
token
user
user_id
class portal.models.auth.AuthProviderPersistable(**kwargs)

For persistence to function, need instance serialization

The base class for AuthProvider implements a non persistence-compliant version of as_fhir() as needed to show FHIR compliant identifiers in demographics.

This subclass (adapter) exists solely to provide serialization methods that work with persistence.

as_fhir()

serialize the AuthProvider

created_at
classmethod from_fhir(data)
id
provider
provider_id
token
update_from_fhir(data)
user
user_id
class portal.models.auth.Grant(**kwargs)
client
client_id
code
delete()
expires
id
redirect_uri
scopes
user
user_id
validate_redirect_uri(redirect_uri)

Validate the redirect_uri from the OAuth Grant request

The RFC requires exact match on the redirect_uri. In practice this is too great of a burden for the interventions. Make sure it’s from the same scheme:://host:port the client registered with

http://tools.ietf.org/html/rfc6749#section-4.1.3

class portal.models.auth.Mock
class portal.models.auth.Token(**kwargs)
access_token
as_json()

serialize the token - used to preserve service tokens

client
client_id
expires
classmethod from_json(data)
id
refresh_token
scopes
token_type
update_from_json(data)
user
user_id
portal.models.auth.create_service_token(client, user)

Generate and return a bearer token for service calls

Partners need a mechanism for automated, authorized API access. This function returns a bearer token for subsequent authorized calls.

NB - as this opens a back door, it’s only offered to users with the single role ‘service’.

portal.models.auth.load_grant(client_id, code)
portal.models.auth.load_token(access_token=None, refresh_token=None)
portal.models.auth.save_grant(client_id, code, request, *args, **kwargs)
portal.models.auth.save_token(token, request, *args, **kwargs)
portal.models.auth.token_janitor()

Called by scheduled job to clean up and send alerts

No value in keeping around stale tokens, so we delete any that have expired.

For service tokens, trigger an email alert if they will be expiring soon.

Returns:list of unreachable email addresses

Model classes for retaining FHIR data

class portal.models.fhir.BundleType

An enumeration.

portal.models.fhir.bundle_results(elements, bundle_type=<BundleType.searchset: 8>, links=None)

Generate FHIR Bundle from element lists

Parameters:
  • elements – iterable of FHIR Resources to bundle
  • bundle_type – limited by FHIR to be of the BundleType enum.
  • links – links related to this bundle, such as API used to generate
Returns:

a FHIR compliant bundle

portal.models.fhir.v_or_first(value, field_name)

Return desired from list or scalar value

Parameters:
  • value – the raw data, may be a single value (directly returned) or a list from which the first element will be returned
  • field_name – used in error text when multiple values are found for a constrained item.

Some fields, such as name were assumed to always be a single dictionary containing single values, whereas the FHIR spec defines them to support 0..* meaning we must handle a list.

NB - as the datamodel still only expects one, a 400 will be raised if given multiple values, using the field_name in the text.

portal.models.fhir.v_or_n(value)

Return None unless the value contains data

class portal.models.flaskdanceprovider.FacebookFlaskDanceProvider(blueprint, token)

fetches user info from Facebook after successfull auth

After the user successfully authenticates with Facebook this class fetches the user’s info from Facebook

send_get_user_json_request()

sends a GET request to Facebook for user data

This function is used to get user information from Facebook that is encoded in json.

:return Response

class portal.models.flaskdanceprovider.FlaskDanceProvider(blueprint, token, standard_key_to_provider_key_map)

base class for flask dance providers

When a new provider is added to the protal’s consumer oauth flow a descendent of this class needs to be created to get the user’s information from the provider after a successful auth

get_user_info()

gets user info from the provider

This function parses json returned from the provider and returns an instance of FlaskProviderUserInfo that is filled with the user’s information

:return FlaskProviderUserInfo with the user’s info

parse_json(user_json)

parses the user’s json and returns it in a standard format

Providers encode user information in json. This function parses the json and stores values in an instance of FlaskProviderUserInfo

Parameters:user_json – info about the user encoded in json

:return instance of FlaskProviderUserInfo with the user’s info

send_get_user_json_request()

sends a request to the provider to get user json

This function must be overriden in descendant classes to return a response with the user’s json

class portal.models.flaskdanceprovider.FlaskProviderUserInfo

a common format for user info fetched from providers

Each provider packages user info a litle differently. Google, for example, uses “given_name” and the key for the user’s first name, and Facebook uses “first_name”. To make it easier for our code to parse responses in a common function this class provides a common format to store the results from each provider.

class portal.models.flaskdanceprovider.GoogleFlaskDanceProvider(blueprint, token)

fetches user info from Google after successfull auth

After the user successfully authenticates with Google this class fetches the user’s info from Google

send_get_user_json_request()

sends a GET request to Google for user data

This function is used to get user information from Google that is encoded in json.

:return Response

class portal.models.flaskdanceprovider.MockFlaskDanceProvider(provider_name, token, user_json, fail_to_get_user_json)

creates user info from test data to validate auth logic

This class should only be used during testing. It simply mocks user json that is normally retrieved from a provider which allows us to granularly test auth logic

send_get_user_json_request()

return a mock request based on test data passed into the constructor

Normally a request is sent to a provider and user json is returned. This function mocks out that request by returning a response with the user json passed through the test backdoor

class portal.models.flaskdanceprovider.MockJsonResponse(ok, user_json)

mocks a GET json response

During auth we send a request to providers that returns user json. During tests we need to mock out providers so we can test our auth logic. This class is used to mock out requests that are normally sent to providers.

json()

returns mock json

Identifier Model Module

class portal.models.identifier.Identifier(**kwargs)

Identifier ORM, for FHIR Identifier resources

add_if_not_found(commit_immediately=False)

Add self to database, or return existing

Queries for similar, matching on system and value alone. Note the database unique constraint to match.

@return: the new or matched Identifier

class portal.models.identifier.UserIdentifier(**kwargs)

ORM class for user_identifiers data

Holds links to any additional identifiers a user may have, such as study participation.

static check_unique(user, identifier)

Raises 409 if given identifier should be unique but is in use

UserIdentifiers are not all unique - depends on the system, namely if the system is part of UNIQUE_IDENTIFIER_SYSTEMS. For example, the region system identifiers are often shared with many users. Others are treated as unique, such as study-id, and therefore raise exceptions if already in use (that is, when the given identifier is already associated with a user other than the named parameter).

Parameters:
  • identifier – identifier to check, or ignore if system isn’t treated as unique
  • user – intended recipient
Raises:

UniqueConstraint if identifier’s system is in UNIQUE_IDENTIFIER_SYSTEMS and the identifier is assigned to another, not deleted, user.

Returns:

True - exception thrown if unique “constraint” broken.

portal.models.identifier.parse_identifier_params(arg)

Parse identifier parameter from given arg

Supports FHIR pipe delimited system|value or legacy FHIR JSON named parameters {‘system’, ‘value’}

Parameters:arg – argument string, may be serialized JSON or pipe delimited
Raises:BadRequest if unable to parse valid system, value
Returns:(system, value) tuple

Intervention Module

class portal.models.intervention.DisplayDetails(access, intervention, user_intervention)

Simple abstraction to communicate display details to front end

To provide a custom experience, intevention access can be set at several levels. For a user, access is either available or not, and when available, the link controls may be intentionally disabled for a reason the intervention should note in the status_text field.

Attributes::
access: {True, False} card_html: Text to display on the card link_label: Text used to label the button or hyperlink link_url: URL for the button or link - link to be disabled when null status_text: Text to inform user of status, or why it’s disabled
class portal.models.intervention.Intervention(**kwargs)
as_json()

Returns the ‘safe to export’ portions of an intervention

The client_id and link_url are non-portable between systems. The id is also independent - return the rest of the not null fields as a simple json dict.

NB for staging exclusions to function, link_url and client_id are now included. Take care to remove it from persistence files where it is NOT portable, for example, when generating persistence files programmatically.

display_for_user(user)

Return the intervention display details for the given user

Somewhat complicated method, depending on intervention configuration. The following ordered steps are used to determine if a user should have access to an intervention. The first ‘true’ found provides access, otherwise the intervention will not be displayed.

  1. call each strategy_function in intervention.access_strategies. Note, on rare occasions, a strategy may alter the UserIntervention attributes given the circumstances.
  2. check for a UserIntervention row defining access for the given user on this intervention.
  3. check if the intervention has public_access set

@return DisplayDetails object defining ‘access’ and other details for how to render the intervention.

fetch_strategies()

Generator to return each registered strategy

Strategies need to be brought to life from their persisted state. This generator does so, and returns them in a call ready fashion, ordered by the strategy’s rank.

quick_access_check(user)

Return boolean representing given user’s access to intervention

Somewhat complicated method, depending on intervention configuration. The following ordered steps are used to determine if a user should have access to an intervention. The first ‘true’ found is returned (as to make the check as quick as possible).

  1. check if the intervention has public_access set
  2. check for a UserIntervention row defining access for the given user on this intervention.
  3. call each strategy_function in intervention.access_strategies.

@return boolean representing ‘access’.

static rct_ids()

returns list of RCT (randomized control trial) intervention ids

class portal.models.intervention.UserIntervention(**kwargs)
classmethod user_access_granted(intervention_id, user_id)

Shortcut to query for specific (intervention, user) access

portal.models.intervention.add_static_interventions()

Seed database with default static interventions

Idempotent - run anytime to push any new interventions into existing dbs

Module for intervention access strategy functions

Determining whether or not to provide access to a given intervention for a user is occasionally tricky business. By way of the access_strategies property on all interventions, one can add additional criteria by defining a function here (or elsewhere) and adding it to the desired intervention.

function signature: takes named parameters (intervention, user) and returns a boolean - True grants access (and short circuits further access tests), False does not.

NB - several functions are closures returning access_strategy functions with the parameters given to the closures.

class portal.models.intervention_strategies.AccessStrategy(**kwargs)

ORM to persist access strategies on an intervention

The function_details field contains JSON defining which strategy to use and how it should be instantiated by one of the closures implementing the access_strategy interface. Said closures must be defined in this module (a security measure to keep unsanitized code out).

as_json()

Return self in JSON friendly dictionary

instantiate()

Bring the serialized access strategy function to life

Using the JSON in self.function_details, instantiate the function and return it ready to use.

portal.models.intervention_strategies.allow_if_not_in_intervention(intervention_name)

Strategy API checks user does not belong to named intervention

portal.models.intervention_strategies.combine_strategies(**kwargs)

Make multiple strategies into a single statement

The nature of the access lookup returns True for the first success in the list of strategies for an intervention. Use this method to chain multiple strategies together into a logical and fashion rather than the built in locical or.

NB - kwargs must have keys such as ‘strategy_n’, ‘strategy_n_kwargs’ for every ‘n’ strategies being combined, starting at 1. Set arbitrary limit of 6 strategies for time being.

Nested strategies may actually want a logical ‘OR’. Optional kwarg combinator takes values {‘any’, ‘all’} - default ‘all’ means all strategies must evaluate true. ‘any’ means just one must eval true for a positive result.

portal.models.intervention_strategies.in_role_list(role_list)

Requires user is associated with any role in the list

portal.models.intervention_strategies.limit_by_clinic_w_id(identifier_value, identifier_system='http://us.truenth.org/identity-codes/decision-support-group', combinator='any', include_children=True)

Requires user is associated with {any,all} clinics with identifier

Parameters:
  • identifier_value – value string for identifer associated with org(s)
  • identifier_system – system string for identifier, defaults to DECISION_SUPPORT_GROUP
  • combinator – determines if the user must be in ‘any’ (default) or ‘all’ of the clinics in the given list. NB combining ‘all’ with include_children=True would mean all orgs in the list AND all chidren of all orgs in list must be associated with the user for a true result.
  • include_children – include children in the organization tree if set (default), otherwise, only include the organizations in the list
portal.models.intervention_strategies.not_in_clinic_w_id(identifier_value, identifier_system='http://us.truenth.org/identity-codes/decision-support-group', include_children=True)

Requires user isn’t associated with any clinic in the list

Parameters:
  • identifier_value – value string for identifer associated with org(s)
  • identifier_system – system string for identifier, defaults to DECISION_SUPPORT_GROUP
  • include_children – include children in the organization tree if set (default), otherwise, only include the organizations directly associated with the identifier
portal.models.intervention_strategies.not_in_role_list(role_list)

Requires user isn’t associated with any role in the list

portal.models.intervention_strategies.observation_check(display, boolean_value, invert_logic=False)

Returns strategy function for a particular observation and logic value

Parameters:
  • display – observation coding.display from TRUENTH_CLINICAL_CODE_SYSTEM
  • boolean_value – ValueQuantity boolean true or false expected
  • invert_logic – Effective binary not to apply to test. If set, will return True only if given observation with boolean_value is NOT defined for user

NB a history of observations is maintained, with the most recent taking precedence.

portal.models.intervention_strategies.tx_begun(boolean_value)

Returns strategy function testing if user is known to have started Tx

Parameters:boolean_value – true for known treatment started (i.e. procedure indicating tx has begun), false to confirm a user doesn’t have a procedure indicating tx has begun
portal.models.intervention_strategies.update_card_html_on_completion()

Update description and card_html depending on state

portal.models.lazy.lazyprop(fn)

Property decorator for lazy intialization (load on first request)

Useful on any expensive to load attribute on any class. Simply decorate the ‘getter’ with @lazyprop, where the function definition loads the object to be assigned to the given attribute.

As the SQLAlchemy session is NOT thread safe and this tends to be the primary use of the lazyprop decorator, we include the thread identifier in the key

portal.models.lazy.query_by_name(cls, name)

returns a lazy load function capable of caching object

Use this alternative for classes with dynamic attributes (names not hardcoded in class definition), as property decorators (i.e. @lazyprop) don’t function properly.

As the SQLAlchemy session is NOT thread safe, we include the thread identifier in the key

NB - attribute instances must be unique over (cls.__name__, name) within the containing class to avoid collisions.

@param cls: ORM class to query @param name: name field in ORM class to uniquely define object

Model classes for message data

class portal.models.message.EmailMessage(**kwargs)
as_json()
body
id
recipients
send_message(cc_address=None)

Send the message

Parameters:cc_address – include valid email address to send a carbon copy

NB the cc isn’t persisted with the rest of the record.

sender
sent_at
static style_message(body)

Implicitly called on send, to wrap body with style tags

subject
user_id
portal.models.message.log_message(message, app)

Configured to handle signals on email_dispatched - log the event

Model classes for organizations and related entities.

Designed around FHIR guidelines for representation of organizations, locations and healthcare services which are used to describe hospitals and clinics.

class portal.models.organization.LocaleExtension(organization, extension)
children
extension_url = 'http://hl7.org/fhir/valueset/languages'
class portal.models.organization.OrgNode(id, parent=None, children=None)

Node in tree of organizations - used by org tree

Simple tree implementation to house organizations in a hierarchical structure. One root - any number of nodes at each tier. The organization identifiers (integers referring to the database primary key) are used as reference keys.

insert(id, partOf_id=None)

Insert new nodes into the org tree

Designed for this special organizaion purpose, we expect the tree is built from the top (root) down, so no rebalancing is necessary.

Parameters:
  • id – of organizaiton to insert
  • partOf_id – if organization has a parent - its identifier
Returns:

the newly inserted node

top_level()

Lookup top_level organization id from the given node

Use OrgTree.find() to locate starter node, if necessary

class portal.models.organization.OrgTree

In-memory organizations tree for hierarchy and structure

Organizations may define a ‘partOf’ in the database records to describe where the organization fits in a hierarchy. As there may be any number of organization tiers, and the need exists to lookup where an organiztion fits in this hiearchy. For example, needing to lookup the top level organization for any node, or all the organizations at or below a level for permission issues. etc.

This singleton class will build up the tree when it’s first needed (i.e. lazy load).

Note, the root of the tree is a dummy object, so the first tier can be multiple top-level organizations.

static all_ids_with_rp(research_protocol)

Returns set of org IDs that are associated with Research Protocol

As child orgs are considered to be associated if the parent org is, this will return the full list for optimized comparisons.

all_leaf_ids()
all_leaves_below_id(organization_id)

Given org at arbitrary level, return list of leaf nodes below it

all_top_level_ids()

Return list of all top level organization identifiers

at_and_above_ids(organization_id)

Returns list of ids from any point in tree and up the parent stack

Parameters:organization_id – node in tree, will be included in return list
Returns:list of organization ids from the one given on up including every parent found in chain
at_or_below_ids(organization_id, other_organizations)

Check if the other_organizations are at or below given organization

Parameters:
  • organization_id – effective parent to check against
  • other_organizations – iterable of organization_ids as potential children.
Returns:

True if any org in other_organizations is equal to the given organization_id, or a child of it.

find(organization_id)

Locates and returns node in OrgTree for given organization_id

Parameters:organization_id – primary key of organization to locate
Returns:OrgNode from OrgTree
Raises:ValueError if not found - unexpected
find_top_level_orgs(organizations, first=False)

Returns top level organization(s) from those provided

Parameters:
  • organizations – organizations against which top level organization(s) will be queried
  • first – if set, return the first org in the result list rather than a set of orgs.
Returns:

set of top level organization(s), or a single org if first is set.

here_and_below_id(organization_id)

Given org at arbitrary level, return list at and below

classmethod invalidate_cache()

Invalidate cache on org changes

lookup_table = None
populate_tree()

Recursively build tree from top down

root = None
top_level_names()

Fetch org names for all_top_level_ids

Returns:list of top level org names
visible_patients(staff_user)

Returns patient IDs for whom the current staff_user can view

Staff users can view all patients at or below their own org level.

NB - no patients should ever have a consent on file with the special organization ‘none of the above’ - said organization is ignored in the search.

class portal.models.organization.Organization(**kwargs)

Model representing a FHIR organization

Organizations represent a collection of people that have come together to achieve an objective. As an example, all the healthcare services provided by the same university hospital will belong to the organization representing said university hospital.

Organizations can reference other organizations via the ‘partOf_id’, where children name their parent organization id.

addresses
as_fhir(include_empties=True)

Return JSON representation of organization

Parameters:include_empties – if True, returns entire object definition; if False, empty elements are removed from the result
Returns:JSON representation of a FHIR Organization resource
coding_options
static consent_agreements(locale_code)

Return consent agreements for all top level organizations

Parameters:locale_code – preferred locale, typically user’s.
Returns:dictionary keyed by top level organization id containing a VersionedResource for each organization IFF the organization has a custom consent agreement on file. The organization_name is also added to the versioned resource to simplify UI code.
default_locale
default_locale_id
email
ethnicity_codings
classmethod from_fhir(data)
classmethod generate_bundle(limit_to_ids=None, include_empties=True)

Generate a FHIR bundle of existing orgs ordered by ID

Parameters:
  • limit_to_ids – if defined, only return the matching set, otherwise all organizations found
  • include_empties – set to include empty attributes
Returns:

id
identifiers
indigenous_codings
invalidation_hook()

Endpoint called during site persistence import on change

Any site persistence aware class may implement invalidation_hook to be notified of changes during import.

Designed to allow for cache invalidation or other flushing needed on state changes. As organizations define users affiliation with questionnaires via research protocol, such a change means flush any existing qb_timeline rows for member users

locales
name
organization_research_protocols
partOf_id
phone
phone_id
race_codings
research_protocol(as_of_date)

Lookup research protocol for this org valid at as_of_date

Complicated scenario as it may only be defined on the parent or further up the tree. Secondly, we keep history of research protocols in case backdated entry is necessary.

Returns:research protocol for org (or parent org) valid as_of_date
research_protocols

A descriptor that presents a read/write view of an object attribute.

rps_w_retired(consider_parents=False)

accessor to collate research protocols and retired_as_of values

The SQLAlchemy association proxy doesn’t provide easy access to intermediary table data - i.e. columns in the link table between a many:many association. This accessor collates the value stored in the intermediary table, retired_as_of with the research protocols for this organization.

Parameters:consider_parents – if set and the org doesn’t have an associated RP, continue up the org hiearchy till one is found.
Returns:ready query for use in iteration or count or other methods. Query will produce a list of tuples (ResearchProtocol, retired_as_of) associated with the organization, ordered by retired_as_of dates with nulls last.
shortname

Return shortname identifier if found, else the org name

timezone
type
type_id
update_from_fhir(data)
use_specific_codings
users
class portal.models.organization.OrganizationAddress(**kwargs)

link table for organization : n addresses

address_id
id
organization_id
class portal.models.organization.OrganizationIdentifier(**kwargs)

link table for organization : n identifiers

id
identifier_id
organization_id
class portal.models.organization.OrganizationLocale(**kwargs)
coding_id
id
organization_id
class portal.models.organization.OrganizationResearchProtocol(research_protocol=None, organization=None, retired_as_of=None)
id
organization
organization_id
research_protocol
research_protocol_id
retired_as_of
class portal.models.organization.ResearchProtocolExtension(organization, extension)
apply_fhir()
as_fhir(include_empties=True)
children
extension_url = 'http://us.truenth.org/identity-codes/research-protocol'
class portal.models.organization.UserOrganization(**kwargs)

link table for users (n) : organizations (n)

id
organization
organization_id
user_id
portal.models.organization.add_static_organization()

Insert special none of the above org at index 0

portal.models.organization.org_extension_map(organization, extension)

Map the given extension to the Organization

FHIR uses extensions for elements beyond base set defined. Lookup an adapter to handle the given extension for the organization.

Parameters:
  • organization – the org to apply to or read the extension from
  • extension – a dictionary with at least a ‘url’ key defining the extension.
Returns:

adapter implementing apply_fhir and as_fhir methods

:raises exceptions.ValueError: if the extension isn’t recognized

Performer module - encapsulate the FHIR Performer resource

class portal.models.performer.ObservationPerformer(**kwargs)

Link table for observation to list of performers

id
observation_id
performer_id
class portal.models.performer.Performer(**kwargs)

ORM for FHIR Performer - performers table

add_if_not_found(commit_immediately=False)

Add self to database, or return existing

Queries for matching, existing Performer. Populates self.id if found, adds to database first if not.

as_fhir()

Return self in JSON FHIR formatted string

FHIR is not currently consistant in performer inclusion. For example, Observation.performer is simply a list of Reference resources, whereas Procedure.performer is a list including the resource labeled as an actor and a codable concept labeled as the role defining the actor’s role.

Returns:the best JSON FHIR formatted string for the instance
codeable_concept
codeable_concept_id

The codeable concept for performers including a role

classmethod from_fhir(fhir)

Return performer instance from JSON FHIR formatted string

See note in as_fhir, the format of a performer depends on context. Populate self.codeable_concept only if it’s included as a role.

Returns:new performer instance from values in given fhir
id
observations
reference_txt

Text for performer (aka actor), i.e. {“reference”: “patient/12”}

Procedure Model

class portal.models.procedure.Procedure(**kwargs)

ORM class for procedures

Similar to the profiles published by SMART

Each Procedure must haveProcedure must have:
 
1 patient:in Procedure.subject (aka Procedure.user)
1 code:in Procedure.code (pointing to a CodeableConcept) with system of http://snomed.info/sct
1 performed datetime:
 in Procedure.performedDateTime
as_fhir()

produces FHIR representation of procedure in JSON format

audit

tracks when and by whom the procedure was retained, included as meta data in the FHIR output

code

procedure.code (a CodeableConcept) defines the procedure. coding.system is required to be http://snomed.info/sct

end_time

when defined, produces a performedPeriod, otherwise start_time is used alone as performedDateTime

classmethod from_fhir(data, audit)

Parses FHIR data to produce a new procedure instance

start_time

required whereas end_time is optional

Reference module - encapsulate FHIR Reference type

exception portal.models.reference.MissingReference

Raised when FHIR references cannot be found

exception portal.models.reference.MultipleReference

Raised when FHIR references retrieve multiple results

class portal.models.reference.Reference
as_fhir()

Return FHIR compliant reference string

FHIR uses the Reference Resource within a number of other resources to define things like who performed an observation or what organization another is a partOf.

Returns:the appropriate JSON formatted reference string.
classmethod intervention(intervention_id)

Create a reference object from given intervention

Intervention references maintained by name - lookup from given id.

classmethod organization(organization_id)

Create a reference object from a known organization id

classmethod parse(reference_dict)

Parse an organization from a FHIR Reference resource

Typical format: “{‘Reference’: ‘Organization/12’}” or “{‘reference’: ‘api/patient/6’}”

FHIR is a little sloppy on upper/lower case, so this parser is also flexible.

Returns:the referenced object - instantiated from the db
:raises portal.models.reference.MissingReference: if
the referenced object can not be found
:raises portal.models.reference.MultipleReference: if
the referenced object retrieves multiple results
:raises exceptions.ValueError: if the text format
can’t be parsed
classmethod patient(patient_id)

Create a reference object from a known patient id

classmethod practitioner(practitioner_id)

Create a reference object from a known patient id

classmethod questionnaire(questionnaire_name)

Create a reference object from a known questionnaire name

classmethod questionnaire_bank(questionnaire_bank_name)

Create a reference object from a known questionnaire bank

classmethod research_protocol(research_protocol_name)

Create a reference object from a known research protocol

Relationship module

Relationship data lives in the relationships table, populated via:
FLASK_APP=manage.py flask seed

To extend the list of roles, add name: description pairs to the STATIC_RELATIONSHIPS dict within, and rerun the seed command above.

class portal.models.relationship.Relationship(**kwargs)

SQLAlchemy class for relationships table

description
id
name
portal.models.relationship.add_static_relationships()

Seed database with default static relationships

Idempotent - run anytime to pick up any new relationships in existing dbs

Role module

Role data lives in the roles table, populated via:
flask seed
To restrict access to a given role, use the ROLE object:
@roles_required(ROLE.ADMIN.value)

To extend the list of roles, add name: description pairs to the STATIC_ROLES dict within.

class portal.models.role.Role(**kwargs)

SQLAlchemy class for roles table

as_json()
description
display_name

Generate and return ‘Title Case’ version of name ‘title_case’

id
name
users
portal.models.role.add_static_roles()

Seed database with default static roles

Idempotent - run anytime to pick up any new roles in existing dbs

Telecom Module

FHIR uses a telecom structure for email, fax, phone, etc.

class portal.models.telecom.ContactPoint(**kwargs)

ContactPoint model for storing FHIR telecom entries

as_fhir()
classmethod from_fhir(data)
id
rank
system
update_from_fhir(data)
use
value
class portal.models.telecom.Telecom(email=None, contact_points=None)

Telecom model - not a formal db front at this time

Several FHIR resources include telecom entries. This helper class wraps common functions.

as_fhir()
cp_dict()
classmethod from_fhir(data)

User model

exception portal.models.user.RoleError
class portal.models.user.User(**kwargs)
active
add_identifier(identifier)
add_observation(fhir, audit)
add_organization(organization_name)

Shortcut to add a clinic/organization by name

add_password_verification_failure()

remembers when a user fails password verification

Each time a user fails password verification this function is called. Use user.is_locked_out to tell whether this has been called enough times to lock the user out of the system

Returns:total failures since last reset
add_relationship(other_user, relationship_name)
add_roles(role_list, acting_user)

Add one or more roles to user’s existing roles

Parameters:
  • role_list – list of role objects defining what roles to add
  • acting_user – user performing action, for permissions, etc.
Raises:

409 if any named roles are already assigned to the user

add_service_account()

Service account generation.

For automated, authenticated access to protected API endpoints, a service user can be created and used to generate a long-life bearer token. The account is a user with the service role, attached to a sposor account - the (self) individual creating it.

Only a single service account is allowed per user. If one is found to exist for this user, simply return it.

all_consents

Access to all consents including deleted and expired

alt_phone
alt_phone_id
as_fhir(include_empties=True)

Return JSON representation of user

Parameters:include_empties – if True, returns entire object definition; if False, empty elements are removed from the result
Returns:JSON representation of a FHIR Patient resource
auth_providers
birthdate
check_role(permission, other_id)

check user for adequate role

if user is an admin or a service account, grant carte blanche otherwise, must be self or have a relationship granting permission to “verb” the other user.

returns true if permission should be granted, raises 404 if the other_id can’t be found, otherwise raise a 401

clinical_history(requestURL=None, patch_dstu2=False)
classmethod column_names()
concept_value(codeable_concept)

Look up logical value for given concept

Returns the most current setting for a given concept, by interpreting the results of a matching fetch_value_status_for_concept() call.

NB - as there are states beyond true/false, such as “unknown” for a given concept, this does NOT return a boolean but a string.

Returns:a string, typically “true”, “false” or “unknown”
confirmed_at
current_encounter

Shortcut to current encounter, if present

An encounter is typically bound to the logged in user, not the subject, if a different user is performing the action.

deactivate_tous(acting_user, types=None)

Mark user’s current active ToU agreements as inactive

Marks the user’s current active ToU agreements as inactive. User must agree to ToUs again upon next login (per CoreData logic). If types provided, only deactivates agreements of that ToU type. Called when the ToU agreement language is updated.

Parameters:
  • acting_user – user behind the request for permission checks
  • types – ToU types for which to invalide agreements (optional)
deceased
deceased_id
delete_roles(role_list, acting_user)

Delete one or more roles from user’s existing roles

Parameters:
  • role_list – list of role objects defining what roles to remove
  • acting_user – user performing action, for permissions, etc.
Raises:

409 if any named roles are not currently assigned to the user

delete_user(acting_user)

Mark user deleted from the system

Due to audit constraints, we do NOT actually delete the user, but mark the user as deleted. See permanently_delete_user for more serious alternative.

Parameters:
  • self – user to mark deleted
  • acting_user – individual executing the command, for audit trail
deleted
deleted_id
display_name
documents
email
email_ready()

Returns (True, None) IFF user has valid email & necessary criteria

As user’s frequently forget their passwords or start in a state without a valid email address, the system should NOT email invites or reminders unless adequate data is on file for the user to perform a reset password loop.

NB exceptions exist for systems with the NO_CHALLENGE_WO_DATA configuration set, as those systems allow for change of password without the verification step, if the user doesn’t have a required field set.

Returns:(Success, Failure message), such as (True, None) if the user account is “email_ready” or (False, _”invalid email”) if the reason for failure is a lack of valid email address.
encounters
ethnicities
external_study_id

Return the value of the user’s external study identifier(s)

If more than one external study identifiers are found for the user, values will be joined by ‘, ‘

failed_login_attempts_before_lockout

Number of failed login attempts before lockout

fetch_datetime_for_concept(codeable_concept)

Return newest issued timestamp from matching observation

fetch_value_status_for_concept(codeable_concept)

Return matching ValueQuantity & status for this user

Given the possibility of multiple matching observations, returns the most current info available.

See also concept_value()

Returns:(value_quantity, status) tuple for the observation if found on the user, else (None, None)
first_name
first_top_organization()

Return first top level organization for user

NB, none of the above doesn’t count and will not be retuned.

A user may have any number of organizations, but most business decisions, assume there is only one. Arbitrarily returning the first from the matching query in case of multiple.

Returns:a single top level organization, or None
classmethod from_fhir(data)
fuzzy_match(first_name, last_name, birthdate)

Returns probability score [0-100] of it being the same user

gender
groups
has_relationship(relationship_name, other_user)
has_role(role_name)

Return True if the user has one of the specified roles. Return False otherwise.

has_roles() accepts a 1 or more role name parameters
has_role(role_name1, role_name2, role_name3).
For example:
has_roles(‘a’, ‘b’)
Translates to:
User has role ‘a’ OR role ‘b’
id
identifiers

Return list of identifiers

Several identifiers are “implicit”, such as the primary key from the user table, and any auth_providers associated with this user. These will be prepended to the existing identifiers but should never be stored, as they’re generated from other fields.

Returns:list of implicit and existing identifiers
image_url
implicit_identifiers()

Generate and return the implicit identifiers

The primary key, email and auth providers are all visible in formats such as demographics, but should never be stored as user_identifiers, less problems of duplicate, out of sync data arise.

This method generates those on the fly for display purposes.

Returns:list of implicit identifiers
indigenous
interventions
is_locked_out

tells if user is temporarily locked out

To slow down brute force password attacks we temporarily lock users out of the system for a short period of time. This property tells whether or not the user is locked out.

is_registered()

Returns True if user has completed registration

Not to be confused with the registered column (which captures the moment when the account was created), is_registered returns true once the user has blessed their account with login credentials, such as a password or auth_provider access.

Roles are considered in this check - special roles such as access_on_verify and write_only should never exist on registered users, and therefore this method will return False for any users with these roles.

last_name
last_password_verification_failure
leaf_organizations()

Return list of ‘leaf’ organization ids for user’s orgs

Users, especially staff, have arbitrary number of organization associations, at any level of the organization hierarchy. This method looks up all child leaf nodes from the users existing orgs.

locale
locale_code
locale_display_options

Collates all the locale options from the user’s orgs to establish which should be visible to the user

locale_id
locale_name
lockout_period_minutes

The lockout period in minutes

lockout_period_timedelta

The lockout period as a timedelta

mask_email(prefix='__invite__')

Mask temporary account email to avoid collision with registered

Temporary user accounts created for the purpose of invites get in the way of the user creating a registered account. Add a hidden prefix to the email address in the temporary account to avoid collision.

merge_with(other_id)

merge details from other user into self

Primary usage stems from different account registration flows. For example, users are created when invited by staff to participate, and when the same user later opts to register, a second account is generated during the registration process (either by flask-user or other mechanisms like add_user).

NB - caller MUST manage email due to unique constraints

notifications
observations
org_coding_display_options

Collates all race/ethnicity/indigenous display options from the user’s orgs to establish which options to display

organizations
password
password_verification_failures
phone
phone_id
practitioner_id
procedure_history(requestURL=None)
procedures
promote_to_registered(registered_user)

Promote a weakly authenticated account to a registered one

questionnaire_responses
races
reactivate_user(acting_user)

Reactivate a previously deleted user

This method clears the deleted status - by removing the link from the user to the audit recording the delete. Audit itself is retained for tracking purposes, and a new one will be created for posterity

Parameters:
  • self – user to reactivate
  • acting_user – individual executing the command, for audit trail
registered
relationships
reset_lockout()

resets variables that track lockout

We track when the user fails password verification to lockout users when they fail too many times. This function resets those variables

reset_password_token
rolelist

Generate UI friendly string of user’s roles by name

roles
save_observation(codeable_concept, value_quantity, audit, status, issued)

Helper method for creating new observations

staff_html()

Helper used from templates to display any custom staff/provider text

Interventions can add personalized HTML for care staff to consume on the /patients list. Look up any values for this user on all interventions.

subject_audits
timezone
update_birthdate(fhir)
update_consents(consent_list, acting_user)

Update user’s consents

Adds the provided list of consent agreements to the user. If the user had pre-existing consent agreements between the same organization_id, the new will replace the old

NB this will only modify/update consents between the user and the organizations named in the given consent_list.

update_deceased(fhir)
update_from_fhir(fhir, acting_user=None)

Update the user’s demographics from the given FHIR

If a field is defined, it is the final definition for the respective field, resulting in a deletion of existing values in said field that are not included.

Parameters:
  • fhir – JSON defining portions of the user demographics to change
  • acting_user – user requesting the change, used in audit logs
update_orgs(org_list, acting_user, excuse_top_check=False)

Update user’s organizations

Uses given list of organizations as the definitive list for the user - meaning any current affiliations not mentioned will be deleted.

Parameters:
  • org_list – list of organization objects for user’s orgs
  • acting_user – user behind the request for permission checks
  • excuse_top_check – Set True to excuse check for changes to top level orgs, say during initial account creation
update_roles(role_list, acting_user)

Update user’s roles

Parameters:
  • role_list – list of role objects defining exactly what roles the user should have. Any existing roles not mentioned will be deleted from user’s list
  • acting_user – user performing action, for permissions, etc.
user_audits
username
valid_consents

Access to consents that have neither been deleted or expired

class portal.models.user.UserEthnicity(**kwargs)
coding_id
id
user_id
class portal.models.user.UserEthnicityExtension(user, extension)
children
extension_url = 'http://hl7.org/fhir/StructureDefinition/us-core-ethnicity'
class portal.models.user.UserIndigenous(**kwargs)
coding_id
id
user_id
class portal.models.user.UserIndigenousStatusExtension(user, extension)
children
extension_url = 'http://us.truenth.org/fhir/StructureDefinition/AU-NHHD-METeOR-id-291036'
class portal.models.user.UserRace(**kwargs)
coding_id
id
user_id
class portal.models.user.UserRaceExtension(user, extension)
children
extension_url = 'http://hl7.org/fhir/StructureDefinition/us-core-race'
class portal.models.user.UserRelationship(**kwargs)

SQLAlchemy class for user_relationships table

Relationship is assumed to be ordered such that:
<user_id> has a <relationship.name> with <other_user_id>
as_json()

serialize the relationship - used to preserve service users

classmethod from_json(data)
id
other_user
other_user_id
relationship
relationship_id
update_from_json(data)
user
user_id
class portal.models.user.UserRoles(**kwargs)
id
role_id
user_id
portal.models.user.active_patients(include_test_role=False, include_deleted=False, require_orgs=None, require_interventions=None, disallow_interventions=None, filter_by_ids=None)

Build query for active patients, filtered as specified

Common query for active (not deleted) patients.

Parameters:
  • include_test_role – Set true to include users with test role
  • include_deleted – Set true to include deleted users
  • require_orgs – Provide list of organization IDs if patients must also have the respective UserOrganization association (different from consents!) Patients required to have at least one, not all orgs in given require_orgs list.
  • require_interventions – Provide list of intervention IDs if patients must also have the respective UserIntervention association. Patients required to have at least one, not all interventions in given require_interventions list.
  • disallow_interventions – Provide list of intervention IDs to exclude associated patients, such as the randomized control trial interventions.
  • filter_by_ids – List of user_ids to include in query filter
Returns:

Live SQLAlchemy Query, for further filter additions or execution

portal.models.user.add_role(user, role_name)
portal.models.user.add_user(user_info)

Given the result from an external IdP, create a new user

portal.models.user.current_user()

Obtain the “current” user object

Works for both remote oauth sessions and locally logged in sessions.

returns current user object, or None if not logged in (local or remote)

portal.models.user.default_email(context=None)

Function to provide a unique, default email if none is provided

Parameters:context – is populated by SQLAlchemy - see Context-Sensitive default functions in http://docs.sqlalchemy.org/en/latest/core/defaults.html
Returns:a unique email string to avoid unique constraints, if an email isn’t provided in the context
portal.models.user.flag_test()

Find all non-service users and flag as test

portal.models.user.get_user(uid)
portal.models.user.get_user_or_abort(uid, allow_deleted=False)

Wraps get_user and raises error if not found

Safe to call with path or parameter info. Confirms integer value before attempting lookup.

Parameters:
  • uid – integer value for user id to look up
  • allow_deleted – set true to allow access to deleted users

:raises werkzeug.exceptions.BadRequest: w/o a uid

:raises werkzeug.exceptions.NotFound: if the given uid isn’t
an integer, or if no matching user
:raises werkzeug.exceptions.Forbidden: if the named user has
been deleted, unless allow_deleted is set
Returns:user if valid and found
portal.models.user.permanently_delete_user(username, user_id=None, acting_user=None, actor=None)

Given a username (email), purge the user from the system

Includes wiping out audit rows, observations, etc. May pass either username or user_id. Will prompt for acting_user if not provided.

Parameters:
  • username – username (email) for user to purge
  • user_id – id of user in liew of username
  • acting_user – user taking the action, for record keeping
portal.models.user.user_extension_map(user, extension)

Map the given extension to the User

FHIR uses extensions for elements beyond base set defined. Lookup an adapter to handle the given extension for the user.

Parameters:
  • user – the user to apply to or read the extension from
  • extension – a dictionary with at least a ‘url’ key defining the extension. Should include a ‘valueCodeableConcept’ structure when being used in an apply context (i.e. direct FHIR data)
Returns:

adapter implementing apply_fhir and as_fhir methods

:raises exceptions.ValueError: if the extension isn’t recognized

portal.models.user.validate_email(email)

Not done at model level, as there are exceptions

We allow for placeholders and masks on email, so not all emails are valid. This validation function is generally only used when an end user changing an address or another use requires validation.

Furthermore, due to the complexity of valid email addresses, just look for some obvious signs - such as the ‘@’ symbol and at least 6 chars.

:raises werkzeug.exceptions.BadRequest: if obviously invalid