Skip to content
Snippets Groups Projects

Add type hints to models.py

Merged Henrik tom Wörden requested to merge f-type-hints into dev
1 unresolved thread
+ 468
174
@@ -45,10 +45,10 @@ from os import listdir
from os.path import isdir
from random import randint
from tempfile import NamedTemporaryFile
from typing import Any, Optional
from typing import Any, Literal, Optional, Type, Union
from warnings import warn
from caosdb.common.datatype import (BOOLEAN, DATETIME, DOUBLE, INTEGER, TEXT,
from caosdb.common.datatype import (BOOLEAN, DATETIME, DOUBLE, INTEGER, TEXT, REFERENCE,
is_list_datatype, is_reference)
from caosdb.common.state import State
from caosdb.common.timezone import TimeZone
@@ -66,17 +66,19 @@ from caosdb.exceptions import (AmbiguousEntityError, AuthorizationError,
TransactionError, UniqueNamesError,
UnqualifiedParentsError,
UnqualifiedPropertiesError)
from lxml import etree
from lxml import etree # type: ignore
_ENTITY_URI_SEGMENT = "Entity"
# importances/inheritance
OBLIGATORY = "OBLIGATORY"
SUGGESTED = "SUGGESTED"
RECOMMENDED = "RECOMMENDED"
FIX = "FIX"
ALL = "ALL"
NONE = "NONE"
OBLIGATORY: Literal["OBLIGATORY"] = "OBLIGATORY"
SUGGESTED: Literal["SUGGESTED"] = "SUGGESTED"
RECOMMENDED: Literal["RECOMMENDED"] = "RECOMMENDED"
FIX: Literal["FIX"] = "FIX"
ALL: Literal["ALL"] = "ALL"
NONE: Literal["NONE"] = "NONE"
TA_add_properties = Union[Literal['ALL', 'FIX', 'NONE'], None]
SPECIAL_ATTRIBUTES = ["name", "role", "datatype", "description",
@@ -96,8 +98,9 @@ class Entity:
by the user to control several server-side plug-ins.
"""
def __init__(self, name=None, id=None, description=None, # @ReservedAssignment
datatype=None, value=None, **kwargs):
def __init__(self, name: Optional[str] = None, id: Optional[int] = None,
description: Optional[str] = None, # @ReservedAssignment
datatype: Optional[str] = None, value=None, **kwargs):
self.__role = kwargs["role"] if "role" in kwargs else None
self._checksum = None
self._size = None
@@ -107,7 +110,7 @@ class Entity:
self._wrapped_entity = None
self._version = None
self._cuid = None
self._flags = dict()
self._flags: dict = dict()
self.__value = None
self.__datatype = None
self.datatype = datatype
@@ -241,6 +244,10 @@ class Entity:
self.__value = _parse_value(new_type, self.__value)
self.__datatype = new_type
@property
def checksum(self):
return self._checksum
@property
def description(self):
if self.__description is not None or self._wrapped_entity is None:
@@ -248,14 +255,14 @@ class Entity:
return self._wrapped_entity.description
@property
def checksum(self):
return self._checksum
@description.setter
def description(self, new_description):
self.__description = new_description
@property
def checksum(self):
return self._checksum
@property
def unit(self):
if self.__unit is not None or self._wrapped_entity is None:
@@ -517,8 +524,12 @@ class Entity:
return self
def add_property(self, property=None, value=None, id=None, name=None, description=None, datatype=None,
unit=None, importance=None, inheritance=None): # @ReservedAssignment
def add_property(self, property: Union[int, str, Entity, None] = None,
value: Union[str, int, Entity, None] = None,
id: Optional[int] = None, name: Optional[str] = None,
description: Optional[str] = None, datatype: Optional[str] = None,
unit: Optional[str] = None, importance: Optional[str] = None,
inheritance: Optional[str] = None) -> Entity: # @ReservedAssignment
"""Add a property to this entity.
The first parameter is meant to identify the property entity either via
@@ -536,35 +547,41 @@ class Entity:
Parameters
----------
property : int, str, Entity, optional
property : Union[int, str, Entity, None], optional
An identifier for the property to be added, either its name, its id,
or the corresponding Entity Python object. If ``None``, either the
`name` or the `id` argument have to be specified explicitly. Default
is ``None``.
value : int, str, bool, datetime, Entity, or list of these types, optional
value : Union[str, int, Entity, bool,list[str], list[int], list[Entity], list[bool]], optional
The value of the new property. In case of a reference to another
entity, this value may be the referenced entities id or the
``Entity`` as a Python object. Default is None.
id : int, optional
id : Optional[int], optional
Id of the property, by default None
name : str, optional
name : Optional[str], optional
Name of the property, by default None
description : str, optional
description : Optional[str], optional
Description of the property, by default None
datatype : str, optional
datatype : Optional[str], optional
Datatype of the property, by default None
unit : str, optional
unit : Optional[str], optional
Unit of the property, by default None
importance :str, optional
importance : Optional[str], optional
Importance of the property, by default None
inheritance : str, optional
inheritance : Optional[str], optional
Inheritance of the property, by default None
Returns
-------
Entity
This Entity object to which the new property has been added.
Raises
------
ValueError
Unusual Property role
Warns
-----
UserWarning
@@ -666,17 +683,29 @@ class Entity:
return self
def add_message(self, msg=None, type=None, code=None, # @ReservedAssignment
description=None, body=None):
def add_message(self, msg: Optional[Message] = None, type: Union[str, int, None] = None,
code: Union[str, int, None] = None, # @ReservedAssignment
description: Optional[str] = None,
body: Optional[str] = None) -> Entity:
"""Add a message (msg) to this entity. If and only if no msg is given
this method will created a new message from the parameters type, code,
description, and body.
@param msg: The message to be added to this entity.
@param type: The type of the message to be added.
@param code: The code of the message to be added.
@param description: The description of the message to be added.
@param body: The body of the message to be added.
Parameters
----------
msg : Message, optional
The message to be added to this entity, by default None
type : str, int, optional
The type of the message to be added, by default None
code : str, int, optional
The code of the message to be added, by default None
body : str, optional
The body of the message to be added, by default None
Returns
-------
Entity
The entity passed
"""
if msg is not None:
@@ -688,26 +717,33 @@ class Entity:
return self
def add_parent(self, parent=None, id=None, name=None, inheritance=None): # @ReservedAssignment
def add_parent(self, parent: Union[Entity, int, str, None] = None,
id: Optional[int] = None, name: Optional[str] = None,
inheritance: Optional[str] = None) -> Entity: # @ReservedAssignment
"""Add a parent to this entity.
Parameters
----------
parent : Entity or int or str or None
parent : Union[Entity, int, str, None], optional
The parent entity, either specified by the Entity object
itself, or its id or its name. Default is None.
id : int
itself, or its id or its name, by default None
id : Optional[int], optional
Integer id of the parent entity. Ignored if `parent`
is not None.
name : str
is not None, by default None
name : Optional[str], optional
Name of the parent entity. Ignored if `parent is not
none`.
inheritance : str
none`, by default None
inheritance : Optional[str], optional
One of ``obligatory``, ``recommended``, ``suggested``, or ``fix``. Specifies the
minimum importance which parent properties need to have to be inherited by this
entity. If no `inheritance` is given, no properties will be inherited by the child.
This parameter is case-insensitive.
Returns
-------
Entity
The entity passed
Notes
-----
Note that the behaviour of the `inheritance` argument currently has not
@@ -721,7 +757,6 @@ class Entity:
UserWarning
If neither a `parent` parameter, nor the `id`, nor `name`
parameter is passed to this method.
"""
pid = id
@@ -746,7 +781,7 @@ class Entity:
return self
def has_parent(self, parent: Entity, recursive: bool = True, retrieve: bool = True,
def has_parent(self, parent: Union[Entity, str, int], recursive: bool = True, retrieve: bool = True,
check_name: bool = True, check_id: bool = False):
"""Check if this entity has a given parent.
@@ -755,30 +790,29 @@ class Entity:
check. Note that, if checked, name or ID should not be None,
lest the check fail.
Parameters
----------
parent: Entity
Check for this parent.
Parameters
----------
recursive: bool, optional
Whether to check recursively.
parent: Entity, str, int
Check for this parent.
check_name: bool, optional
Whether to use the name for ancestry check.
recursive: bool, optional
Whether to check recursively.
check_id: bool, optional
Whether to use the ID for ancestry check.
check_name: bool, optional
Whether to use the name for ancestry check.
retrieve: bool, optional
If False, do not retrieve parents from the server.
check_id: bool, optional
Whether to use the ID for ancestry check.
Returns
-------
out: bool
True if ``parent`` is a true parent, False otherwise.
"""
retrieve: bool, optional
If False, do not retrieve parents from the server.
Returns
-------
out: bool
True if ``parent`` is a true parent, False otherwise.
"""
if recursive:
parents = self.get_parents_recursively(retrieve=retrieve)
else:
@@ -790,23 +824,30 @@ out: bool
if not (check_name or check_id):
return parent in parents
name_result = (
not check_name or
(parent.name is not None and
parent.name in [pp.name for pp in parents]))
id_result = (
not check_id or
(parent.id is not None and
parent.id in [pp.id for pp in parents]))
parent_name = getattr(parent, "name", None)
parent_id = getattr(parent, "id", None)
name_result: bool = False
id_result: bool = False
if parent_name:
name_result = (
not check_name or
(parent_name is not None and
parent_name in [pp.name for pp in parents]))
if parent_id:
id_result = (
not check_id or
(parent_id is not None and
parent_id in [pp.id for pp in parents]))
return name_result and id_result
def get_parents(self):
def get_parents(self) -> list:
"""Get all parents of this entity.
@return: _ParentList(list)
Returns
-------
list[ Parents]
"""
return self.parents
def get_parents_recursively(self, retrieve: bool = True):
@@ -832,14 +873,13 @@ out: List[Entity]
def _get_parent_recursively(self, all_parents: list, retrieve: bool = True):
"""Get all ancestors with a little helper.
As a side effect of this method, the ancestors are added to
all_parents.
Important: As a side effect of this method, the ancestors are
added to all_parents.
@param all_parents: list, The added parents so far.
@return: None, but see side effects.
"""
for parent in self.parents:
# TODO:
# Comment on _wrap and _wrapped_entity
@@ -860,24 +900,22 @@ out: List[Entity]
all_parents.append(w_parent)
w_parent._get_parent_recursively(all_parents, retrieve=retrieve)
def get_parent(self, key):
def get_parent(self, key: Union[Entity, int, str]) -> Optional[Entity]:
"""Return the first parent matching the key or None if no match exists.
Parameters
---------
key : int or Enity or str
----------
key : Entity, int, str
The id, Entity, or name of the parent that should be
returned. If an Entity is given, its id or its name is
used to find a matching parent.
Returns
-------
parent : Entity
Entity
The first parent of this entity that matches the given id,
entity, or name.
"""
if isinstance(key, int):
for p in self.parents:
if p.id is not None and int(p.id) == int(key):
@@ -901,41 +939,49 @@ out: List[Entity]
return None
def get_properties(self):
def get_properties(self) -> list:
"""Get all properties of this entity.
@return: _Properties(list)
Returns
-------
list
_Properties(list)
"""
return self.properties
def get_property(self, pattern):
""" Return the first matching property or None.
def get_property(self, pattern: Union[Entity, str, int]) -> Optional[Property]:
"""Return the first matching property or None.
Parameters
----------
pattern : str or int or Entity
pattern : Entity, str, int
The name or id to look for (case-insensitive) or an Entity where
the name or id is used to match the properites of this instance.
Returns
-------
property : Property
Property
The first Property of this Entity with a matching name or id.
Raises
------
ValueError
Wrong input type.
"""
# entity given
if (hasattr(pattern, "name") or hasattr(pattern, "id")):
# entity given
pattern_name = getattr(pattern, "name", None)
pattern_id = getattr(pattern, "id", None)
if pattern_name or pattern_id:
# only return if a result was found, otherwise use id
if (hasattr(pattern, "name") and pattern.name is not None
and self.get_property(pattern.name) is not None):
if (pattern_name is not None
and self.get_property(pattern_name) is not None):
return self.get_property(pattern.name)
return self.get_property(pattern_name)
if hasattr(pattern, "id") and pattern.id is not None:
return self.get_property(pattern.id)
if pattern_id is not None:
return self.get_property(pattern_id)
# int given
elif isinstance(pattern, int):
@@ -954,15 +1000,24 @@ out: List[Entity]
return None
def _get_value_for_selector(self, selector):
"""return the value described by the selector
def _get_value_for_selector(self, selector: Union[tuple, list, str]) -> Any:
"""Return the value described by the selector
Parameters
----------
selector : tuple, list, str
A selector is a list or a tuple of strings describing a path in an
entity tree with self as root. The last selector may be a special one
like unit or name.
A selector is a list or a tuple of strings describing a path in an
entity tree with self as root. The last selector may be a special one
like unit or name.
Returns
-------
Any
The value described by the selector.
See also get_property_values()
"""
SPECIAL_SELECTORS = ["unit", "value", "description", "id", "name"]
if not isinstance(selector, (tuple, list)):
@@ -1010,13 +1065,12 @@ out: List[Entity]
# if we saved a special selector before, apply it
if special_selector is None:
return prop.value
return prop.value # type: ignore[union-attr]
else:
return getattr(ref, special_selector.lower())
def get_property_values(self, *selectors):
""" Return a tuple with the values described by the given selectors.
def get_property_values(self, *selectors: Union[str, tuple, list]) -> tuple:
"""Return a tuple with the values described by the given selectors.
This represents an entity's properties as if it was a row of a table
with the given columns.
@@ -1034,16 +1088,16 @@ out: List[Entity]
Parameters
----------
*selectors : str or tuple of str
*selectors : str, tuple, list
Each selector is a list or tuple of property names, e.g. `"height",
"width"`.
Returns
-------
row : tuple
tuple
A row-like representation of the entity's properties.
"""
row = tuple()
row: tuple = tuple()
for selector in selectors:
val = self._get_value_for_selector(selector)
@@ -1054,18 +1108,20 @@ out: List[Entity]
return row
def get_messages(self):
def get_messages(self) -> dict:
"""Get all messages of this entity.
@return: Messages(list)
"""
return self.messages
def get_warnings(self):
def get_warnings(self) -> dict:
"""Get all warning messages of this entity.
@return Messages(list): Warning messages.
Returns
-------
list
_Messages(list): Warning messages.
"""
ret = Messages()
@@ -1075,10 +1131,13 @@ out: List[Entity]
return ret
def get_errors(self):
def get_errors(self) -> dict:
"""Get all error messages of this entity.
@return Messages(list): Error messages.
Returns
-------
list
_Messages(list): Error messages.
"""
ret = Messages()
@@ -1091,12 +1150,21 @@ out: List[Entity]
return ret
def get_errors_deep(self, roots=None):
def get_errors_deep(self, roots: Union[Entity, list, None] = None) -> list:
"""Get all error messages of this entity and all sub-entities /
parents / properties.
@return A list of tuples. Tuple index 0 contains the error message
and tuple index 1 contains the tree.
Parameters
----------
roots : Entity, optional
Used to find sub-entities / parents / properties recursivly,
by default None
Returns
-------
list
Tuple index 0 contains the error message
and tuple index 1 contains the tree.
"""
roots = [] if roots is None else roots
result_list = list()
@@ -1111,35 +1179,53 @@ out: List[Entity]
return result_list
def has_errors(self):
'''
@return True: if and only if this entities has any error messages.
'''
def has_errors(self) -> bool:
"""Check if this entity has any error message.
Returns
-------
bool
If and only if this entities has any error messages.
"""
for m in self.messages:
if m.type.lower() == "error":
return True
return False
def to_xml(self, xml=None, add_properties=ALL, local_serialization=False):
"""Generate an xml representation of this entity. If the parameter xml
def to_xml(self, xml: Optional[etree._Element] = None,
add_properties: TA_add_properties = ALL,
local_serialization: bool = False) -> etree._Element:
"""Generate an xml representation of this entity. If the parameter `xml`
is given, all attributes, parents, properties, and messages of this
entity will be added to it instead of creating a new element.
Raise an error if xml is not a lxml.etree.Element
Parameters
----------
xml : etree._Element, optional
an xml element to which all attributes, parents,
properties, and messages are to be added, by default None
add_properties : 'ALL', 'FIX', 'NONE', optional
By default ALL
local_serialization : bool, optional
For file handling, by default False
Returns
-------
etree._Element
xml representation of this entity.
@param xml: an xml element to which all attributes, parents,
properties, and messages
are to be added.
@return: xml representation of this entity.
Raises
------
TypeError
Raise an error if `xml` is not a lxml.etree.Element
"""
if xml is None:
# use role as xml tag name, fall-back to "Entity"
elem_tag = "Entity" if self.role is None else self.role
xml = etree.Element(elem_tag)
assert isinstance(xml, etree._Element)
if not isinstance(xml, etree._Element):
raise TypeError("xml has to be a lxml.etree.Element")
# unwrap wrapped entity
@@ -1250,13 +1336,26 @@ out: List[Entity]
return xml
@staticmethod
def _from_xml(entity, elem):
def _from_xml(entity: Any, elem: etree._Element) -> Entity:
"""Parse a single string representation of an xml element to an entity.
@param entity: the entity
@param elem: the xml element
"""
Parameters
----------
entity : Entity
The entity
elem : etree._Element
The xml element
Returns
-------
Entity
The entity
Raises
------
TypeError
If a Child was neither a Property, nor a Parent, nor a Message.
"""
if isinstance(entity, Entity):
entity.role = elem.tag
entity._cuid = elem.get("cuid")
@@ -1270,7 +1369,8 @@ out: List[Entity]
entity.unit = elem.get("unit")
entity.file = elem.get("file")
if hasattr(entity, "affiliation"):
entity_affiliation = getattr(entity, "affiliation", None)
if entity_affiliation:
entity.affiliation = elem.get("affiliation")
vals = list()
@@ -1324,11 +1424,24 @@ out: List[Entity]
def __repr__(self):
return xml2str(self.to_xml())
def retrieve_acl(self):
def retrieve_acl(self) -> None:
self.acl = Entity(name=self.name, id=self.id).retrieve(
flags={"ACL": None}).acl
def update_acl(self):
def update_acl(self) -> Entity:
"""Update acl
Returns
-------
Entity
The updated Entity
Raises
------
TransactionError
If the entity does not exist on the server
TransactionError
If the entity could not be identified by its name
"""
if self.id is None:
c = Container().retrieve(query=self.name, sync=False)
@@ -1353,11 +1466,12 @@ out: List[Entity]
return e
def delete(self, raise_exception_on_error=True):
def delete(self, raise_exception_on_error: bool = True):
return Container().append(self).delete(
raise_exception_on_error=raise_exception_on_error)[0]
def retrieve(self, unique=True, raise_exception_on_error=True, flags=None):
def retrieve(self, unique: bool = True, raise_exception_on_error: bool = True,
flags: Optional[dict] = None) -> Container:
"""Retrieve this entity identified via its id if present and via its
name otherwise. Any locally already existing attributes (name,
description, ...) will be preserved. Any such properties and parents
@@ -1370,14 +1484,26 @@ out: List[Entity]
returns a Container object which carries the returned entities. They are
distinct from this one. This entity will no be changed somehow.
@param unique=True: flag to suppress the ambiguity exception.
Parameters
----------
unique : bool, optional
Flag to suppress the ambiguity exception, by default True
raise_exception_on_error : bool, optional
Flag to raise an exception when an error occurs, by default True
flags : dict, optional
A dictionary of flags to be send with the insertion, by default None
@return
Container with the returned entities or single entity if and only
if unique was True and no exception was raised.
Returns
-------
Container
Container with the returned entities or single entity if and only
if unique was True and no exception was raised
Raises
------
QueryNotUniqueError
Raised if a retrieval is not unique and the unique flag is set to True.
"""
if unique:
c = Container().append(self).retrieve(
unique=unique, raise_exception_on_error=raise_exception_on_error, flags=flags)
@@ -1392,8 +1518,9 @@ out: List[Entity]
return Container().append(self).retrieve(
unique=unique, raise_exception_on_error=raise_exception_on_error, flags=flags)
def insert(self, raise_exception_on_error=True, unique=True,
sync=True, strict=False, flags=None):
def insert(self, raise_exception_on_error: bool = True, unique: bool = True,
sync: bool = True, strict: bool = False,
flags: Optional[dict] = None) -> Container:
"""Insert this entity into a CaosDB server. A successful insertion will
generate a new persistent ID for this entity. This entity can be
identified, retrieved, updated, and deleted via this ID until it has
@@ -1409,6 +1536,7 @@ out: List[Entity]
Parameters
----------
raise_exception_on_error : bool, optional
strict : bool, optional
Flag for strict mode. Default is False.
raise_exception_on_error : bool, optional
@@ -1422,6 +1550,16 @@ out: List[Entity]
"""
Returns
-------
Container
Inserted Entity appended to a container
Raises
------
CaosDBException
Raised if the insertion fails
"""
return Container().append(self).insert(
strict=strict,
raise_exception_on_error=raise_exception_on_error,
@@ -1429,8 +1567,9 @@ out: List[Entity]
sync=sync,
flags=flags)[0]
def update(self, strict=False, raise_exception_on_error=True,
unique=True, flags=None, sync=True):
def update(self, strict: bool = False, raise_exception_on_error: bool = True,
unique: bool = True, flags: Optional[dict] = None,
sync: bool = True) -> Container:
"""Update this entity.
There are two possible work-flows to perform this update:
@@ -1459,9 +1598,29 @@ Second:
anyway. Set flag 'strict' to True in order to force the server to take all warnings as errors.
This prevents the server from updating this entity if any warnings occur.
@param strict=False: Flag for strict mode.
"""
Parameters
----------
strict : bool, optional
Flag for strict mode, by default False
raise_exception_on_error : bool, optional
Flag to raise an exception when an error occurs, by default True
unique : bool, optional
Flag to only allow insertion of elements with unique name, by default True
flags : dict, optional
A dictionary of flags to be send with the insertion, by default None
sync : bool, optional
Flag for sync, by default True
Returns
-------
Container
Updated Entity appended to a container
Raises
------
CaosDBException
Raised if the update fails
"""
return Container().append(self).update(
strict=strict,
sync=sync,
@@ -1590,7 +1749,9 @@ def _log_response(body):
class QueryTemplate():
def __init__(self, id=None, name=None, query=None, description=None): # @ReservedAssignment
def __init__(self, id: Optional[int] = None, name: Optional[str] = None,
query: Optional[str] = None,
description: Optional[str] = None) -> None: # @ReservedAssignment
self.id = (int(id) if id is not None else None)
self.role = "QueryTemplate"
@@ -1616,8 +1777,36 @@ class QueryTemplate():
self.version = None
self.state = None
def retrieve(self, raise_exception_on_error=True, unique=True, sync=True,
flags=None):
def retrieve(self, raise_exception_on_error: bool = True, unique: bool = True,
sync: bool = True, flags: Optional[dict] = None) -> Container:
"""Retrieve this entity. Any locally already existing attributes (name,
description, ...) will be preserved. Any such properties and parents
will be synchronized as well. They will not be overridden. This method
returns a Container containing the this entity.
Note: If only a name is given this could lead to ambiguities. Usually
this would raise a CaosDBException. Set the flag 'unique' to False if
this Exception should be suppressed. If unique is False this method
returns a Container object which carries the returned entities. They are
distinct from this one. This entity will no be changed somehow.
Parameters
----------
raise_exception_on_error : bool, optional
Flag to raise an exception when an error occurs, by default True
unique : bool, optional
Flag to suppress the ambiguity exception, by default True
sync : bool, optional
Flag for sync, by default True
flags : dict, optional
A dictionary of flags to be send with the insertion, by default None
Returns
-------
Container
Container with the returned entities or single entity if and only
if unique was True and no exception was raised
"""
return Container().append(self).retrieve(
raise_exception_on_error=raise_exception_on_error,
@@ -1625,9 +1814,45 @@ class QueryTemplate():
sync=sync,
flags=flags)[0]
def insert(self, strict=True, raise_exception_on_error=True,
unique=True, sync=True, flags=None):
def insert(self, strict: bool = True, raise_exception_on_error: bool = True,
unique: bool = True, sync: bool = True,
flags: Optional[dict] = None) -> Container:
"""Insert this entity into a CaosDB server. A successful insertion will
generate a new persistent ID for this entity. This entity can be
identified, retrieved, updated, and deleted via this ID until it has
been deleted.
If the insertion fails, a CaosDBException will be raised. The server will have returned at
least one error-message describing the reason why it failed in that case (call
<this_entity>.get_all_messages() in order to get these error-messages).
Some insertions might cause warning-messages on the server-side, but the entities are inserted
anyway. Set the flag 'strict' to True in order to force the server to take all warnings as errors.
This prevents the server from inserting this entity if any warning occurs.
Parameters
----------
strict : bool, optional
Flag for strict mode, by default False
raise_exception_on_error : bool, optional
Flag to raise an exception when an error occurs, by default True
unique : bool, optional
Flag to only allow insertion of elements with unique name, by default True
sync : bool, optional
Flag for sync, by default True
flags : dict, optional
A dictionary of flags to be send with the insertion, by default None
Returns
-------
Container
Inserted Entity appended to a container
Raises
------
CaosDBException
Raised if the insertion fails
"""
return Container().append(self).insert(
strict=strict,
raise_exception_on_error=raise_exception_on_error,
@@ -1635,9 +1860,60 @@ class QueryTemplate():
sync=sync,
flags=flags)[0]
def update(self, strict=True, raise_exception_on_error=True,
unique=True, sync=True, flags=None):
def update(self, strict: bool = True, raise_exception_on_error: bool = True,
unique: bool = True, sync: bool = True,
flags: Optional[dict] = None) -> Container:
"""Update this entity.
There are two possible work-flows to perform this update:
First:
1) retrieve an entity
2) do changes
3) call update method
Second:
1) construct entity with id
2) call update method.
For slight changes the second one it is more comfortable. Furthermore, it is possible to stay
off-line until calling the update method. The name, description, unit, datatype, path,
and value of an entity may be changed. Additionally, properties, parents and messages may be added.
However, the first one is more powerful: It is possible to delete and change properties, parents
and attributes, which is not possible via the second one for internal reasons (which are reasons
of definiteness).
If the update fails, a CaosDBException will be raised. The server will have returned at
least one error message describing the reason why it failed in that case (call
<this_entity>.get_all_messages() in order to get these error-messages).
Some updates might cause warning messages on the server-side, but the updates are performed
anyway. Set flag 'strict' to True in order to force the server to take all warnings as errors.
This prevents the server from updating this entity if any warnings occur.
Parameters
----------
strict : bool, optional
Flag for strict mode, by default False
raise_exception_on_error : bool, optional
Flag to raise an exception when an error occurs, by default True
unique : bool, optional
Flag to only allow insertion of elements with unique name, by default True
sync : bool, optional
Flag for sync, by default True
flags : dict, optional
A dictionary of flags to be send with the insertion, by default None
Returns
-------
Container
Updated Entity appended to a container
Raises
------
CaosDBException
Raised if the update fails
"""
return Container().append(self).update(
strict=strict,
raise_exception_on_error=raise_exception_on_error,
@@ -1645,14 +1921,26 @@ class QueryTemplate():
sync=sync,
flags=flags)[0]
def delete(self, raise_exception_on_error=True):
def delete(self, raise_exception_on_error: bool = True) -> Container:
"""Delete this entity
Parameters
----------
raise_exception_on_error : bool, optional
Flag to raise an exception when an error occurs, by default True
Returns
-------
Container
"""
return Container().append(self).delete(
raise_exception_on_error=raise_exception_on_error)[0]
def __repr__(self):
return xml2str(self.to_xml())
def to_xml(self, xml=None):
def to_xml(self, xml: Optional[etree._Element] = None) -> etree._Element:
if xml is None:
xml = etree.Element("QueryTemplate")
@@ -1682,10 +1970,11 @@ class QueryTemplate():
return xml
@staticmethod
def _from_xml(xml):
def _from_xml(xml: etree._Element):
if xml.tag.lower() == "querytemplate":
q = QueryTemplate(name=xml.get("name"),
q: Any = QueryTemplate(name=xml.get("name"),
description=xml.get("description"), query=None)
# TODO: Add correct type hint. Any is just a fix. 17.08.2021, AK
for e in xml:
if e.tag.lower() == "query":
@@ -1707,16 +1996,16 @@ class QueryTemplate():
else:
return None
def clear_server_messages(self):
def clear_server_messages(self) -> None:
self.messages.clear_server_messages()
def get_parents(self):
def get_parents(self) -> list:
return []
def get_properties(self):
def get_properties(self) -> list:
return []
def has_id(self):
def has_id(self) -> bool:
return self.id is not None
def get_errors(self):
@@ -1728,10 +2017,10 @@ class QueryTemplate():
return ret
def get_messages(self):
def get_messages(self) -> _Messages:
return self.messages
def has_errors(self):
def has_errors(self) -> bool:
return len(self.get_errors()) > 0
@@ -1751,14 +2040,17 @@ class Parent(Entity):
def affiliation(self, affiliation):
self.__affiliation = affiliation
def __init__(self, id=None, name=None, description=None, inheritance=None): # @ReservedAssignment
def __init__(self, id: Optional[int] = None, name: Optional[str] = None,
description: Optional[str] = None,
inheritance: Optional[str] = None) -> None: # @ReservedAssignment
Entity.__init__(self, id=id, name=name, description=description)
if inheritance is not None:
self.set_flag("inheritance", inheritance)
self.__affiliation = None
def to_xml(self, xml=None, add_properties=None):
def to_xml(self, xml: Optional[etree._Element] = None,
add_properties: TA_add_properties = None, *args, **kwargs) -> etree._Element:
if xml is None:
xml = etree.Element("Parent")
@@ -1777,7 +2069,7 @@ class Property(Entity):
"""CaosDB's Property object."""
def add_property(self, property=None, value=None, id=None, name=None, description=None, datatype=None,
def add_property(self, property: Union[int, str, Entity, None]=None, value: Union[str, int, Entity, None]=None, id=None, name=None, description=None, datatype=None,
unit=None, importance=FIX, inheritance=FIX): # @ReservedAssignment
"""See ``Entity.add_property``."""
@@ -1785,13 +2077,12 @@ class Property(Entity):
property=property, id=id, name=name, description=description, datatype=datatype,
value=value, unit=unit, importance=importance, inheritance=inheritance)
def add_parent(self, parent=None, id=None, name=None, inheritance=FIX):
def add_parent(self, parent: Union[Entity, str, int, None] = None, id=None, name=None, inheritance=FIX) -> Entity:
"""Add a parent Entity to this Property.
Parameters
----------
Parameters
----------
parent : Entity or int or str or None
The parent entity, either specified by the Entity object
itself, or its id or its name. Default is None.
@@ -1807,10 +2098,13 @@ class Property(Entity):
entity. If no `inheritance` is given, no properties will be inherited by the child.
This parameter is case-insensitive.
Returns
-------
Entity
See Also
--------
Entity.add_parent
"""
return super(Property, self).add_parent(parent=parent, id=id, name=name, inheritance=inheritance)
Loading