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.