diff --git a/setup.cfg b/setup.cfg
index c46089e4d24843d7d4cc4f83dad6ec1351e4cc3f..3389b7dca593d0b6350f51ad708de447c789feff 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -1,4 +1,8 @@
[aliases]
test=pytest
+
[pycodestyle]
ignore=E501,E121,E123,E126,E226,E24,E704,W503,W504
+
+[mypy]
+ignore_missing_imports = True
\ No newline at end of file
diff --git a/src/linkahead/apiutils.py b/src/linkahead/apiutils.py
index 4235b28ae46b3394ed22e621f338ce71daeeed7a..4ae8edd16f1fdc00eb7ba2c17661eea6e114885e 100644
--- a/src/linkahead/apiutils.py
+++ b/src/linkahead/apiutils.py
@@ -6,6 +6,8 @@
# Max-Planck-Institute for Dynamics and Self-Organization Göttingen
# Copyright (C) 2020 Timm Fitschen <t.fitschen@indiscale.com>
# Copyright (C) 2020-2022 IndiScale GmbH <info@indiscale.com>
+# Copyright (C) 2024 Indiscale GmbH <info@indiscale.com>
+# Copyright (C) 2024 Joscha Schmiedt <joscha@schmiedt.dev>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
@@ -29,7 +31,7 @@ from __future__ import annotations
import logging
import warnings
from collections.abc import Iterable
-from typing import Any, Dict, List, Union, Optional, Tuple
+from typing import Any, Union, Optional
from .common.datatype import is_reference
from .common.models import (SPECIAL_ATTRIBUTES, Container, Entity, File,
@@ -94,14 +96,14 @@ def new_record(record_type: Union[str],
return r
-def id_query(ids: List[int]) -> Container:
+def id_query(ids: list[int]) -> Container:
warnings.warn("Please use 'create_id_query', which only creates"
"the string.", DeprecationWarning)
return execute_query(create_id_query(ids)) # type: ignore
-def create_id_query(ids: List[int]) -> str:
+def create_id_query(ids: list[int]) -> str:
return "FIND ENTITY WITH " + " OR ".join(
["ID={}".format(id) for id in ids])
@@ -133,7 +135,7 @@ def retrieve_entity_with_id(eid: int):
return execute_query("FIND ENTITY WITH ID={}".format(eid), unique=True)
-def retrieve_entities_with_ids(entities: List) -> Container:
+def retrieve_entities_with_ids(entities: list) -> Container:
collection = Container()
step = 20
@@ -180,7 +182,7 @@ def getCommitIn(folder):
def compare_entities(old_entity: Entity,
new_entity: Entity,
compare_referenced_records: bool = False
- ) -> Tuple[Dict[str, Any], Dict[str, Any]]:
+ ) -> tuple[dict[str, Any], dict[str, Any]]:
"""Compare two entites.
Return a tuple of dictionaries, the first index belongs to additional information for old
@@ -214,8 +216,8 @@ def compare_entities(old_entity: Entity,
identical records are stored in different objects. Default is False.
"""
- olddiff: Dict[str, Any] = {"properties": {}, "parents": []}
- newdiff: Dict[str, Any] = {"properties": {}, "parents": []}
+ olddiff: dict[str, Any] = {"properties": {}, "parents": []}
+ newdiff: dict[str, Any] = {"properties": {}, "parents": []}
if old_entity is new_entity:
return (olddiff, newdiff)
@@ -295,12 +297,15 @@ def compare_entities(old_entity: Entity,
elif isinstance(prop.value, list) and isinstance(matching[0].value, list):
# all elements in both lists actually are entity objects
# TODO: check, whether mixed cases can be allowed or should lead to an error
- if all([isinstance(x, Entity) for x in prop.value]) and all([isinstance(x, Entity) for x in matching[0].value]):
+ if (all([isinstance(x, Entity) for x in prop.value])
+ and all([isinstance(x, Entity) for x in matching[0].value])):
# can't be the same if the lengths are different
if len(prop.value) == len(matching[0].value):
- # do a one-by-one comparison; the values are the same, if all diffs are empty
+ # do a one-by-one comparison:
+ # the values are the same if all diffs are empty
same_value = all(
- [empty_diff(x, y, False) for x, y in zip(prop.value, matching[0].value)])
+ [empty_diff(x, y, False) for x, y
+ in zip(prop.value, matching[0].value)])
if not same_value:
olddiff["properties"][prop.name]["value"] = prop.value
@@ -333,7 +338,8 @@ def compare_entities(old_entity: Entity,
return (olddiff, newdiff)
-def empty_diff(old_entity: Entity, new_entity: Entity, compare_referenced_records: bool = False) -> bool:
+def empty_diff(old_entity: Entity, new_entity: Entity,
+ compare_referenced_records: bool = False) -> bool:
"""Check whether the `compare_entities` found any differences between
old_entity and new_entity.
@@ -606,7 +612,7 @@ def resolve_reference(prop: Property):
prop.value = retrieve_entity_with_id(prop.value)
-def create_flat_list(ent_list: List[Entity], flat: List[Entity]):
+def create_flat_list(ent_list: list[Entity], flat: list[Entity]):
"""
Recursively adds all properties contained in entities from ent_list to
the output list flat. Each element will only be added once to the list.
diff --git a/src/linkahead/cached.py b/src/linkahead/cached.py
index d746a56071e2d472b5ab2afdd98713f98874bebc..cf1d1d34362335f87c5eca094b5aa9d6b750f68d 100644
--- a/src/linkahead/cached.py
+++ b/src/linkahead/cached.py
@@ -5,6 +5,8 @@
# Copyright (C) 2023 IndiScale GmbH <info@indiscale.com>
# Copyright (C) 2023 Henrik tom Wörden <h.tomwoerden@indiscale.com>
# Copyright (C) 2023 Daniel Hornung <d.hornung@indiscale.com>
+# Copyright (C) 2024 Indiscale GmbH <info@indiscale.com>
+# Copyright (C) 2024 Joscha Schmiedt <joscha@schmiedt.dev>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
@@ -35,7 +37,7 @@ See also
from __future__ import annotations
from enum import Enum
from functools import lru_cache
-from typing import Union, Optional, Tuple, Any, Dict
+from typing import Any, Optional, Union
from .exceptions import EmptyUniqueQueryError, QueryNotUniqueError
from .utils import get_entity
@@ -46,7 +48,7 @@ from .common.models import execute_query, Entity, Container
DEFAULT_SIZE = 33333
# This dict cache is solely for filling the real cache manually (e.g. to reuse older query results)
-_DUMMY_CACHE: Dict[Union[str, int], Any] = {}
+_DUMMY_CACHE: dict[Union[str, int], Any] = {}
class AccessType(Enum):
@@ -63,7 +65,7 @@ class AccessType(Enum):
def cached_get_entity_by(eid: Union[str, int, None] = None,
name: Optional[str] = None,
path: Optional[str] = None,
- query: Optional[str] = None) -> Union[Entity, Tuple[None]]:
+ query: Optional[str] = None) -> Union[Entity, tuple[None]]:
"""Return a single entity that is identified uniquely by one argument.
You must supply exactly one argument.
@@ -178,7 +180,7 @@ def cache_initialize(maxsize: int = DEFAULT_SIZE) -> None:
_cached_access = lru_cache(maxsize=maxsize)(_cached_access.__wrapped__)
-def cache_fill(items: Dict[Union[str, int], Any],
+def cache_fill(items: dict[Union[str, int], Any],
kind: AccessType = AccessType.EID,
unique: bool = True) -> None:
"""Add entries to the cache manually.
diff --git a/src/linkahead/common/administration.py b/src/linkahead/common/administration.py
index cb43fc1e07a20dcf34bf2a25089bf821c7d49202..dee341fa84dd85cbd41a77c0e2d510a96f2c4824 100644
--- a/src/linkahead/common/administration.py
+++ b/src/linkahead/common/administration.py
@@ -24,7 +24,7 @@
# ** end header
#
from __future__ import annotations
-"""missing docstring."""
+"""Utility functions for server and user administration."""
import random
import re
@@ -38,7 +38,7 @@ from ..exceptions import (EntityDoesNotExistError, HTTPClientError,
ServerConfigurationException)
from .utils import xml2str
-from typing import Dict, Optional, TYPE_CHECKING, Union
+from typing import Optional, TYPE_CHECKING, Union
if TYPE_CHECKING:
from ..common.models import Entity
@@ -69,7 +69,7 @@ def set_server_property(key: str, value: str):
"Debug mode in server is probably disabled.") from None
-def get_server_properties() -> Dict[str, Optional[str]]:
+def get_server_properties() -> dict[str, Optional[str]]:
"""get_server_properties.
Get all server properties as a dict.
@@ -88,7 +88,7 @@ def get_server_properties() -> Dict[str, Optional[str]]:
"Debug mode in server is probably disabled.") from None
xml = etree.parse(body)
- props: Dict[str, Optional[str]] = dict()
+ props: dict[str, Optional[str]] = dict()
for elem in xml.getroot():
props[elem.tag] = elem.text
@@ -184,7 +184,7 @@ def _update_user(name: str,
email: Optional[str] = None,
entity: Optional[Entity] = None, **kwargs):
con = get_connection()
- params: Dict[str, Optional[str]] = {}
+ params: dict[str, Optional[str]] = {}
if password is not None:
params["password"] = password
@@ -218,7 +218,7 @@ def _insert_user(name: str,
email: Optional[str] = None,
entity: Optional[Entity] = None, **kwargs):
con = get_connection()
- params: Dict[str, Union[str, Entity]] = {"username": name}
+ params: dict[str, Union[str, Entity]] = {"username": name}
if password is not None:
params["password"] = password
diff --git a/src/linkahead/common/datatype.py b/src/linkahead/common/datatype.py
index 5d54865c3cac6ab5bb95f2789f3d6b7eff9f24ca..7afcb7a5beee26a99934640ac41ccf403f9325fe 100644
--- a/src/linkahead/common/datatype.py
+++ b/src/linkahead/common/datatype.py
@@ -24,11 +24,10 @@
#
from __future__ import annotations
import re
-import sys
from typing import TYPE_CHECKING
-if TYPE_CHECKING and sys.version_info > (3, 7):
- from typing import Literal, Union, List
+if TYPE_CHECKING:
+ from typing import Literal, Union
from linkahead.common.models import Entity, Container
DATATYPE = Literal["DOUBLE", "REFERENCE", "TEXT", "DATETIME", "INTEGER", "FILE", "BOOLEAN"]
@@ -46,8 +45,7 @@ BOOLEAN = "BOOLEAN"
def LIST(datatype: Union[str, Entity, DATATYPE]) -> str:
# FIXME May be ambiguous (if name duplicate) or insufficient (if only ID exists).
- if hasattr(datatype, "name"):
- datatype = datatype.name
+ datatype = getattr(datatype, "name", datatype)
return "LIST<" + str(datatype) + ">"
@@ -179,7 +177,7 @@ def get_id_of_datatype(datatype: str) -> int:
res: Container = execute_query(q) # type: ignore
if isinstance(res, int):
raise ValueError("FIND RECORDTYPE query returned an `int`")
- res: List[Entity] = [el for el in res if el.name.lower() == datatype.lower()] # type: ignore
+ res: list[Entity] = [el for el in res if el.name.lower() == datatype.lower()] # type: ignore
if len(res) > 1:
raise QueryNotUniqueError(
diff --git a/src/linkahead/common/models.py b/src/linkahead/common/models.py
index 87645de2d937a08bcd71913dc9e9bd763a2d17de..b4ab36d95d3d4cb800e31e6046ba691fafc15e2b 100644
--- a/src/linkahead/common/models.py
+++ b/src/linkahead/common/models.py
@@ -35,7 +35,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
@@ -50,15 +49,14 @@ from tempfile import NamedTemporaryFile
from typing import TYPE_CHECKING
-# FIXME We don't support py3.7 any more, so the "if" can be removed, right?
-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 typing import Any, Final, Literal, Optional, TextIO, Union
from .datatype import DATATYPE
from tempfile import _TemporaryFileWrapper
from io import BufferedWriter
from os import PathLike
- QueryDict = Dict[str, Optional[str]]
+ QueryDict = dict[str, Optional[str]]
from warnings import warn
@@ -105,12 +103,12 @@ 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"]
@@ -152,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
@@ -172,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.
@@ -248,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
@@ -258,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
@@ -605,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,
@@ -673,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
------
@@ -690,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:
@@ -713,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.
@@ -736,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:
@@ -915,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
@@ -926,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
@@ -1062,7 +1072,7 @@ out: List[Entity]
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
@@ -1156,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)
@@ -1204,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.
@@ -1551,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
----------
@@ -1595,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.
"""
@@ -1970,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``."""
@@ -2006,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,
@@ -2042,7 +2056,8 @@ class Property(Entity):
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
-------
@@ -2112,7 +2127,8 @@ 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
@@ -2126,7 +2142,8 @@ 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``."""
@@ -2204,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``."""
@@ -2406,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]
@@ -2442,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
@@ -2696,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)
@@ -3171,7 +3190,8 @@ class Container(list):
"""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
@@ -3321,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)
@@ -3344,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,
@@ -3410,8 +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 = []
@@ -3659,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)
@@ -3670,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
@@ -3771,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):
@@ -3898,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."""
@@ -3910,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 = {}
@@ -3981,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:
@@ -4059,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 = {}
@@ -4113,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 +
@@ -4220,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.
@@ -4259,7 +4283,7 @@ class Container(list):
return table
-def sync_global_acl():
+def sync_global_acl() -> None:
c = get_connection()
http_response = c.retrieve(entity_uri_segments=["EntityPermissions"])
body = http_response.read()
@@ -4277,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():
@@ -4313,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:
@@ -4411,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) + \
@@ -5061,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)
@@ -5176,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.
@@ -5306,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):
diff --git a/src/linkahead/common/state.py b/src/linkahead/common/state.py
index 7f2d77067799929c1d4c84d8c285b9967eb933c3..e352f82d9820620d1692cb6337eb218210e799e6 100644
--- a/src/linkahead/common/state.py
+++ b/src/linkahead/common/state.py
@@ -26,7 +26,7 @@ from lxml import etree
from typing import TYPE_CHECKING
import sys
-if TYPE_CHECKING and sys.version_info > (3, 7):
+if TYPE_CHECKING:
from typing import Optional
from linkahead.common.models import ACL, ACI
diff --git a/src/linkahead/common/versioning.py b/src/linkahead/common/versioning.py
index ea4fe3e7dcc411083bccd090e959f7b4f7b1635d..2e292e6bb031725fbd6da618c4b888c05072c46b 100644
--- a/src/linkahead/common/versioning.py
+++ b/src/linkahead/common/versioning.py
@@ -26,14 +26,13 @@
Currently this module defines nothing but a single class, `Version`.
"""
-from __future__ import absolute_import, annotations
+from __future__ import annotations
from .utils import xml2str
from lxml import etree
from typing import TYPE_CHECKING
-import sys
-if TYPE_CHECKING and sys.version_info > (3, 7):
- from typing import Optional, List, Union, Literal
+if TYPE_CHECKING:
+ from typing import Optional, List, Union
class Version():
@@ -206,10 +205,8 @@ object."""
version : Version
a new version instance
"""
- predecessors = [Version.from_xml(
- p) for p in xml if p.tag.lower() == "predecessor"]
- successors = [Version.from_xml(s)
- for s in xml if s.tag.lower() == "successor"]
+ predecessors = [Version.from_xml(p) for p in xml if p.tag.lower() == "predecessor"]
+ successors = [Version.from_xml(s) for s in xml if s.tag.lower() == "successor"]
return Version(id=xml.get("id"), date=xml.get("date"),
is_head=xml.get("head"),
is_complete_history=xml.get("completeHistory"),
diff --git a/src/linkahead/configuration.py b/src/linkahead/configuration.py
index adea29aaf164ad225f39901d06fa58729386cb31..b020467c8c53e26d464a6a2fb473cc912b0e0612 100644
--- a/src/linkahead/configuration.py
+++ b/src/linkahead/configuration.py
@@ -37,7 +37,7 @@ from configparser import ConfigParser
from os import environ, getcwd
from os.path import expanduser, isfile, join
-from typing import Dict, Union, Callable, Optional
+from typing import Union, Callable, Optional
_pycaosdbconf = ConfigParser(allow_no_value=False)
@@ -72,8 +72,8 @@ def get_config() -> ConfigParser:
return _pycaosdbconf
-def config_to_yaml(config: ConfigParser) -> Dict[str, Dict[str, Union[int, str, bool]]]:
- valobj: Dict[str, Dict[str, Union[int, str, bool]]] = {}
+def config_to_yaml(config: ConfigParser) -> dict[str, dict[str, Union[int, str, bool]]]:
+ valobj: dict[str, dict[str, Union[int, str, bool]]] = {}
for s in config.sections():
valobj[s] = {}
for key, value in config[s].items():
@@ -88,7 +88,7 @@ def config_to_yaml(config: ConfigParser) -> Dict[str, Dict[str, Union[int, str,
return valobj
-def validate_yaml_schema(valobj: Dict[str, Dict[str, Union[int, str, bool]]]):
+def validate_yaml_schema(valobj: dict[str, dict[str, Union[int, str, bool]]]):
if optional_jsonschema_validate:
with open(os.path.join(os.path.dirname(__file__), "schema-pycaosdb-ini.yml")) as f:
schema = yaml.load(f, Loader=yaml.SafeLoader)
diff --git a/src/linkahead/connection/authentication/input.py b/src/linkahead/connection/authentication/input.py
index 697ca5ddfaf2a1eaf223f0c39f48c94bc1497ba1..14e8196e72a9e6266631511cf36d5e8c0c5c68c4 100644
--- a/src/linkahead/connection/authentication/input.py
+++ b/src/linkahead/connection/authentication/input.py
@@ -25,7 +25,7 @@
A CredentialsProvider which reads the password from the input line.
"""
-from __future__ import absolute_import, unicode_literals, print_function, annotations
+from __future__ import annotations
from .interface import CredentialsProvider, CredentialsAuthenticator
from typing import Optional
import getpass
diff --git a/src/linkahead/connection/authentication/interface.py b/src/linkahead/connection/authentication/interface.py
index cd61427be05af78bdc99be50756a01850e7d913e..b48e27c08312bf1358d32a9a1203627a9d0007c2 100644
--- a/src/linkahead/connection/authentication/interface.py
+++ b/src/linkahead/connection/authentication/interface.py
@@ -1,10 +1,11 @@
# -*- coding: utf-8 -*-
#
-# ** header v3.0
# This file is a part of the LinkAhead Project.
#
# Copyright (C) 2018 Research Group Biomedical Physics,
# Max-Planck-Institute for Dynamics and Self-Organization Göttingen
+# Copyright (C) 2024 Indiscale GmbH <info@indiscale.com>
+# Copyright (C) 2024 Joscha Schmiedt <joscha@schmiedt.dev>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
@@ -18,14 +19,13 @@
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
-#
-# ** end header
-#
+
"""This module provides the interfaces for authenticating requests to the
LinkAhead server.
-Implementing modules muts provide a `get_authentication_provider()` method.
+Implementing modules must provide a `get_authentication_provider()` method.
"""
+
from __future__ import annotations
from abc import ABC, abstractmethod
import logging
@@ -33,10 +33,10 @@ from ..utils import urlencode
from ..interface import CaosDBServerConnection
from ..utils import parse_auth_token, auth_token_to_cookie
from ...exceptions import LoginFailedError
-from typing import TYPE_CHECKING, Dict, Optional
+from typing import TYPE_CHECKING, Optional
if TYPE_CHECKING:
from ..interface import CaosDBHTTPResponse
- QueryDict = Dict[str, Optional[str]]
+ QueryDict = dict[str, Optional[str]]
_LOGGER = logging.getLogger(__name__)
diff --git a/src/linkahead/connection/connection.py b/src/linkahead/connection/connection.py
index 22035aefc124ffdc519f4c078ca71c0d42d5515b..294d9457d064f03bbe06a3347b2d2064dcf12b8c 100644
--- a/src/linkahead/connection/connection.py
+++ b/src/linkahead/connection/connection.py
@@ -1,11 +1,11 @@
# -*- coding: utf-8 -*-
-#
-# ** header v3.0
# This file is a part of the LinkAhead Project.
#
# Copyright (C) 2018 Research Group Biomedical Physics,
# Max-Planck-Institute for Dynamics and Self-Organization Göttingen
# Copyright (c) 2019 Daniel Hornung
+# Copyright (C) 2024 Indiscale GmbH <info@indiscale.com>
+# Copyright (C) 2024 Joscha Schmiedt <joscha@schmiedt.dev>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
@@ -23,7 +23,7 @@
# ** end header
#
"""Connection to a LinkAhead server."""
-from __future__ import absolute_import, print_function, unicode_literals, annotations
+from __future__ import annotations
import logging
import ssl
@@ -33,7 +33,6 @@ from builtins import str # pylint: disable=redefined-builtin
from errno import EPIPE as BrokenPipe
from socket import error as SocketError
from urllib.parse import ParseResult, quote, urlparse
-from warnings import warn
from requests import Session as HTTPSession
from requests.adapters import HTTPAdapter
@@ -52,15 +51,13 @@ try:
except ModuleNotFoundError:
version = "uninstalled"
-from pkg_resources import resource_filename
-
from .encode import MultipartYielder, ReadableMultiparts
from .interface import CaosDBHTTPResponse, CaosDBServerConnection
-from .utils import make_uri_path, parse_url, urlencode
+from .utils import make_uri_path, urlencode
from typing import TYPE_CHECKING
-if TYPE_CHECKING and sys.version_info > (3, 7):
- from typing import Optional, List, Any, Iterator, Dict, Union
+if TYPE_CHECKING:
+ from typing import Optional, Any, Iterator, Union
from requests.models import Response
from ssl import _SSLMethod
from .authentication.interface import AbstractAuthenticator, CredentialsAuthenticator
@@ -171,7 +168,7 @@ class _DefaultCaosDBServerConnection(CaosDBServerConnection):
def request(self,
method: str, path: str,
- headers: Optional[Dict[str, str]] = None,
+ headers: Optional[dict[str, str]] = None,
body: Union[str, bytes, None] = None,
**kwargs) -> _WrappedHTTPResponse:
"""request.
@@ -250,7 +247,8 @@ class _DefaultCaosDBServerConnection(CaosDBServerConnection):
"do so via linkahead.configure_connection(...) or in a config "
"file.")
url_string: str = config["url"]
- if (not url_string.lower().startswith("https://") and not url_string.lower().startswith("http://")):
+ if (not url_string.lower().startswith("https://")
+ and not url_string.lower().startswith("http://")):
raise LinkAheadConnectionError("The connection url is expected "
"to be a http or https url and "
"must include the url scheme "
@@ -289,7 +287,7 @@ class _DefaultCaosDBServerConnection(CaosDBServerConnection):
if "timeout" in config:
self._timeout = config["timeout"]
- def _setup_ssl(self, config: Dict[str, Any]):
+ def _setup_ssl(self, config: dict[str, Any]):
if "ssl_version" in config and config["cacert"] is not None:
ssl_version = getattr(ssl, config["ssl_version"])
else:
@@ -431,7 +429,9 @@ def configure_connection(**kwargs):
auth_token : str (optional)
An authentication token which has been issued by the LinkAhead Server.
- Implies `password_method="auth_token"` if set. An example token string would be `["O","OneTimeAuthenticationToken","anonymous",["administration"],[],1592995200000,604800000,"3ZZ4WKRB-5I7DG2Q6-ZZE6T64P-VQ","197d0d081615c52dc18fb323c300d7be077beaad4020773bb58920b55023fa6ee49355e35754a4277b9ac525c882bcd3a22e7227ba36dfcbbdbf8f15f19d1ee9",1,30000]`.
+ Implies `password_method="auth_token"` if set. An example token string would be
+ ``["O","OneTimeAuthenticationToken","anonymous",["administration"],[],1592995200000,
+ 604800000,"3ZZ4WKRB-5I7DG2Q6-ZZE6T64P-VQ","197d0d081615...1ee9",1,30000]``.
https_proxy : str, optional
Define a proxy for the https connections, e.g. `http://localhost:8888`,
@@ -599,8 +599,8 @@ class _Connection(object): # pylint: disable=useless-object-inheritance
return self
def retrieve(self,
- entity_uri_segments: Optional[List[str]] = None,
- query_dict: Optional[Dict[str, Optional[str]]] = None,
+ entity_uri_segments: Optional[list[str]] = None,
+ query_dict: Optional[dict[str, Optional[str]]] = None,
**kwargs) -> CaosDBHTTPResponse:
path = make_uri_path(entity_uri_segments, query_dict)
@@ -608,8 +608,9 @@ class _Connection(object): # pylint: disable=useless-object-inheritance
return http_response
- def delete(self, entity_uri_segments: Optional[List[str]] = None,
- query_dict: Optional[Dict[str, Optional[str]]] = None, **kwargs) -> CaosDBHTTPResponse:
+ def delete(self, entity_uri_segments: Optional[list[str]] = None,
+ query_dict: Optional[dict[str, Optional[str]]] = None, **kwargs) -> (
+ CaosDBHTTPResponse):
path = make_uri_path(entity_uri_segments, query_dict)
http_response = self._http_request(
@@ -617,8 +618,9 @@ class _Connection(object): # pylint: disable=useless-object-inheritance
return http_response
- def update(self, entity_uri_segment: Optional[List[str]],
- query_dict: Optional[Dict[str, Optional[str]]] = None, **kwargs) -> CaosDBHTTPResponse:
+ def update(self, entity_uri_segment: Optional[list[str]],
+ query_dict: Optional[dict[str, Optional[str]]] = None, **kwargs) -> (
+ CaosDBHTTPResponse):
path = make_uri_path(entity_uri_segment, query_dict)
http_response = self._http_request(method="PUT", path=path, **kwargs)
@@ -640,13 +642,15 @@ class _Connection(object): # pylint: disable=useless-object-inheritance
return self._form_data_request(
method="PUT", path=entity_uri_segment, params=params)
- def post_form_data(self, entity_uri_segment: str, params: Dict[str, Optional[str]]) -> CaosDBHTTPResponse:
+ def post_form_data(self, entity_uri_segment: str, params: dict[str, Optional[str]]) -> (
+ CaosDBHTTPResponse):
return self._form_data_request(
method="POST",
path=entity_uri_segment,
params=params)
- def _form_data_request(self, method: str, path: str, params: Dict[str, Optional[str]]) -> CaosDBHTTPResponse:
+ def _form_data_request(self, method: str, path: str, params: dict[str, Optional[str]]) -> (
+ CaosDBHTTPResponse):
body = urlencode(params)
headers = {}
headers["Content-Type"] = "application/x-www-form-urlencoded"
@@ -658,8 +662,8 @@ class _Connection(object): # pylint: disable=useless-object-inheritance
return response
- def insert(self, entity_uri_segment: Optional[List[str]],
- query_dict: Optional[Dict[str, Optional[str]]] = None,
+ def insert(self, entity_uri_segment: Optional[list[str]],
+ query_dict: Optional[dict[str, Optional[str]]] = None,
body: Union[str, bytes, None] = None, **kwargs) -> CaosDBHTTPResponse:
path = make_uri_path(entity_uri_segment, query_dict)
@@ -686,7 +690,7 @@ class _Connection(object): # pylint: disable=useless-object-inheritance
self._authenticator.logout()
def _http_request(self, method: str, path: str,
- headers: Optional[Dict["str", Any]] = None,
+ headers: Optional[dict["str", Any]] = None,
body: Union[str, bytes, None] = None, **kwargs):
try:
return self._retry_http_request(method=method, path=path,
@@ -713,7 +717,7 @@ class _Connection(object): # pylint: disable=useless-object-inheritance
def _retry_http_request(self,
method: str,
path: str,
- headers: Optional[Dict["str", Any]],
+ headers: Optional[dict["str", Any]],
body: Union[str, bytes, None], **kwargs) -> CaosDBHTTPResponse:
if hasattr(body, "encode") and body is not None:
diff --git a/src/linkahead/connection/encode.py b/src/linkahead/connection/encode.py
index 68de1be316bf448b893d93e50c8262f26fff0c12..a76197803c9652e2d0c4e32819ee3e3f97758bfc 100644
--- a/src/linkahead/connection/encode.py
+++ b/src/linkahead/connection/encode.py
@@ -5,6 +5,8 @@
#
# Copyright (C) 2018 Research Group Biomedical Physics,
# Max-Planck-Institute for Dynamics and Self-Organization Göttingen
+# Copyright (C) 2024 Indiscale GmbH <info@indiscale.com>
+# Copyright (C) 2024 Joscha Schmiedt <joscha@schmiedt.dev>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
@@ -63,8 +65,8 @@ import os
import mimetypes
from email.header import Header
from typing import TYPE_CHECKING
-import sys
-if TYPE_CHECKING and sys.version_info > (3, 7):
+
+if TYPE_CHECKING:
from typing import Optional
@@ -152,14 +154,6 @@ class MultipartParam(object):
except BaseException:
raise ValueError("Could not determine filesize")
- def __cmp__(self, other):
- attrs = [
- 'name', 'value', 'filename', 'filetype', 'filesize', 'fileobj'
- ]
- myattrs = [getattr(self, a) for a in attrs]
- oattrs = [getattr(other, a) for a in attrs]
- return cmp(myattrs, oattrs)
-
def reset(self):
"""Reset the file object's read pointer."""
if self.fileobj is not None:
diff --git a/src/linkahead/connection/interface.py b/src/linkahead/connection/interface.py
index 32e5c8fd340edf14d5653790d3d50cb031b3a56a..fc22577dffb4f2e0d30924324cd7a4901d2c8b1a 100644
--- a/src/linkahead/connection/interface.py
+++ b/src/linkahead/connection/interface.py
@@ -29,7 +29,7 @@ from warnings import warn
from typing import TYPE_CHECKING
if TYPE_CHECKING:
- from typing import Optional, Dict, Union
+ from typing import Optional, Union
class CaosDBHTTPResponse(ABC):
@@ -59,7 +59,7 @@ class CaosDBHTTPResponse(ABC):
"""Status code of the response."""
@abstractmethod
- def getheaders(self) -> Dict[str, str]:
+ def getheaders(self) -> dict[str, str]:
"""Return all headers."""
def __enter__(self):
@@ -85,7 +85,7 @@ class CaosDBServerConnection(ABC):
def request(self,
method: str,
path: str,
- headers: Optional[Dict[str, str]] = None,
+ headers: Optional[dict[str, str]] = None,
body: Union[str, bytes, None] = None,
**kwargs) -> CaosDBHTTPResponse:
"""Abstract method. Implement this method for HTTP requests to the
diff --git a/src/linkahead/connection/utils.py b/src/linkahead/connection/utils.py
index 099db32946895956022a2d1c32837442395e5fb3..deb97f808ea7bb8e6e35a206d1da66e18a39b7eb 100644
--- a/src/linkahead/connection/utils.py
+++ b/src/linkahead/connection/utils.py
@@ -5,6 +5,8 @@
#
# Copyright (C) 2018 Research Group Biomedical Physics,
# Max-Planck-Institute for Dynamics and Self-Organization Göttingen
+# Copyright (C) 2024 Indiscale GmbH <info@indiscale.com>
+# Copyright (C) 2024 Joscha Schmiedt <joscha@schmiedt.dev>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
@@ -22,19 +24,18 @@
# ** end header
#
"""Utility functions for the connection module."""
-from __future__ import unicode_literals, print_function, annotations
-from builtins import str as unicode
-from urllib.parse import (urlencode as _urlencode, quote as _quote,
- urlparse, urlunparse, unquote as _unquote)
+from __future__ import annotations
+
import re
+from urllib.parse import (urlencode as _urlencode, quote as _quote,
+ urlparse, urlunparse, unquote)
from typing import TYPE_CHECKING
-import sys
-if TYPE_CHECKING and sys.version_info > (3, 7):
- from typing import Optional, Dict, List
+if TYPE_CHECKING:
+ from typing import Optional
-def urlencode(query: Dict[str, Optional[str]]) -> str:
+def urlencode(query: dict[str, Optional[str]]) -> str:
"""Convert a dict of into a url-encoded (unicode) string.
This is basically a python2/python3 compatibility wrapper for the respective
@@ -84,8 +85,8 @@ modules when they are called with only the query parameter.
}))
-def make_uri_path(segments: Optional[List[str]] = None,
- query: Optional[Dict[str, Optional[str]]] = None) -> str:
+def make_uri_path(segments: Optional[list[str]] = None,
+ query: Optional[dict[str, Optional[str]]] = None) -> str:
"""Url-encode all segments, concat them with slashes and append the query.
Examples
@@ -141,18 +142,6 @@ def parse_url(url: str):
_PATTERN = re.compile(r"^SessionToken=([^;]*);.*$")
-def unquote(string) -> str:
- """unquote.
-
- Decode an urlencoded string into a plain text string.
- """
- bts = _unquote(string)
- if hasattr(bts, "decode"):
- # python 2
- return bts.decode("utf-8")
- return bts
-
-
def parse_auth_token(cookie: Optional[str]) -> Optional[str]:
"""parse_auth_token.