Module fusion_platform.models.user

User model class file.

author: Matthew Casey

© Digital Content Analysis Technology Ltd

Classes

class User (session, schema=None)
Expand source code
class User(Model):
    """
    User model class providing attributes and methods to manipulate user details.
    """

    # Override the schema.
    _SCHEMA = UserSchema()

    # Base path.
    _PATH_BASE = '/users/{user_id}'

    # Override the standard model paths.
    _PATH_DELETE = _PATH_BASE
    _PATH_GET = _PATH_BASE
    _PATH_PATCH = _PATH_BASE

    # Add in the custom model paths.
    _PATH_CHANGE_PASSWORD = f"{_PATH_BASE}/change_password"

    def change_password(self, old, new):
        """
        Changes the user password. The new password must conform to the current password policy.

        Args:
            old: The old password.
            new: The new password.

        Raises:
            RequestError: if the update fails.
        """
        body = {self.__class__.__name__: {'old_password': old, 'new_password': new}}
        self._session.request(path=self._get_path(self.__class__._PATH_CHANGE_PASSWORD), method=Session.METHOD_POST, body=body)

    def find_organisations(self, id=None, name=None):
        """
        Searches for organisations with the specified id and/or (non-unique) name, returning the first object found and an iterator.

        Args:
            id: The organisation id to search for.
            name: The name to search for (case-sensitive).

        Returns:
            The first found organisation object, or None if not found, and an iterator through the found organisation objects.

        Raises:
            RequestError: if any get fails.
            ModelError: if a model could not be loaded or validated from the Fusion Platform<sup>&reg;</sup>.
        """
        # Search for the id and/or name.
        organisation = None

        for item in self.organisations:
            if ((id is not None) and (str(id).lower() == str(item.id).lower())) or ((name is not None) and (item.name.lower().startswith(name.lower()))):
                organisation = item
                break

        return organisation, self.organisations

    @property
    def organisations(self):
        """
        Provides an iterator through the organisations which the user belongs to.

        Returns:
            An iterator through the organisations.

        Raises:
            RequestError: if any get fails.
            ModelError: if a model could not be loaded or validated from the Fusion Platform<sup>&reg;</sup>.
        """
        return Organisation._models_from_api_ids(self._session, [{self.__class__._FIELD_ID: organisation.get(self.__class__._FIELD_ID)} for organisation in
                                                                 self._model.get('organisations', [])])

User model class providing attributes and methods to manipulate user details.

Initialises the object.

Args

session
The linked session object for interfacing with the Fusion Platform®.
schema
The optional schema to use instead of the schema defined by the class.

Ancestors

  • Model
  • fusion_platform.base.Base

Instance variables

prop attributes

Inherited from: Model.attributes

Returns

The model attributes as a dictionary.

prop organisations
Expand source code
@property
def organisations(self):
    """
    Provides an iterator through the organisations which the user belongs to.

    Returns:
        An iterator through the organisations.

    Raises:
        RequestError: if any get fails.
        ModelError: if a model could not be loaded or validated from the Fusion Platform<sup>&reg;</sup>.
    """
    return Organisation._models_from_api_ids(self._session, [{self.__class__._FIELD_ID: organisation.get(self.__class__._FIELD_ID)} for organisation in
                                                             self._model.get('organisations', [])])

Provides an iterator through the organisations which the user belongs to.

Returns

An iterator through the organisations.

Raises

RequestError
if any get fails.
ModelError
if a model could not be loaded or validated from the Fusion Platform®.

Methods

def change_password(self, old, new)
Expand source code
def change_password(self, old, new):
    """
    Changes the user password. The new password must conform to the current password policy.

    Args:
        old: The old password.
        new: The new password.

    Raises:
        RequestError: if the update fails.
    """
    body = {self.__class__.__name__: {'old_password': old, 'new_password': new}}
    self._session.request(path=self._get_path(self.__class__._PATH_CHANGE_PASSWORD), method=Session.METHOD_POST, body=body)

Changes the user password. The new password must conform to the current password policy.

Args

old
The old password.
new
The new password.

Raises

RequestError
if the update fails.
def delete(self)

Inherited from: Model.delete

Attempts to delete the model object. This assumes the model is deleted using a DELETE RESTful request …

def find_organisations(self, id=None, name=None)
Expand source code
def find_organisations(self, id=None, name=None):
    """
    Searches for organisations with the specified id and/or (non-unique) name, returning the first object found and an iterator.

    Args:
        id: The organisation id to search for.
        name: The name to search for (case-sensitive).

    Returns:
        The first found organisation object, or None if not found, and an iterator through the found organisation objects.

    Raises:
        RequestError: if any get fails.
        ModelError: if a model could not be loaded or validated from the Fusion Platform<sup>&reg;</sup>.
    """
    # Search for the id and/or name.
    organisation = None

    for item in self.organisations:
        if ((id is not None) and (str(id).lower() == str(item.id).lower())) or ((name is not None) and (item.name.lower().startswith(name.lower()))):
            organisation = item
            break

    return organisation, self.organisations

Searches for organisations with the specified id and/or (non-unique) name, returning the first object found and an iterator.

Args

id
The organisation id to search for.
name
The name to search for (case-sensitive).

Returns

The first found organisation object, or None if not found, and an iterator through the found organisation objects.

Raises

RequestError
if any get fails.
ModelError
if a model could not be loaded or validated from the Fusion Platform®.
def get(self, **kwargs)

Inherited from: Model.get

Gets the model object by loading it from the Fusion Platform®. Uses the model's current id and base model id for the get unless …

def to_csv(self, exclude=None)

Inherited from: Model.to_csv

Converts the model attributes into a CSV string …

def update(self, **kwargs)

Inherited from: Model.update

Attempts to update the model object with the given values. For models which have not been persisted, the relevant fields are updated without …

class UserSchema (*,
only: types.StrSequenceOrSet | None = None,
exclude: types.StrSequenceOrSet = (),
many: bool | None = None,
load_only: types.StrSequenceOrSet = (),
dump_only: types.StrSequenceOrSet = (),
partial: bool | types.StrSequenceOrSet | None = None,
unknown: types.UnknownOption | None = None)
Expand source code
class UserSchema(Schema):
    """
    Schema class for user model.

    Each user model has the following fields (and nested fields):

    .. include::user.md
    """
    id = fields.UUID(required=True, metadata={'read_only': True})  # Changed to prevent this being updated.

    created_at = fields.DateTime(required=True, metadata={'read_only': True})  # Changed to prevent this being updated.
    updated_at = fields.DateTime(required=True, metadata={'read_only': True})  # Changed to prevent this being updated.

    email = fields.Email(required=True, metadata={'read_only': True})  # Changed to prevent this being updated.
    # Removed email_verified.
    given_name = fields.String(required=True)
    family_name = fields.String(required=True)
    telephone = fields.String(allow_none=True)

    role = fields.Integer(allow_none=True)  # Changed to optional.

    # Removed location_ip_address.
    # Removed location_ip_country.

    default_output_storage_period = fields.Integer(required=True)

    time_zone = fields.String(required=True)
    locale = fields.String(required=True)
    units = fields.String(required=True)

    notification_service_status = fields.List(fields.String(required=True), allow_none=True)
    notification_account = fields.List(fields.String(required=True), allow_none=True)
    notification_processing_chain = fields.List(fields.String(required=True), allow_none=True)
    notification_new_features = fields.List(fields.String(required=True), allow_none=True)
    notification_contact = fields.List(fields.String(required=True), allow_none=True)

    organisations = fields.List(fields.Nested(UserOrganisationSchema()), allow_none=True, metadata={'hide': True})  # Changed to hide as an attribute.

    # Removed various flags.

    last_request_at = fields.DateTime(allow_none=True, metadata={'read_only': True})  # Changed to prevent this being updated.

    # Removed search.

    class Meta:
        """
        When loading an object, make sure we exclude any unknown fields, rather than raising an exception, and put fields in their definition order.
        """
        unknown = EXCLUDE

Schema class for user model.

Each user model has the following fields (and nested fields):

  • id: The unique identifier for the record.
  • created_at: When was the record created?
  • updated_at: When was the record last updated?
  • email: Current email address.
  • given_name: Given name.
  • family_name: Family name.
  • telephone: Telephone number - typically a mobile to receive text messages.
  • role: Assigned user role.
  • default_output_storage_period: The default number of days for which outputs are stored.
  • time_zone: Time zone.
  • locale: Locale.
  • units: Preferred measurement units.
  • notification_service_status: How will notifications will be sent for service status updates?
  • notification_account: How will notifications will be sent for account updates?
  • notification_processing_chain: How will notifications will be sent for processing chain execution?
  • notification_new_features: How will notifications will be sent for new features?
  • notification_contact: How will notifications will be sent for contact requests?
  • last_request_at: The last time a request was made.

Ancestors

  • marshmallow.schema.Schema

Class variables

var OPTIONS_CLASS : type

Defines defaults for marshmallow.Schema.Meta.

var TYPE_MAPPING : dict[type, type[Field]]

The type of the None singleton.

var dict_class : type[dict]

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

var error_messages : dict[str, str]

The type of the None singleton.