diff --git a/src/linkahead/common/models.py b/src/linkahead/common/models.py
index 412c48c306d23e2a52f058ff2e237d7ff7609621..192b2cb5da9a54b8f3f3ea1229648efd6a8db7a3 100644
--- a/src/linkahead/common/models.py
+++ b/src/linkahead/common/models.py
@@ -56,6 +56,8 @@ if TYPE_CHECKING and sys.version_info > (3, 7):
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
@@ -109,8 +111,10 @@ FIX = "FIX"
ALL = "ALL"
NONE = "NONE"
if TYPE_CHECKING:
- INHERITANCE = Literal["OBLIGATORY", "SUGGESTED", "RECOMMENDED", "ALL", "NONE"]
- IMPORTANCE = Literal["OBLIGATORY", "RECOMMENDED", "SUGGESTED", "FIX", "NONE"]
+ INHERITANCE = Literal["OBLIGATORY",
+ "SUGGESTED", "RECOMMENDED", "ALL", "NONE"]
+ IMPORTANCE = Literal["OBLIGATORY",
+ "RECOMMENDED", "SUGGESTED", "FIX", "NONE"]
SPECIAL_ATTRIBUTES = ["name", "role", "datatype", "description",
"id", "path", "checksum", "size", "value"]
@@ -449,7 +453,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,
@@ -479,7 +484,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:
@@ -1045,7 +1051,8 @@ 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
@@ -1747,8 +1754,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 +1769,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 +1785,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 +1802,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")
@@ -1999,7 +2006,7 @@ 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
@@ -2055,7 +2062,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))
@@ -2079,10 +2086,10 @@ class Message(object):
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):
@@ -2266,7 +2273,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 +2285,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 +2295,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 +2350,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()
@@ -2382,7 +2390,8 @@ class _Properties(list):
return self._importance.get(property)
- def set_importance(self, property: Optional[Property], importance: IMPORTANCE): # @ReservedAssignment
+ # @ReservedAssignment
+ def set_importance(self, property: Optional[Property], importance: IMPORTANCE):
if property is not None:
self._importance[property] = importance
@@ -2561,7 +2570,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")
@@ -2734,7 +2743,8 @@ class Messages(list):
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:
@@ -2742,7 +2752,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
@@ -2768,7 +2779,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:
@@ -2821,7 +2833,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)
@@ -2973,7 +2985,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.
@@ -3010,7 +3022,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.
@@ -3043,7 +3056,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.
@@ -3122,7 +3135,8 @@ 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.
@@ -3365,7 +3379,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 = []
@@ -3394,7 +3409,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)
@@ -3419,7 +3435,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)
@@ -3449,7 +3466,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)
@@ -3479,7 +3497,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)
@@ -3492,13 +3511,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)
@@ -3537,7 +3558,7 @@ class Container(list):
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
@@ -3560,7 +3581,7 @@ 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):
@@ -3593,7 +3614,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
@@ -3707,7 +3729,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
@@ -3783,7 +3805,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) +
@@ -3816,7 +3838,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)
@@ -3976,7 +3999,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
@@ -4341,7 +4364,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)
@@ -4683,7 +4706,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
@@ -4709,7 +4732,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()
@@ -4722,7 +4745,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,7 +4861,7 @@ 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]:
"""Execute a query (via a server-requests) and return the results.
@@ -4895,8 +4918,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
@@ -5106,9 +5128,10 @@ def _parse_single_xml_element(elem: etree._Element):
elif elem.tag == "UserInfo":
return UserInfo(xml=elem)
elif elem.tag == "TimeZone":
+ offset = elem.get("offset")
return TimeZone(
zone_id=elem.get("id"),
- offset=elem.get("offset"),
+ offset=int(offset) if offset is not None else None,
display_name=elem.text.strip() if elem.text is not None else "",
)
else: