Module fusion_platform.models.organisation
Organisation model class file.
author: Matthew Casey
Classes
class Organisation (session, schema=None)-
Expand source code
class Organisation(Model): """ Organisation model class providing attributes and methods to manipulate organisation details. """ # Override the schema. _SCHEMA = OrganisationSchema() # Base path. _PATH_BASE = '/organisations/{organisation_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_DATA = f"{_PATH_BASE}/data/uploaded" _PATH_OWN_SERVICES = f"{_PATH_BASE}/services" _PATH_PROCESSES = f"{_PATH_BASE}/processes" _PATH_SERVICES = f"{_PATH_BASE}/services/latest" _PATH_DISPATCHERS = f"{_PATH_BASE}/services/dispatchers" def create_data(self, name, file_type, files, wait=False, **kwargs): """ Creates a data object for the organisation and uploads the corresponding files. Optionally waits for the upload and analysis to complete. Args: name: The name of the data object. file_type: The type of files that the data object will hold. files: The list of file paths to be uploaded. wait: Optionally wait for the upload and analysis to complete? Default False. kwargs: The model attributes which override those already in the template. Returns: The created data object. Raises: RequestError: if the create fails. ModelError: if the model could not be created and validated by the Fusion Platform<sup>®</sup>. """ # Get a new template for the data model. data = Data(self._session) data._new(organisation_id=self.id) # Now attempt to create the data item with any overriding keyword arguments. data._create(name, file_type, files, wait=wait, **kwargs) return data @property def credit(self): """ Returns: The credit model for the organisation. Raises: RequestError: if any get fails. ModelError: if a model could not be loaded or validated from the Fusion Platform<sup>®</sup>. """ return Credit._model_from_api_id(self._session, organisation_id=self.id, id=self.id) # Credit model id is the same as the organisation's id. @property def data(self): """ Provides an iterator through the organisation's uploaded data objects. Returns: An iterator through the data objects. Raises: RequestError: if any get fails. ModelError: if a model could not be loaded or validated from the Fusion Platform<sup>®</sup>. """ return Data._models_from_api_path(self._session, self._get_path(self.__class__._PATH_DATA)) def find_data(self, id=None, name=None, search=None): """ Searches for uploaded data objects with the specified id and/or (non-unique) name, returning the first object found and an iterator. Args: id: The data id to search for. name: The name to search for (case-sensitive). search: The term to search for (case-insensitive). Returns: The first found data object, or None if not found, and an iterator through the found data objects. Raises: RequestError: if any get fails. ModelError: if a model could not be loaded or validated from the Fusion Platform<sup>®</sup>. """ # Note that name is a key field, and hence we can only search using begins with. filter = self.__class__._build_filter( [(self.__class__._FIELD_ID, self.__class__._FILTER_MODIFIER_EQ, id), (self.__class__._FIELD_NAME, self.__class__._FILTER_MODIFIER_BEGINS_WITH, name)]) # Build the partial find generator and execute it. find = partial(Data._models_from_api_path, self._session, self._get_path(self.__class__._PATH_DATA), filter=filter, search=search) return self.__class__._first_and_generator(find) def find_dispatchers(self, id=None, ssd_id=None, name=None, keyword=None, search=None): """ Searches for dispatcher services with the specified id, SSD id, (non-unique) name and/or keywords, returning the first object found and an iterator. Dispatchers are specific services used by processes to dispatch outputs to particular destinations. These services should not be used to form processes themselves. Args: id: The service id to search for. ssd_id: The SSD id to search for. name: The name to search for (case-sensitive). keyword: The keyword to search for (case-sensitive). search: The term to search for (case-insensitive). Returns: The first found dispatcher service object, or None if not found, and an iterator through the found service objects. Raises: RequestError: if any get fails. ModelError: if a model could not be loaded or validated from the Fusion Platform<sup>®</sup>. """ return self.__find_services(self.__class__._PATH_DISPATCHERS, id, ssd_id, name, keyword, search) def find_processes(self, id=None, name=None, search=None): """ Searches for the organisation's processes with the specified id and/or (non-unique) name, returning the first object found and an iterator. Args: id: The process id to search for. name: The name to search for (case-sensitive). search: The term to search for (case-insensitive). Returns: The first found process object, or None if not found, and an iterator through the found process objects. Raises: RequestError: if any get fails. ModelError: if a model could not be loaded or validated from the Fusion Platform<sup>®</sup>. """ filter = self.__class__._build_filter( [(self.__class__._FIELD_ID, self.__class__._FILTER_MODIFIER_EQ, id), (self.__class__._FIELD_NAME, self.__class__._FILTER_MODIFIER_CONTAINS, name)]) # Build the partial find generator and execute it. find = partial(Process._models_from_api_path, self._session, self._get_path(self.__class__._PATH_PROCESSES), filter=filter, search=search) return self.__class__._first_and_generator(find) def __find_services(self, path, id=None, ssd_id=None, name=None, keyword=None, search=None): """ Searches for services with the specified id, SSD id, (non-unique) name and/or keywords, returning the first object found and an iterator. Args: path: The API path against which the search should be performed. id: The service id to search for. ssd_id: The SSD id to search for. name: The name to search for (case-sensitive). keyword: The keyword to search for (case-sensitive). search: The term to search for (case-insensitive). Returns: The first found service object, or None if not found, and an iterator through the found service objects. Raises: RequestError: if any get fails. ModelError: if a model could not be loaded or validated from the Fusion Platform<sup>®</sup>. """ # Note that name is a key field, and hence we can only search using begins with. filter = self.__class__._build_filter( [(self.__class__._FIELD_ID, self.__class__._FILTER_MODIFIER_EQ, id), (self.__class__._FIELD_SSD_ID, self.__class__._FILTER_MODIFIER_EQ, ssd_id), (self.__class__._FIELD_NAME, self.__class__._FILTER_MODIFIER_BEGINS_WITH, name), (self.__class__._FIELD_KEYWORDS, self.__class__._FILTER_MODIFIER_CONTAINS, keyword)]) # Build the partial find generator and execute it. find = partial(Service._models_from_api_path, self._session, self._get_path(path), filter=filter, search=search) return self.__class__._first_and_generator(find) def find_services(self, id=None, ssd_id=None, name=None, keyword=None, search=None): """ Searches for services with the specified id, SSD id, (non-unique) name and/or keywords, returning the first object found and an iterator. Args: id: The service id to search for. ssd_id: The SSD id to search for. name: The name to search for (case-sensitive). keyword: The keyword to search for (case-sensitive). search: The term to search for (case-insensitive). Returns: The first found service object, or None if not found, and an iterator through the found service objects. Raises: RequestError: if any get fails. ModelError: if a model could not be loaded or validated from the Fusion Platform<sup>®</sup>. """ return self.__find_services(self.__class__._PATH_SERVICES, id, ssd_id, name, keyword, search) def new_process(self, name, service): """ Creates a new template process from the service object. This process is not persisted to the Fusion Platform<sup>®</sup>. Args: name: The name of the process. service: The service for which the process is to be created. Returns: The new template process object. Raises: RequestError: if the new fails. ModelError: if the model could not be created and validated by the Fusion Platform<sup>®</sup>. """ # Get a new template for the process model using the service. process = Process(self._session) process._new(query_parameters={Model._get_id_name(Service.__name__): service.id}, organisation_id=self.id, name=name) return process @property def own_services(self): """ Provides an iterator through the services owned by the organisation. This includes services which may not yet be approved. Returns: An iterator through the service objects. Raises: RequestError: if any get fails. ModelError: if a model could not be loaded or validated from the Fusion Platform<sup>®</sup>. """ return Service._models_from_api_path(self._session, self._get_path(self.__class__._PATH_OWN_SERVICES)) @property def processes(self): """ Provides an iterator through the organisation's processes. Returns: An iterator through the process objects. Raises: RequestError if any get fails. ModelError if a model could not be loaded or validated from the Fusion Platform<sup>®</sup>. """ return Process._models_from_api_path(self._session, self._get_path(self.__class__._PATH_PROCESSES)) @property def services(self): """ Provides an iterator through the services available to the organisation for execution. Returns: An iterator through the service objects. Raises: RequestError: if any get fails. ModelError: if a model could not be loaded or validated from the Fusion Platform<sup>®</sup>. """ return Service._models_from_api_path(self._session, self._get_path(self.__class__._PATH_SERVICES))Organisation model class providing attributes and methods to manipulate organisation 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.attributesReturns
The model attributes as a dictionary.
prop credit-
Expand source code
@property def credit(self): """ Returns: The credit model for the organisation. Raises: RequestError: if any get fails. ModelError: if a model could not be loaded or validated from the Fusion Platform<sup>®</sup>. """ return Credit._model_from_api_id(self._session, organisation_id=self.id, id=self.id) # Credit model id is the same as the organisation's id.Returns
The credit model for the organisation.
Raises
RequestError- if any get fails.
ModelError- if a model could not be loaded or validated from the Fusion Platform®.
prop data-
Expand source code
@property def data(self): """ Provides an iterator through the organisation's uploaded data objects. Returns: An iterator through the data objects. Raises: RequestError: if any get fails. ModelError: if a model could not be loaded or validated from the Fusion Platform<sup>®</sup>. """ return Data._models_from_api_path(self._session, self._get_path(self.__class__._PATH_DATA))Provides an iterator through the organisation's uploaded data objects.
Returns
An iterator through the data objects.
Raises
RequestError- if any get fails.
ModelError- if a model could not be loaded or validated from the Fusion Platform®.
prop own_services-
Expand source code
@property def own_services(self): """ Provides an iterator through the services owned by the organisation. This includes services which may not yet be approved. Returns: An iterator through the service objects. Raises: RequestError: if any get fails. ModelError: if a model could not be loaded or validated from the Fusion Platform<sup>®</sup>. """ return Service._models_from_api_path(self._session, self._get_path(self.__class__._PATH_OWN_SERVICES))Provides an iterator through the services owned by the organisation. This includes services which may not yet be approved.
Returns
An iterator through the service objects.
Raises
RequestError- if any get fails.
ModelError- if a model could not be loaded or validated from the Fusion Platform®.
prop processes-
Expand source code
@property def processes(self): """ Provides an iterator through the organisation's processes. Returns: An iterator through the process objects. Raises: RequestError if any get fails. ModelError if a model could not be loaded or validated from the Fusion Platform<sup>®</sup>. """ return Process._models_from_api_path(self._session, self._get_path(self.__class__._PATH_PROCESSES))Provides an iterator through the organisation's processes.
Returns
An iterator through the process objects.
Raises
RequestError if any get fails. ModelError if a model could not be loaded or validated from the Fusion Platform®.
prop services-
Expand source code
@property def services(self): """ Provides an iterator through the services available to the organisation for execution. Returns: An iterator through the service objects. Raises: RequestError: if any get fails. ModelError: if a model could not be loaded or validated from the Fusion Platform<sup>®</sup>. """ return Service._models_from_api_path(self._session, self._get_path(self.__class__._PATH_SERVICES))Provides an iterator through the services available to the organisation for execution.
Returns
An iterator through the service objects.
Raises
RequestError- if any get fails.
ModelError- if a model could not be loaded or validated from the Fusion Platform®.
Methods
def create_data(self, name, file_type, files, wait=False, **kwargs)-
Expand source code
def create_data(self, name, file_type, files, wait=False, **kwargs): """ Creates a data object for the organisation and uploads the corresponding files. Optionally waits for the upload and analysis to complete. Args: name: The name of the data object. file_type: The type of files that the data object will hold. files: The list of file paths to be uploaded. wait: Optionally wait for the upload and analysis to complete? Default False. kwargs: The model attributes which override those already in the template. Returns: The created data object. Raises: RequestError: if the create fails. ModelError: if the model could not be created and validated by the Fusion Platform<sup>®</sup>. """ # Get a new template for the data model. data = Data(self._session) data._new(organisation_id=self.id) # Now attempt to create the data item with any overriding keyword arguments. data._create(name, file_type, files, wait=wait, **kwargs) return dataCreates a data object for the organisation and uploads the corresponding files. Optionally waits for the upload and analysis to complete.
Args
name- The name of the data object.
file_type- The type of files that the data object will hold.
files- The list of file paths to be uploaded.
wait- Optionally wait for the upload and analysis to complete? Default False.
kwargs- The model attributes which override those already in the template.
Returns
The created data object.
Raises
RequestError- if the create fails.
ModelError- if the model could not be created and validated by the Fusion Platform®.
def delete(self)-
Attempts to delete the model object. This assumes the model is deleted using a DELETE RESTful request …
def find_data(self, id=None, name=None, search=None)-
Expand source code
def find_data(self, id=None, name=None, search=None): """ Searches for uploaded data objects with the specified id and/or (non-unique) name, returning the first object found and an iterator. Args: id: The data id to search for. name: The name to search for (case-sensitive). search: The term to search for (case-insensitive). Returns: The first found data object, or None if not found, and an iterator through the found data objects. Raises: RequestError: if any get fails. ModelError: if a model could not be loaded or validated from the Fusion Platform<sup>®</sup>. """ # Note that name is a key field, and hence we can only search using begins with. filter = self.__class__._build_filter( [(self.__class__._FIELD_ID, self.__class__._FILTER_MODIFIER_EQ, id), (self.__class__._FIELD_NAME, self.__class__._FILTER_MODIFIER_BEGINS_WITH, name)]) # Build the partial find generator and execute it. find = partial(Data._models_from_api_path, self._session, self._get_path(self.__class__._PATH_DATA), filter=filter, search=search) return self.__class__._first_and_generator(find)Searches for uploaded data objects with the specified id and/or (non-unique) name, returning the first object found and an iterator.
Args
id- The data id to search for.
name- The name to search for (case-sensitive).
search- The term to search for (case-insensitive).
Returns
The first found data object, or None if not found, and an iterator through the found data objects.
Raises
RequestError- if any get fails.
ModelError- if a model could not be loaded or validated from the Fusion Platform®.
def find_dispatchers(self, id=None, ssd_id=None, name=None, keyword=None, search=None)-
Expand source code
def find_dispatchers(self, id=None, ssd_id=None, name=None, keyword=None, search=None): """ Searches for dispatcher services with the specified id, SSD id, (non-unique) name and/or keywords, returning the first object found and an iterator. Dispatchers are specific services used by processes to dispatch outputs to particular destinations. These services should not be used to form processes themselves. Args: id: The service id to search for. ssd_id: The SSD id to search for. name: The name to search for (case-sensitive). keyword: The keyword to search for (case-sensitive). search: The term to search for (case-insensitive). Returns: The first found dispatcher service object, or None if not found, and an iterator through the found service objects. Raises: RequestError: if any get fails. ModelError: if a model could not be loaded or validated from the Fusion Platform<sup>®</sup>. """ return self.__find_services(self.__class__._PATH_DISPATCHERS, id, ssd_id, name, keyword, search)Searches for dispatcher services with the specified id, SSD id, (non-unique) name and/or keywords, returning the first object found and an iterator. Dispatchers are specific services used by processes to dispatch outputs to particular destinations. These services should not be used to form processes themselves.
Args
id- The service id to search for.
ssd_id- The SSD id to search for.
name- The name to search for (case-sensitive).
keyword- The keyword to search for (case-sensitive).
search- The term to search for (case-insensitive).
Returns
The first found dispatcher service object, or None if not found, and an iterator through the found service objects.
Raises
RequestError- if any get fails.
ModelError- if a model could not be loaded or validated from the Fusion Platform®.
def find_processes(self, id=None, name=None, search=None)-
Expand source code
def find_processes(self, id=None, name=None, search=None): """ Searches for the organisation's processes with the specified id and/or (non-unique) name, returning the first object found and an iterator. Args: id: The process id to search for. name: The name to search for (case-sensitive). search: The term to search for (case-insensitive). Returns: The first found process object, or None if not found, and an iterator through the found process objects. Raises: RequestError: if any get fails. ModelError: if a model could not be loaded or validated from the Fusion Platform<sup>®</sup>. """ filter = self.__class__._build_filter( [(self.__class__._FIELD_ID, self.__class__._FILTER_MODIFIER_EQ, id), (self.__class__._FIELD_NAME, self.__class__._FILTER_MODIFIER_CONTAINS, name)]) # Build the partial find generator and execute it. find = partial(Process._models_from_api_path, self._session, self._get_path(self.__class__._PATH_PROCESSES), filter=filter, search=search) return self.__class__._first_and_generator(find)Searches for the organisation's processes with the specified id and/or (non-unique) name, returning the first object found and an iterator.
Args
id- The process id to search for.
name- The name to search for (case-sensitive).
search- The term to search for (case-insensitive).
Returns
The first found process object, or None if not found, and an iterator through the found process objects.
Raises
RequestError- if any get fails.
ModelError- if a model could not be loaded or validated from the Fusion Platform®.
def find_services(self, id=None, ssd_id=None, name=None, keyword=None, search=None)-
Expand source code
def find_services(self, id=None, ssd_id=None, name=None, keyword=None, search=None): """ Searches for services with the specified id, SSD id, (non-unique) name and/or keywords, returning the first object found and an iterator. Args: id: The service id to search for. ssd_id: The SSD id to search for. name: The name to search for (case-sensitive). keyword: The keyword to search for (case-sensitive). search: The term to search for (case-insensitive). Returns: The first found service object, or None if not found, and an iterator through the found service objects. Raises: RequestError: if any get fails. ModelError: if a model could not be loaded or validated from the Fusion Platform<sup>®</sup>. """ return self.__find_services(self.__class__._PATH_SERVICES, id, ssd_id, name, keyword, search)Searches for services with the specified id, SSD id, (non-unique) name and/or keywords, returning the first object found and an iterator.
Args
id- The service id to search for.
ssd_id- The SSD id to search for.
name- The name to search for (case-sensitive).
keyword- The keyword to search for (case-sensitive).
search- The term to search for (case-insensitive).
Returns
The first found service object, or None if not found, and an iterator through the found service 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)-
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 new_process(self, name, service)-
Expand source code
def new_process(self, name, service): """ Creates a new template process from the service object. This process is not persisted to the Fusion Platform<sup>®</sup>. Args: name: The name of the process. service: The service for which the process is to be created. Returns: The new template process object. Raises: RequestError: if the new fails. ModelError: if the model could not be created and validated by the Fusion Platform<sup>®</sup>. """ # Get a new template for the process model using the service. process = Process(self._session) process._new(query_parameters={Model._get_id_name(Service.__name__): service.id}, organisation_id=self.id, name=name) return processCreates a new template process from the service object. This process is not persisted to the Fusion Platform®.
Args
name- The name of the process.
service- The service for which the process is to be created.
Returns
The new template process object.
Raises
RequestError- if the new fails.
ModelError- if the model could not be created and validated by the Fusion Platform®.
def to_csv(self, exclude=None)-
Converts the model attributes into a CSV string …
def update(self, **kwargs)-
Attempts to update the model object with the given values. For models which have not been persisted, the relevant fields are updated without …
class OrganisationSchema (*,
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 OrganisationSchema(Schema): """ Schema class for organisation model. Each organisation model has the following fields (and nested fields): .. include::organisation.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. name = fields.String(required=True) address_line_1 = fields.String(required=True) address_line_2 = fields.String(allow_none=True) address_town_city = fields.String(required=True) address_post_zip_code = fields.String(required=True) address_country = fields.String(required=True) payment_customer = fields.String(allow_none=True) payment_valid = fields.Boolean(allow_none=True, metadata={'read_only': True}) # Changed to prevent this being updated and optional. # Removed payment_last_checked. income_customer = fields.String(allow_none=True) income_valid = fields.Boolean(allow_none=True, metadata={'read_only': True}) # Changed to prevent this being updated and optional. # Removed income_last_checked. income_tax_rate = fields.Decimal(allow_none=True) income_tax_reference = fields.String(allow_none=True) currency = fields.String(allow_none=True) # Changed to optional. users = fields.List(fields.Nested(OrganisationUserSchema()), allow_none=True, metadata={'read_only': True}) # Changed to prevent this being updated. agreed_licenses = fields.List(fields.UUID(required=True), allow_none=True, metadata={'read_only': True}) # Changed to prevent this being updated. offers = fields.List(fields.UUID(required=True), allow_none=True, metadata={'read_only': True}) # Changed to prevent this being updated. # Removed runtime_stop_charge. # Removed runtime_error_charge. # Removed audit_services. maximum_output_storage_period = fields.Integer(allow_none=True, metadata={'read_only': True}) # Changed to prevent this being updated. maximum_file_downloads = fields.Integer(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 = EXCLUDESchema class for organisation model.
Each organisation 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?
- name: The unique name of the organisation
- address_line_1: The first line of the address.
- address_line_2: The second line of the address.
- address_town_city: The town or city.
- address_post_zip_code: The postal or zip code.
- address_country: The country.
- payment_customer: The payment customer reference from the payment service.
- payment_valid: Are the payment service details valid?
- income_customer: The income customer reference from the income service.
- income_valid: Are the income service details valid?
- income_tax_rate: The percentage tax to be applied to income.
- income_tax_reference: The income tax reference.
- currency: The currency to be used for payment and income.
- users: The list of users linked to this organisation.
- id: The user identifier.
- email: The user's email address.
- roles: Assigned organisational roles.
- agreed_licenses: The list of licences agreed on behalf of the organisation.
- offers: The list of offers linked to this organisation.
- maximum_output_storage_period: The maximum period in days for this organisation for which an output can be stored before it is deleted.
- maximum_file_downloads: The maximum number of time any file can be downloaded for this organisation.
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.