Skip to content
Snippets Groups Projects

Add and fix more type hints

1 file
+ 12
2
Compare changes
  • Side-by-side
  • Inline
+ 222
135
@@ -34,8 +34,6 @@ transactions.
"""
from __future__ import annotations # Can be removed with 3.10.
from __future__ import print_function, unicode_literals
from enum import Enum
import re
import sys
@@ -49,13 +47,15 @@ from random import randint
from tempfile import NamedTemporaryFile
from typing import TYPE_CHECKING
from typing import Any, Final, Literal, Optional, TextIO, Union
if TYPE_CHECKING and sys.version_info > (3, 7):
if TYPE_CHECKING:
from datetime import datetime
from typing import Any, Dict, Optional, Type, Union, List, TextIO, Tuple, Literal
from .datatype import DATATYPE
from tempfile import _TemporaryFileWrapper
from io import BufferedWriter
from os import PathLike
QueryDict = dict[str, Optional[str]]
from warnings import warn
@@ -102,15 +102,17 @@ from .versioning import Version
_ENTITY_URI_SEGMENT = "Entity"
OBLIGATORY = "OBLIGATORY"
SUGGESTED = "SUGGESTED"
RECOMMENDED = "RECOMMENDED"
FIX = "FIX"
ALL = "ALL"
NONE = "NONE"
OBLIGATORY: Final = "OBLIGATORY"
SUGGESTED: Final = "SUGGESTED"
RECOMMENDED: Final = "RECOMMENDED"
FIX: Final = "FIX"
ALL: Final = "ALL"
NONE: Final = "NONE"
if TYPE_CHECKING:
INHERITANCE = Literal["OBLIGATORY", "SUGGESTED", "RECOMMENDED", "ALL", "NONE", "FIX"]
IMPORTANCE = Literal["OBLIGATORY", "RECOMMENDED", "SUGGESTED", "FIX", "NONE"]
ROLE = Literal["Entity", "Record", "RecordType", "Property", "File"]
SPECIAL_ATTRIBUTES = ["name", "role", "datatype", "description",
"id", "path", "checksum", "size", "value"]
@@ -139,7 +141,7 @@ class Entity:
**kwargs,
):
self.__role = kwargs["role"] if "role" in kwargs else None
self.__role: Optional[ROLE] = kwargs["role"] if "role" in kwargs else None
self._checksum: Optional[str] = None
self._size = None
self._upload = None
@@ -148,7 +150,7 @@ class Entity:
self._wrapped_entity: Optional[Entity] = None
self._version: Optional[Version] = None
self._cuid: Optional[str] = None
self._flags: Dict[str, str] = dict()
self._flags: dict[str, str] = dict()
self.__value = None
self.__datatype: Optional[DATATYPE] = None
self.datatype: Optional[DATATYPE] = datatype
@@ -168,7 +170,7 @@ class Entity:
self.id: Optional[int] = id
self.state: Optional[State] = None
def copy(self):
def copy(self) -> Entity:
"""
Return a copy of entity.
@@ -179,6 +181,7 @@ class Entity:
Special attributes, as defined by the global variable SPECIAL_ATTRIBUTES and additionaly
the "value" are copied using setattr.
"""
new: Union[File, Property, RecordType, Record, Entity]
if self.role == "File":
new = File()
elif self.role == "Property":
@@ -243,7 +246,7 @@ class Entity:
return self._wrapped_entity.size
@property
def id(self):
def id(self) -> Any:
if self.__id is not None:
return self.__id
@@ -253,9 +256,9 @@ class Entity:
return self._wrapped_entity.id
@id.setter
def id(self, new_id):
def id(self, new_id) -> None:
if new_id is not None:
self.__id = int(new_id)
self.__id: Optional[int] = int(new_id)
else:
self.__id = None
@@ -449,7 +452,8 @@ class Entity:
"""
# @review Florian Spreckelsen 2022-03-17
if self.acl is None:
raise EntityHasNoAclError("This entity does not have an ACL (yet).")
raise EntityHasNoAclError(
"This entity does not have an ACL (yet).")
self.acl.deny(realm=realm, username=username, role=role,
permission=permission, priority=priority,
@@ -457,6 +461,8 @@ class Entity:
def revoke_denial(self, realm=None, username=None,
role=None, permission=None, priority=False):
if self.acl is None:
raise EntityHasNoAclError("This entity does not have an ACL (yet).")
self.acl.revoke_denial(
realm=realm,
username=username,
@@ -466,6 +472,8 @@ class Entity:
def revoke_grant(self, realm=None, username=None,
role=None, permission=None, priority=False):
if self.acl is None:
raise EntityHasNoAclError("This entity does not have an ACL (yet).")
self.acl.revoke_grant(
realm=realm,
username=username,
@@ -479,7 +487,8 @@ class Entity:
return permission in self.permissions
if self.acl is None:
raise EntityHasNoAclError("This entity does not have an ACL (yet).")
raise EntityHasNoAclError(
"This entity does not have an ACL (yet).")
return self.acl.is_permitted(role=role, permission=permission)
def get_all_messages(self) -> Messages:
@@ -594,10 +603,10 @@ class Entity:
bool,
datetime,
Entity,
List[int],
List[str],
List[bool],
List[Entity],
list[int],
list[str],
list[bool],
list[Entity],
None,
] = None,
id: Optional[int] = None,
@@ -662,8 +671,8 @@ class Entity:
If the first parameter is an integer then it is interpreted as the id and id must be
undefined or None.
UserWarning
If the first parameter is not None and neither an instance of Entity nor an integer it is
interpreted as the name and name must be undefined or None.
If the first parameter is not None and neither an instance of Entity nor an integer it
is interpreted as the name and name must be undefined or None.
Raises
------
@@ -679,7 +688,8 @@ class Entity:
>>> import linkahead as db
>>> rec = db.Record(name="TestRec").add_parent(name="TestType")
>>> rec.add_property("TestProp", value=27) # specified by name, you could equally use the property's id if it is known
>>> rec.add_property("TestProp", value=27) # specified by name, you could equally use the
>>> # property's id if it is known
You can also use the Python object:
@@ -702,10 +712,12 @@ class Entity:
Note that since `TestProp` is a scalar integer Property, the datatype
`LIST<INTEGER>` has to be specified explicitly.
Finally, we can also add reference properties, specified by the RecordType of the referenced entity.
Finally, we can also add reference properties, specified by the RecordType of the referenced
entity.
>>> ref_rec = db.Record(name="ReferencedRecord").add_parent(name="OtherRT")
>>> rec.add_property(name="OtherRT", value=ref_rec) # or value=ref_rec.id if ref_rec has one set by the server
>>> rec.add_property(name="OtherRT", value=ref_rec) # or value=ref_rec.id if ref_rec has
>>> # one set by the server
See more on adding properties and inserting data in
https://docs.indiscale.com/caosdb-pylib/tutorials/Data-Insertion.html.
@@ -725,12 +737,21 @@ class Entity:
abstract_property = property
elif isinstance(property, int):
if pid is not None:
raise UserWarning("The first parameter was an integer which would normally be interpreted as the id of the property which is to be added. But you have also specified a parameter 'id' in the method call. This is ambiguous and cannot be processed.")
raise UserWarning(
"The first parameter was an integer which would normally be interpreted as the"
" id of the property which is to be added. But you have also specified a"
" parameter 'id' in the method call. This is ambiguous and cannot be processed."
)
pid = property
id = pid
elif property is not None:
if name is not None:
raise UserWarning("The first parameter was neither an instance of Entity nor an integer. Therefore the string representation of your first parameter would normally be interpreted name of the property which is to be added. But you have also specified a parameter 'name' in the method call. This is ambiguous and cannot be processed.")
raise UserWarning(
"The first parameter was neither an instance of Entity nor an integer."
" Therefore the string representation of your first parameter would normally be"
" interpreted name of the property which is to be added. But you have also"
" specified a parameter 'name' in the method call. This is ambiguous and cannot"
" be processed.")
name = str(property)
if property is None and name is None and pid is None:
@@ -904,7 +925,7 @@ out: bool
return self.parents
def get_parents_recursively(self, retrieve: bool = True) -> List[Entity]:
def get_parents_recursively(self, retrieve: bool = True) -> list[Entity]:
"""Get all ancestors of this entity.
Parameters
@@ -915,16 +936,16 @@ retrieve: bool, optional
Returns
-------
out: List[Entity]
out: list[Entity]
The parents of this Entity
"""
all_parents: List[Entity] = []
all_parents: list[Entity] = []
self._get_parent_recursively(all_parents, retrieve=retrieve)
return all_parents
def _get_parent_recursively(self, all_parents: List[Entity], retrieve: bool = True):
def _get_parent_recursively(self, all_parents: list[Entity], retrieve: bool = True):
"""Get all ancestors with a little helper.
As a side effect of this method, the ancestors are added to
@@ -1045,12 +1066,13 @@ out: List[Entity]
return p
else:
raise ValueError("`pattern` argument should be an Entity, int or str.")
raise ValueError(
"`pattern` argument should be an Entity, int or str.")
return None
def _get_value_for_selector(
self, selector: Union[str, List[str], Tuple[str]]
self, selector: Union[str, list[str], tuple[str]]
) -> Any:
"""return the value described by the selector
@@ -1144,7 +1166,7 @@ out: List[Entity]
row : 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)
@@ -1192,7 +1214,7 @@ out: List[Entity]
return ret
def get_errors_deep(self, roots=None) -> List[Tuple[str, List[Entity]]]:
def get_errors_deep(self, roots=None) -> list[tuple[str, list[Entity]]]:
"""Get all error messages of this entity and all sub-entities /
parents / properties.
@@ -1437,7 +1459,30 @@ out: List[Entity]
self.acl = Entity(name=self.name, id=self.id).retrieve(
flags={"ACL": None}).acl
def update_acl(self):
def update_acl(self, **kwargs):
"""Update this entity's ACL on the server.
A typical workflow is to first edit ``self.acl`` and then call this
method.
Note
----
This overwrites any existing ACL, so you may want to run
``retrieve_acl`` before updating the ACL in this entity.
Parameters
----------
**kwargs : dict
Keyword arguments that are passed through to the
``Entity.update`` method. Useful for e.g. ``unique=False`` in the
case of naming collisions.
Returns
-------
e : Entity
This entity after the update of the ACL.
"""
if self.id is None:
c = Container().retrieve(query=self.name, sync=False)
@@ -1457,8 +1502,10 @@ out: List[Entity]
raise TransactionError(ae)
else:
e = Container().retrieve(query=self.id, sync=False)[0]
if self.acl is None:
raise EntityHasNoAclError("This entity does not have an ACL yet. Please set one first.")
e.acl = ACL(self.acl.to_xml())
e.update()
e.update(**kwargs)
return e
@@ -1514,13 +1561,14 @@ out: List[Entity]
identified, retrieved, updated, and deleted via this ID until it has
been deleted.
If the insertion fails, a LinkAheadException will be raised. The server will have returned at
least one error-message describing the reason why it failed in that case (call
If the insertion fails, a LinkAheadException 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.
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
----------
@@ -1558,21 +1606,22 @@ 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.
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).
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 LinkAheadException 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.
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.
"""
@@ -1747,8 +1796,8 @@ class QueryTemplate():
raise_exception_on_error: bool = True,
unique: bool = True,
sync: bool = True,
flags: Optional[Dict[str, Optional[str]]] = None,
):
flags: Optional[QueryDict] = None,
) -> Container:
return Container().append(self).retrieve(
raise_exception_on_error=raise_exception_on_error,
@@ -1762,8 +1811,8 @@ class QueryTemplate():
raise_exception_on_error: bool = True,
unique: bool = True,
sync: bool = True,
flags: Optional[Dict[str, Optional[str]]] = None,
):
flags: Optional[QueryDict] = None,
) -> Container:
return Container().append(self).insert(
strict=strict,
@@ -1778,8 +1827,8 @@ class QueryTemplate():
raise_exception_on_error: bool = True,
unique: bool = True,
sync: bool = True,
flags: Optional[Dict[str, Optional[str]]] = None,
):
flags: Optional[QueryDict] = None,
) -> Container:
return Container().append(self).update(
strict=strict,
@@ -1795,7 +1844,7 @@ class QueryTemplate():
def __repr__(self):
return xml2str(self.to_xml())
def to_xml(self, xml: Optional[etree._Element] = None):
def to_xml(self, xml: Optional[etree._Element] = None) -> etree._Element:
if xml is None:
xml = etree.Element("QueryTemplate")
@@ -1933,7 +1982,8 @@ class Property(Entity):
"""LinkAhead's Property object."""
def add_property(self, property=None, value=None, id=None, name=None, description=None, datatype=None,
def add_property(self, property=None, value=None, id=None, name=None, description=None,
datatype=None,
unit=None, importance=FIX, inheritance=FIX): # @ReservedAssignment
"""See ``Entity.add_property``."""
@@ -1969,7 +2019,8 @@ class Property(Entity):
"""
return super(Property, self).add_parent(parent=parent, id=id, name=name, inheritance=inheritance)
return super(Property, self).add_parent(parent=parent, id=id, name=name,
inheritance=inheritance)
def __init__(
self,
@@ -1999,13 +2050,14 @@ class Property(Entity):
local_serialization=local_serialization,
)
def is_reference(self, server_retrieval=False):
def is_reference(self, server_retrieval: bool = False) -> Optional[bool]:
"""Returns whether this Property is a reference
Parameters
----------
server_retrieval : bool, optional
If True and the datatype is not set, the Property is retrieved from the server, by default False
If True and the datatype is not set, the Property is retrieved from the server, by
default False
Returns
-------
@@ -2055,7 +2107,7 @@ class Message(object):
self.code = int(code) if code is not None else None
self.body = body
def to_xml(self, xml: Optional[etree._Element] = None):
def to_xml(self, xml: Optional[etree._Element] = None) -> etree._Element:
if xml is None:
xml = etree.Element(str(self.type))
@@ -2075,21 +2127,23 @@ class Message(object):
def __eq__(self, obj):
if isinstance(obj, Message):
return self.type == obj.type and self.code == obj.code and self.description == obj.description
return (self.type == obj.type and self.code == obj.code
and self.description == obj.description)
return False
def get_code(self):
def get_code(self) -> Optional[int]:
warn(("get_code is deprecated and will be removed in future. "
"Use self.code instead."), DeprecationWarning)
return int(self.code)
return int(self.code) if self.code is not None else None
class RecordType(Entity):
"""This class represents LinkAhead's RecordType entities."""
def add_property(self, property=None, value=None, id=None, name=None, description=None, datatype=None,
def add_property(self, property=None, value=None, id=None, name=None, description=None,
datatype=None,
unit=None, importance=RECOMMENDED, inheritance=FIX): # @ReservedAssignment
"""See ``Entity.add_property``."""
@@ -2167,7 +2221,8 @@ class Record(Entity):
"""This class represents LinkAhead's Record entities."""
def add_property(self, property=None, value=None, id=None, name=None, description=None, datatype=None,
def add_property(self, property=None, value=None, id=None, name=None, description=None,
datatype=None,
unit=None, importance=FIX, inheritance=FIX): # @ReservedAssignment
"""See ``Entity.add_property``."""
@@ -2266,7 +2321,7 @@ class File(Record):
xml: Optional[etree._Element] = None,
add_properties: INHERITANCE = "ALL",
local_serialization: bool = False,
):
) -> etree._Element:
"""Convert this file to an xml element.
@return: xml element
@@ -2278,7 +2333,7 @@ class File(Record):
return Entity.to_xml(self, xml=xml, add_properties=add_properties,
local_serialization=local_serialization)
def download(self, target: Optional[str] = None):
def download(self, target: Optional[str] = None) -> str:
"""Download this file-entity's actual file from the file server. It
will be stored to the target or will be hold as a temporary file.
@@ -2288,7 +2343,8 @@ class File(Record):
self.clear_server_messages()
if target:
file_: Union[BufferedWriter, _TemporaryFileWrapper] = open(target, "wb")
file_: Union[BufferedWriter,
_TemporaryFileWrapper] = open(target, "wb")
else:
file_ = NamedTemporaryFile(mode='wb', delete=False)
checksum = File.download_from_path(file_, self.path)
@@ -2342,7 +2398,7 @@ class File(Record):
return File._get_checksum_single_file(files)
@staticmethod
def _get_checksum_single_file(single_file):
def _get_checksum_single_file(single_file: Union[str, bytes, PathLike[str], PathLike[bytes]]):
_file = open(single_file, 'rb')
data = _file.read(1000)
checksum = sha512()
@@ -2368,10 +2424,10 @@ class _Properties(list):
def __init__(self):
list.__init__(self)
self._importance: Dict[Entity, IMPORTANCE] = dict()
self._inheritance: Dict[Entity, INHERITANCE] = dict()
self._element_by_name: Dict[str, Entity] = dict()
self._element_by_id: Dict[str, Entity] = dict()
self._importance: dict[Entity, IMPORTANCE] = dict()
self._inheritance: dict[Entity, INHERITANCE] = dict()
self._element_by_name: dict[str, Entity] = dict()
self._element_by_id: dict[str, Entity] = dict()
def get_importance(
self, property: Union[Property, Entity, str, None]
@@ -2404,7 +2460,7 @@ class _Properties(list):
def append(
self,
property: Union[List[Entity], Entity, Property],
property: Union[list[Entity], Entity, Property],
importance: Optional[IMPORTANCE] = None,
inheritance: Optional[INHERITANCE] = None,
): # @ReservedAssignment
@@ -2435,6 +2491,7 @@ class _Properties(list):
return self
def to_xml(self, add_to_element: etree._Element, add_properties: INHERITANCE):
p: Property
for p in self:
importance = self._importance.get(p)
@@ -2562,7 +2619,7 @@ class _ParentList(list):
return self
def to_xml(self, add_to_element):
def to_xml(self, add_to_element: etree._Element):
for p in self:
pelem = etree.Element("Parent")
@@ -2657,7 +2714,8 @@ class Messages(list):
<<< msgs = Messages()
<<< # create Message
<<< msg = Message(type="HelloWorld", code=1, description="Greeting the world", body="Hello, world!")
<<< msg = Message(type="HelloWorld", code=1, description="Greeting the world",
... body="Hello, world!")
<<< # append it to the Messages
<<< msgs.append(msg)
@@ -2731,11 +2789,12 @@ class Messages(list):
if isinstance(value, Message):
body = value.body
description = value.description
m = Message
m = Message()
else:
body = value
description = None
m = Message(type=type, code=code, description=description, body=body)
m = Message(type=type, code=code,
description=description, body=body)
if isinstance(key, int):
super().__setitem__(key, m)
else:
@@ -2743,7 +2802,8 @@ class Messages(list):
def __getitem__(self, key):
if not isinstance(key, int):
warn("__getitem__ only supports integer keys in future.", DeprecationWarning)
warn("__getitem__ only supports integer keys in future.",
DeprecationWarning)
if isinstance(key, tuple):
if len(key) == 2:
type = key[0] # @ReservedAssignment
@@ -2769,7 +2829,8 @@ class Messages(list):
def __delitem__(self, key):
if isinstance(key, tuple):
warn("__delitem__ only supports integer keys in future.", DeprecationWarning)
warn("__delitem__ only supports integer keys in future.",
DeprecationWarning)
if self.get(key[0], key[1]) is not None:
self.remove(self.get(key[0], key[1]))
else:
@@ -2822,7 +2883,7 @@ class Messages(list):
return default
def to_xml(self, add_to_element):
def to_xml(self, add_to_element: etree._Element):
for m in self:
melem = m.to_xml()
add_to_element.append(melem)
@@ -2974,7 +3035,7 @@ class Container(list):
def __hash__(self):
return object.__hash__(self)
def remove(self, entity):
def remove(self, entity: Entity):
"""Remove the first entity from this container which is equal to the
given entity. Raise a ValueError if there is no such entity.
@@ -3011,7 +3072,8 @@ class Container(list):
return e
raise KeyError("No entity with such cuid (" + str(cuid) + ")!")
def get_entity_by_id(self, id): # @ReservedAssignment
# @ReservedAssignment
def get_entity_by_id(self, id: Union[int, str]) -> Entity:
"""Get the first entity which has the given id. Note: If several
entities are in this list which have the same id, this method will only
return the first and ignore the others.
@@ -3044,7 +3106,7 @@ class Container(list):
return error_list
def get_entity_by_name(self, name: str, case_sensitive: bool = True):
def get_entity_by_name(self, name: str, case_sensitive: bool = True) -> Entity:
"""Get the first entity which has the given name. Note: If several
entities are in this list which have the same name, this method will
only return the first and ignore the others.
@@ -3123,11 +3185,13 @@ class Container(list):
return self
def to_xml(self, add_to_element=None, local_serialization=False):
def to_xml(self, add_to_element: Optional[etree._Element] = None,
local_serialization: bool = False) -> etree._Element:
"""Get an xml tree representing this Container or append all entities
to the given xml element.
@param add_to_element=None: optional element to which all entities of this container is to be appended.
@param add_to_element=None: optional element to which all entities of this container is to
be appended.
@return xml element
"""
tmpid = 0
@@ -3277,7 +3341,7 @@ class Container(list):
if isinstance(e, Message):
c.messages.append(e)
elif isinstance(e, Query):
c.query = e
c.query = e # type: ignore
if e.messages is not None:
c.messages.extend(e.messages)
@@ -3300,7 +3364,8 @@ class Container(list):
return c
else:
raise LinkAheadException(
"The server's response didn't contain the expected elements. The configuration of this client might be invalid (especially the url).")
"The server's response didn't contain the expected elements. The configuration of"
" this client might be invalid (especially the url).")
def _sync(
self,
@@ -3366,7 +3431,8 @@ class Container(list):
# which is to be synced with which:
# sync_dict[local_entity]=sync_remote_enities
sync_dict: Dict[Union[Container, Entity], Optional[List[Entity]]] = dict()
sync_dict: dict[Union[Container, Entity],
Optional[list[Entity]]] = dict()
# list of remote entities which already have a local equivalent
used_remote_entities = []
@@ -3395,7 +3461,8 @@ class Container(list):
msg = "Request was not unique. CUID " + \
str(local_entity._cuid) + " was found " + \
str(len(sync_remote_entities)) + " times."
local_entity.add_message(Message(description=msg, type="Error"))
local_entity.add_message(
Message(description=msg, type="Error"))
if raise_exception_on_error:
raise MismatchingEntitiesError(msg)
@@ -3420,7 +3487,8 @@ class Container(list):
msg = "Request was not unique. ID " + \
str(local_entity.id) + " was found " + \
str(len(sync_remote_entities)) + " times."
local_entity.add_message(Message(description=msg, type="Error"))
local_entity.add_message(
Message(description=msg, type="Error"))
if raise_exception_on_error:
raise MismatchingEntitiesError(msg)
@@ -3450,7 +3518,8 @@ class Container(list):
msg = "Request was not unique. Path " + \
str(local_entity.path) + " was found " + \
str(len(sync_remote_entities)) + " times."
local_entity.add_message(Message(description=msg, type="Error"))
local_entity.add_message(
Message(description=msg, type="Error"))
if raise_exception_on_error:
raise MismatchingEntitiesError(msg)
@@ -3480,7 +3549,8 @@ class Container(list):
msg = "Request was not unique. Name " + \
str(local_entity.name) + " was found " + \
str(len(sync_remote_entities)) + " times."
local_entity.add_message(Message(description=msg, type="Error"))
local_entity.add_message(
Message(description=msg, type="Error"))
if raise_exception_on_error:
raise MismatchingEntitiesError(msg)
@@ -3493,13 +3563,15 @@ class Container(list):
sync_remote_entities.append(remote_entity)
if len(sync_remote_entities) > 0:
sync_dict[self] = sync_remote_entities # FIXME: How is this supposed to work?
# FIXME: How is this supposed to work?
sync_dict[self] = sync_remote_entities
if unique and len(sync_remote_entities) != 0:
msg = "Request was not unique. There are " + \
str(len(sync_remote_entities)) + \
" entities which could not be matched to one of the requested ones."
remote_container.add_message(Message(description=msg, type="Error"))
remote_container.add_message(
Message(description=msg, type="Error"))
if raise_exception_on_error:
raise MismatchingEntitiesError(msg)
@@ -3529,14 +3601,16 @@ class Container(list):
dependent_references = set()
dependencies = set()
container_item: Entity
for container_item in container:
item_id.add(container_item.id)
for parents in container_item.get_parents():
is_parent.add(parents.id)
prop: Property
for prop in container_item.get_properties():
prop_dt = prop.datatype
prop_dt: Union[DATATYPE, str, None] = prop.datatype
if prop_dt is not None and is_reference(prop_dt):
# add only if it is a reference, not a simple property
# Step 1: look for prop.value
@@ -3559,7 +3633,8 @@ class Container(list):
if is_list_datatype(prop_dt):
ref_name = get_list_datatype(prop_dt)
try:
is_being_referenced.add(container.get_entity_by_name(ref_name).id)
is_being_referenced.add(
container.get_entity_by_name(ref_name).id) # type: ignore
except KeyError:
pass
elif isinstance(prop_dt, str):
@@ -3592,7 +3667,8 @@ class Container(list):
return dependencies
def delete(self, raise_exception_on_error=True, flags=None, chunk_size=100):
def delete(self, raise_exception_on_error: bool = True,
flags: Optional[QueryDict] = None, chunk_size: int = 100):
"""Delete all entities in this container.
Entities are identified via their id if present and via their
@@ -3604,7 +3680,8 @@ class Container(list):
"""
item_count = len(self)
# Split Container in 'chunk_size'-sized containers (if necessary) to avoid error 414 Request-URI Too Long
# Split Container in 'chunk_size'-sized containers (if necessary) to avoid error 414
# Request-URI Too Long
if item_count > chunk_size:
dependencies = Container._find_dependencies_in_container(self)
@@ -3615,7 +3692,8 @@ class Container(list):
if len(dependencies) == item_count:
if raise_exception_on_error:
te = TransactionError(
msg="The container is too large and with too many dependencies within to be deleted.",
msg=("The container is too large and with too many dependencies within to"
" be deleted."),
container=self)
raise te
@@ -3706,7 +3784,7 @@ class Container(list):
unique: bool = True,
raise_exception_on_error: bool = True,
sync: bool = True,
flags=None,
flags: Optional[QueryDict] = None,
):
"""Retrieve all entities in this container identified via their id if
present and via their name otherwise. Any locally already existing
@@ -3716,9 +3794,9 @@ class Container(list):
If any entity has no id and no name a LinkAheadException will be raised.
Note: If only a name is given this could lead to ambiguities. All entities with the name in question
will be returned. Therefore, the container could contain more elements after the retrieval than
before.
Note: If only a name is given this could lead to ambiguities. All entities with the name in
question will be returned. Therefore, the container could contain more elements after the
retrieval than before.
"""
if isinstance(query, list):
@@ -3782,7 +3860,7 @@ class Container(list):
return (entities[0:hl], entities[hl:len(entities)])
def _retrieve(self, entities, flags: Dict[str, Optional[str]]):
def _retrieve(self, entities, flags: Optional[QueryDict]):
c = get_connection()
try:
_log_request("GET: " + _ENTITY_URI_SEGMENT + str(entities) +
@@ -3815,7 +3893,8 @@ class Container(list):
return self
@staticmethod
def _dir_to_http_parts(root, d, upload): # @ReservedAssignment
# @ReservedAssignment
def _dir_to_http_parts(root: str, d: Optional[str], upload: str):
ret = []
x = (root + '/' + d if d is not None else root)
@@ -3842,7 +3921,7 @@ class Container(list):
raise_exception_on_error: bool = True,
unique: bool = True,
sync: bool = True,
flags: Optional[Dict[str, Any]] = None,
flags: Optional[dict[str, Any]] = None,
):
"""Update these entites."""
@@ -3854,7 +3933,7 @@ class Container(list):
self.clear_server_messages()
insert_xml = etree.Element("Update")
http_parts: List[MultipartParam] = []
http_parts: list[MultipartParam] = []
if flags is None:
flags = {}
@@ -3925,7 +4004,7 @@ class Container(list):
@staticmethod
def _process_file_if_present_and_add_to_http_parts(
http_parts: List[MultipartParam], entity: Union[File, Entity]
http_parts: list[MultipartParam], entity: Union[File, Entity]
):
if isinstance(entity, File) and hasattr(
entity, 'file') and entity.file is not None:
@@ -3975,7 +4054,7 @@ class Container(list):
raise_exception_on_error: bool = True,
unique: bool = True,
sync: bool = True,
flags: Optional[Dict[str, Optional[str]]] = None,
flags: Optional[QueryDict] = None,
):
"""Insert this file entity into LinkAhead. A successful insertion will
generate a new persistent ID for this entity. This entity can be
@@ -4003,7 +4082,7 @@ class Container(list):
self.clear_server_messages()
insert_xml = etree.Element("Insert")
http_parts: List[MultipartParam] = []
http_parts: list[MultipartParam] = []
if flags is None:
flags = {}
@@ -4057,7 +4136,8 @@ class Container(list):
if len(self) > 0 and len(insert_xml) < 1:
te = TransactionError(
msg="There are no entities to be inserted. This container contains existent entities only.",
msg=("There are no entities to be inserted. This container contains existent"
" entities only."),
container=self)
raise te
_log_request("POST: " + _ENTITY_URI_SEGMENT +
@@ -4164,8 +4244,8 @@ class Container(list):
return self
def get_property_values(
self, *selectors: Union[str, Tuple[str]]
) -> List[Tuple[str]]:
self, *selectors: Union[str, tuple[str]]
) -> list[tuple[str]]:
""" Return a list of tuples with values of the given selectors.
I.e. a tabular representation of the container's content.
@@ -4221,7 +4301,8 @@ def sync_global_acl():
ACL.global_acl = ACL(xml=pelem)
else:
raise LinkAheadException(
"The server's response didn't contain the expected elements. The configuration of this client might be invalid (especially the url).")
"The server's response didn't contain the expected elements. The configuration of this"
" client might be invalid (especially the url).")
def get_known_permissions():
@@ -4257,11 +4338,13 @@ class ACI():
return hash(self.__repr__())
def __eq__(self, other):
return isinstance(other, ACI) and (self.role is None and self.username == other.username and self.realm ==
other.realm) or self.role == other.role and self.permission == other.permission
return (isinstance(other, ACI) and
(self.role is None and self.username == other.username
and self.realm == other.realm)
or self.role == other.role and self.permission == other.permission)
def __repr__(self):
return str(self.realm) + ":" + str(self.username) + ":" + str(self.role) + ":" + str(self.permission)
return ":".join([str(self.realm), str(self.username), str(self.role), str(self.permission)])
def add_to_element(self, e: etree._Element):
if self.role is not None:
@@ -4340,7 +4423,7 @@ class ACL():
permission=permission, priority=priority,
revoke_grant=False)
def combine(self, other: ACL):
def combine(self, other: ACL) -> ACL:
""" Combine and return new instance."""
result = ACL()
result._grants.update(other._grants)
@@ -4355,7 +4438,11 @@ class ACL():
return result
def __eq__(self, other):
return isinstance(other, ACL) and other._grants == self._grants and self._denials == other._denials and self._priority_grants == other._priority_grants and self._priority_denials == other._priority_denials
return (isinstance(other, ACL)
and other._grants == self._grants
and self._denials == other._denials
and self._priority_grants == other._priority_grants
and self._priority_denials == other._priority_denials)
def is_empty(self):
return len(self._grants) + len(self._priority_grants) + \
@@ -4682,7 +4769,7 @@ class Query():
return self.flags.get(key)
def __init__(self, q: Union[str, etree._Element]):
self.flags: Dict[str, Optional[str]] = dict()
self.flags: QueryDict = dict()
self.messages = Messages()
self.cached: Optional[bool] = None
self.etag = None
@@ -4708,7 +4795,7 @@ class Query():
else:
self.q = q
def _query_request(self, query_dict: Dict[str, Optional[str]]):
def _query_request(self, query_dict: QueryDict):
"""Used internally to execute the query request..."""
_log_request("GET Entity?" + str(query_dict), None)
connection = get_connection()
@@ -4721,7 +4808,7 @@ class Query():
def _paging_generator(
self,
first_page: Container,
query_dict: Dict[str, Optional[str]],
query_dict: QueryDict,
page_length: int,
):
"""Used internally to create a generator of pages instead instead of a
@@ -4838,9 +4925,9 @@ def execute_query(
unique: bool = False,
raise_exception_on_error: bool = True,
cache: bool = True,
flags: Optional[Dict[str, Optional[str]]] = None,
flags: Optional[QueryDict] = None,
page_length: Optional[int] = None,
) -> Union[Container, int]:
) -> Union[Container, Entity, int]:
"""Execute a query (via a server-requests) and return the results.
Parameters
@@ -4879,7 +4966,7 @@ def execute_query(
Returns
-------
results : Container or integer
results : Container or Entity or integer
Returns an integer when it was a `COUNT` query. Otherwise, returns a
Container with the resulting entities.
"""
@@ -4895,8 +4982,7 @@ def execute_query(
class DropOffBox(list):
def __init__(self, *args, **kwargs):
warn(DeprecationWarning(
"The DropOffBox is deprecated and will be removed in future."))
warn(DeprecationWarning("The DropOffBox is deprecated and will be removed in future."))
super().__init__(*args, **kwargs)
path = None
@@ -5006,7 +5092,7 @@ class Permission():
class Permissions():
known_permissions: Optional[List[Permissions]] = None
known_permissions: Optional[Permissions] = None
def __init__(self, xml: etree._Element):
self.parse_xml(xml)
@@ -5121,7 +5207,8 @@ def _parse_single_xml_element(elem: etree._Element):
)
def _evaluate_and_add_error(parent_error: TransactionError, ent: Union[Entity, QueryTemplate, Container]):
def _evaluate_and_add_error(parent_error: TransactionError,
ent: Union[Entity, QueryTemplate, Container]):
"""Evaluate the error message(s) attached to entity and add a
corresponding exception to parent_error.
@@ -5251,7 +5338,7 @@ def raise_errors(arg0: Union[Entity, QueryTemplate, Container]):
raise transaction_error
def delete(ids: Union[List[int], range], raise_exception_on_error: bool = True):
def delete(ids: Union[list[int], range], raise_exception_on_error: bool = True):
c = Container()
if isinstance(ids, list) or isinstance(ids, range):
Loading