From b09ad9c1768c86f7837d7be93ca43c37fd5553e5 Mon Sep 17 00:00:00 2001
From: Daniel <d.hornung@indiscale.com>
Date: Fri, 3 May 2024 16:32:33 +0200
Subject: [PATCH] MAINT: Styling, linting
---
setup.cfg | 4 +
src/linkahead/apiutils.py | 30 +--
src/linkahead/cached.py | 10 +-
src/linkahead/common/administration.py | 12 +-
src/linkahead/common/datatype.py | 10 +-
src/linkahead/common/models.py | 202 ++++++++++--------
src/linkahead/common/state.py | 2 +-
src/linkahead/common/versioning.py | 13 +-
src/linkahead/configuration.py | 8 +-
.../connection/authentication/input.py | 2 +-
.../connection/authentication/interface.py | 14 +-
src/linkahead/connection/connection.py | 54 ++---
src/linkahead/connection/encode.py | 14 +-
src/linkahead/connection/interface.py | 6 +-
src/linkahead/connection/utils.py | 33 +--
15 files changed, 220 insertions(+), 194 deletions(-)
diff --git a/setup.cfg b/setup.cfg
index c46089e4..3389b7dc 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 4235b28a..4ae8edd1 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 d746a560..cf1d1d34 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 cb43fc1e..dee341fa 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 5d54865c..7afcb7a5 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 87645de2..b4ab36d9 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 7f2d7706..e352f82d 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 ea4fe3e7..2e292e6b 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 adea29aa..b020467c 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 697ca5dd..14e8196e 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 cd61427b..b48e27c0 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 22035aef..294d9457 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 68de1be3..a7619780 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 32e5c8fd..fc22577d 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 099db329..deb97f80 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.
--
GitLab