diff --git a/CHANGELOG.md b/CHANGELOG.md
index 569d2fc31161d059d191d142783be74c11fc33ac..6969f647584c34742594b01228fbd17160aeb5e3 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -11,6 +11,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
* Support for Python 3.12
* The `linkahead` module now opts into type checking and supports mypy.
+* [#112](https://gitlab.com/linkahead/linkahead-pylib/-/issues/112)
+ `Entity.update_acl` now supports optional `**kwargs` that are passed to the
+ `Entity.update` method that is called internally, thus allowing, e.g.,
+ updating the ACL despite possible naming collisions with `unique=False`.
### Changed ###
diff --git a/Makefile b/Makefile
index 9e4d30dbf8dab85892c220136466360f48d89042..21ea40ac8a6eb34032aba75c089e278fa354a6f5 100644
--- a/Makefile
+++ b/Makefile
@@ -44,7 +44,7 @@ lint:
.PHONY: lint
mypy:
- mypy src/linkahead/common unittests --exclude high_level_api.py --exclude connection.py
+ mypy src/linkahead/common unittests
.PHONY: mypy
unittest:
diff --git a/examples/pycaosdb_example.py b/examples/pycaosdb_example.py
deleted file mode 100755
index 9a3d766791ca7a6fd111d734d08ac4cf3b85b75a..0000000000000000000000000000000000000000
--- a/examples/pycaosdb_example.py
+++ /dev/null
@@ -1,39 +0,0 @@
-#!/usr/bin/env python3
-"""A small example to get started with caosdb-pylib.
-
-Make sure that a `pylinkahead.ini` is readable at one of the expected locations.
-"""
-
-import random
-
-import caosdb as db
-
-
-def reconfigure_connection():
- """Change the current connection configuration."""
- conf = db.configuration.get_config()
- conf.set("Connection", "url", "https://demo.indiscale.com")
- db.configure_connection()
-
-
-def main():
- """Shows a few examples how to use the CaosDB library."""
- conf = dict(db.configuration.get_config().items("Connection"))
- print("##### Config:\n{}\n".format(conf))
-
- if conf["cacert"] == "/path/to/caosdb.ca.pem":
- print("Very likely, the path the the TLS certificate is not correct, "
- "please fix it.")
-
- # Query the server, the result is a Container
- result = db.Query("FIND Record").execute()
- print("##### First query result:\n{}\n".format(result[0]))
-
- # Retrieve a random Record
- rec_id = random.choice([rec.id for rec in result])
- rec = db.Record(id=rec_id).retrieve()
- print("##### Randomly retrieved Record:\n{}\n".format(rec))
-
-
-if __name__ == "__main__":
- main()
diff --git a/examples/pylinkahead_example.py b/examples/pylinkahead_example.py
new file mode 100755
index 0000000000000000000000000000000000000000..6effd57c73669c0aaa0284cb28105ae349dac608
--- /dev/null
+++ b/examples/pylinkahead_example.py
@@ -0,0 +1,57 @@
+#!/usr/bin/env python3
+# This file is a part of the LinkAhead Project.
+#
+# Copyright (C) 2024 IndiScale GmbH <info@indiscale.com>
+# Copyright (C) 2024 Daniel Hornung <d.hornung@indiscale.com>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+#
+# 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/>.
+
+"""A small example to get started with linkahead-pylib.
+
+Make sure that a `pylinkahead.ini` is readable at one of the expected locations.
+"""
+
+import random
+
+import caosdb as db
+
+
+def reconfigure_connection():
+ """Change the current connection configuration."""
+ conf = db.configuration.get_config()
+ conf.set("Connection", "url", "https://demo.indiscale.com")
+ db.configure_connection()
+
+
+def main():
+ """Shows a few examples how to use the LinkAhead library."""
+ conf = dict(db.configuration.get_config().items("Connection"))
+ print("##### Config:\n{}\n".format(conf))
+
+ if conf["cacert"] == "/path/to/caosdb.ca.pem":
+ print("Very likely, the path to the TLS certificate is not correct, "
+ "please fix it.")
+
+ # Query the server, the result is a Container
+ result = db.Query("FIND Record").execute()
+ print("##### First query result:\n{}\n".format(result[0]))
+
+ # Retrieve a random Record
+ rec_id = random.choice([rec.id for rec in result])
+ rec = db.Record(id=rec_id).retrieve()
+ print("##### Randomly retrieved Record:\n{}\n".format(rec))
+
+
+if __name__ == "__main__":
+ main()
diff --git a/setup.cfg b/setup.cfg
index c46089e4d24843d7d4cc4f83dad6ec1351e4cc3f..b7f1a7395a53c32c2e43c72db0b359e4a7aaadb6 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -1,4 +1,11 @@
[aliases]
test=pytest
+
[pycodestyle]
ignore=E501,E121,E123,E126,E226,E24,E704,W503,W504
+
+[mypy]
+ignore_missing_imports = True
+
+# [mypy-linkahead.*]
+# check_untyped_defs = True
\ No newline at end of file
diff --git a/src/linkahead/apiutils.py b/src/linkahead/apiutils.py
index e2ed0facea84e6056b1ac877b4417ce6ad8ef504..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
@@ -25,11 +27,11 @@
"""API-Utils: Some simplified functions for generation of records etc.
"""
-
+from __future__ import annotations
import logging
import warnings
from collections.abc import Iterable
-from typing import Any, Dict, List
+from typing import Any, Union, Optional
from .common.datatype import is_reference
from .common.models import (SPECIAL_ATTRIBUTES, Container, Entity, File,
@@ -47,12 +49,14 @@ class EntityMergeConflictError(LinkAheadException):
"""
-def new_record(record_type, name=None, description=None,
- tempid=None, insert=False, **kwargs):
+def new_record(record_type: Union[str],
+ name: Optional[str] = None,
+ description: Optional[str] = None,
+ tempid: Optional[int] = None,
+ insert: bool = False, **kwargs) -> Record:
"""Function to simplify the creation of Records.
record_type: The name of the RecordType to use for this record.
- (ids should also work.)
name: Name of the new Record.
kwargs: Key-value-pairs for the properties of this Record.
@@ -92,19 +96,19 @@ def new_record(record_type, name=None, description=None,
return r
-def id_query(ids):
+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))
+ return execute_query(create_id_query(ids)) # type: ignore
-def create_id_query(ids):
+def create_id_query(ids: list[int]) -> str:
return "FIND ENTITY WITH " + " OR ".join(
["ID={}".format(id) for id in ids])
-def get_type_of_entity_with(id_):
+def get_type_of_entity_with(id_: int):
objs = retrieve_entities_with_ids([id_])
if len(objs) == 0:
@@ -127,11 +131,11 @@ def get_type_of_entity_with(id_):
return Entity
-def retrieve_entity_with_id(eid):
+def retrieve_entity_with_id(eid: int):
return execute_query("FIND ENTITY WITH ID={}".format(eid), unique=True)
-def retrieve_entities_with_ids(entities):
+def retrieve_entities_with_ids(entities: list) -> Container:
collection = Container()
step = 20
@@ -175,7 +179,10 @@ def getCommitIn(folder):
return get_commit_in(folder)
-def compare_entities(old_entity: Entity, new_entity: Entity, compare_referenced_records: bool = False):
+def compare_entities(old_entity: Entity,
+ new_entity: Entity,
+ compare_referenced_records: bool = False
+ ) -> tuple[dict[str, Any], dict[str, Any]]:
"""Compare two entites.
Return a tuple of dictionaries, the first index belongs to additional information for old
@@ -209,8 +216,8 @@ def compare_entities(old_entity: Entity, new_entity: Entity, compare_referenced_
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)
@@ -290,12 +297,15 @@ def compare_entities(old_entity: Entity, new_entity: Entity, compare_referenced_
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
@@ -328,7 +338,8 @@ def compare_entities(old_entity: Entity, new_entity: Entity, compare_referenced_
return (olddiff, newdiff)
-def empty_diff(old_entity: Entity, new_entity: Entity, compare_referenced_records: bool = False):
+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.
@@ -357,8 +368,12 @@ def empty_diff(old_entity: Entity, new_entity: Entity, compare_referenced_record
return True
-def merge_entities(entity_a: Entity, entity_b: Entity, merge_references_with_empty_diffs=True,
- force=False, merge_id_with_resolved_entity: bool = False):
+def merge_entities(entity_a: Entity,
+ entity_b: Entity,
+ merge_references_with_empty_diffs=True,
+ force=False,
+ merge_id_with_resolved_entity: bool = False
+ ) -> Entity:
"""Merge entity_b into entity_a such that they have the same parents and properties.
datatype, unit, value, name and description will only be changed in entity_a
@@ -441,8 +456,12 @@ def merge_entities(entity_a: Entity, entity_b: Entity, merge_references_with_emp
if merge_id_with_resolved_entity is True and attribute == "value":
# Do a special check for the case of an id value on the
# one hand, and a resolved entity on the other side.
- this = entity_a.get_property(key).value
- that = entity_b.get_property(key).value
+ prop_a = entity_a.get_property(key)
+ assert prop_a is not None, f"Property {key} not found in entity_a"
+ prop_b = entity_b.get_property(key)
+ assert prop_b is not None, f"Property {key} not found in entity_b"
+ this = prop_a.value
+ that = prop_b.value
same = False
if isinstance(this, list) and isinstance(that, list):
if len(this) == len(that):
@@ -465,11 +484,13 @@ def merge_entities(entity_a: Entity, entity_b: Entity, merge_references_with_emp
else:
# TODO: This is a temporary FIX for
# https://gitlab.indiscale.com/caosdb/src/caosdb-pylib/-/issues/105
- entity_a.add_property(id=entity_b.get_property(key).id,
- name=entity_b.get_property(key).name,
- datatype=entity_b.get_property(key).datatype,
- value=entity_b.get_property(key).value,
- unit=entity_b.get_property(key).unit,
+ prop_b = entity_b.get_property(key)
+ assert prop_b is not None, f"Property {key} not found in entity_b"
+ entity_a.add_property(id=prop_b.id,
+ name=prop_b.name,
+ datatype=prop_b.datatype,
+ value=prop_b.value,
+ unit=prop_b.unit,
importance=entity_b.get_importance(key))
# entity_a.add_property(
# entity_b.get_property(key),
@@ -591,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 b27afe0469bcaac733ece4c0be3d8d124f6305c0..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
@@ -32,9 +34,10 @@ See also
- ``cached_get_entity_by(...)`` : Get an Entity by name, id, ...
"""
+from __future__ import annotations
from enum import Enum
from functools import lru_cache
-from typing import Union
+from typing import Any, Optional, Union
from .exceptions import EmptyUniqueQueryError, QueryNotUniqueError
from .utils import get_entity
@@ -45,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 = {}
+_DUMMY_CACHE: dict[Union[str, int], Any] = {}
class AccessType(Enum):
@@ -59,8 +62,10 @@ class AccessType(Enum):
NAME = 4
-def cached_get_entity_by(eid: Union[str, int] = None, name: str = None, path: str = None, query:
- str = None) -> Entity:
+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]]:
"""Return a single entity that is identified uniquely by one argument.
You must supply exactly one argument.
@@ -99,7 +104,7 @@ If a query phrase is given, the result must be unique. If this is not what you
raise RuntimeError("This line should never be reached.")
-def cached_query(query_string) -> Container:
+def cached_query(query_string: str) -> Container:
"""A cached version of :func:`linkahead.execute_query<linkahead.common.models.execute_query>`.
All additional arguments are at their default values.
@@ -111,8 +116,8 @@ All additional arguments are at their default values.
return result
-@lru_cache(maxsize=DEFAULT_SIZE)
-def _cached_access(kind: AccessType, value: Union[str, int], unique=True):
+@ lru_cache(maxsize=DEFAULT_SIZE)
+def _cached_access(kind: AccessType, value: Union[str, int], unique: bool = True):
# This is the function that is actually cached.
# Due to the arguments, the cache has kind of separate sections for cached_query and
# cached_get_entity_by with the different AccessTypes. However, there is only one cache size.
@@ -123,12 +128,24 @@ def _cached_access(kind: AccessType, value: Union[str, int], unique=True):
try:
if kind == AccessType.QUERY:
+ if not isinstance(value, str):
+ raise TypeError(
+ f"If AccessType is QUERY, value must be a string, not {type(value)}.")
return execute_query(value, unique=unique)
if kind == AccessType.NAME:
+ if not isinstance(value, str):
+ raise TypeError(
+ f"If AccessType is NAME, value must be a string, not {type(value)}.")
return get_entity.get_entity_by_name(value)
if kind == AccessType.EID:
+ if not isinstance(value, (str, int)):
+ raise TypeError(
+ f"If AccessType is EID, value must be a string or int, not {type(value)}.")
return get_entity.get_entity_by_id(value)
if kind == AccessType.PATH:
+ if not isinstance(value, str):
+ raise TypeError(
+ f"If AccessType is PATH, value must be a string, not {type(value)}.")
return get_entity.get_entity_by_path(value)
except (QueryNotUniqueError, EmptyUniqueQueryError) as exc:
return exc
@@ -152,7 +169,7 @@ out: named tuple
return _cached_access.cache_info()
-def cache_initialize(maxsize=DEFAULT_SIZE) -> None:
+def cache_initialize(maxsize: int = DEFAULT_SIZE) -> None:
"""Create a new cache with the given size for `cached_query` and `cached_get_entity_by`.
This implies a call of :func:`cache_clear`, the old cache is emptied.
@@ -163,7 +180,9 @@ def cache_initialize(maxsize=DEFAULT_SIZE) -> None:
_cached_access = lru_cache(maxsize=maxsize)(_cached_access.__wrapped__)
-def cache_fill(items: dict, kind: AccessType = AccessType.EID, unique: bool = True) -> None:
+def cache_fill(items: dict[Union[str, int], Any],
+ kind: AccessType = AccessType.EID,
+ unique: bool = True) -> None:
"""Add entries to the cache manually.
This allows to fill the cache without actually submitting queries. Note that this does not
@@ -186,6 +205,19 @@ unique: bool, optional
:func:`cached_query`.
"""
+
+ if kind == AccessType.QUERY:
+ assert all(isinstance(key, str) for key in items.keys()), "Keys must be strings."
+ elif kind == AccessType.NAME:
+ assert all(isinstance(key, str) for key in items.keys()), "Keys must be strings."
+ elif kind == AccessType.EID:
+ assert all(isinstance(key, (str, int))
+ for key in items.keys()), "Keys must be strings or integers."
+ elif kind == AccessType.PATH:
+ assert all(isinstance(key, str) for key in items.keys()), "Keys must be strings."
+ else:
+ raise ValueError(f"Unknown AccessType: {kind}")
+
# 1. add the given items to the corresponding dummy dict cache
_DUMMY_CACHE.update(items)
diff --git a/src/linkahead/common/administration.py b/src/linkahead/common/administration.py
index 417081b0dad19ce15049b8ce05aeef8cc86607f7..dee341fa84dd85cbd41a77c0e2d510a96f2c4824 100644
--- a/src/linkahead/common/administration.py
+++ b/src/linkahead/common/administration.py
@@ -23,8 +23,8 @@
#
# ** end header
#
-
-"""missing docstring."""
+from __future__ import annotations
+"""Utility functions for server and user administration."""
import random
import re
@@ -38,8 +38,12 @@ from ..exceptions import (EntityDoesNotExistError, HTTPClientError,
ServerConfigurationException)
from .utils import xml2str
+from typing import Optional, TYPE_CHECKING, Union
+if TYPE_CHECKING:
+ from ..common.models import Entity
+
-def set_server_property(key, value):
+def set_server_property(key: str, value: str):
"""set_server_property.
Set a server property.
@@ -65,7 +69,7 @@ def set_server_property(key, value):
"Debug mode in server is probably disabled.") from None
-def get_server_properties():
+def get_server_properties() -> dict[str, Optional[str]]:
"""get_server_properties.
Get all server properties as a dict.
@@ -84,7 +88,7 @@ def get_server_properties():
"Debug mode in server is probably disabled.") from None
xml = etree.parse(body)
- props = dict()
+ props: dict[str, Optional[str]] = dict()
for elem in xml.getroot():
props[elem.tag] = elem.text
@@ -92,7 +96,7 @@ def get_server_properties():
return props
-def get_server_property(key):
+def get_server_property(key: str) -> Optional[str]:
"""get_server_property.
Get a server property.
@@ -149,7 +153,7 @@ def generate_password(length: int):
return password
-def _retrieve_user(name, realm=None, **kwargs):
+def _retrieve_user(name: str, realm: Optional[str] = None, **kwargs):
con = get_connection()
try:
return con._http_request(method="GET", path="User/" + (realm + "/" + name if realm is not None else name), **kwargs).read()
@@ -161,7 +165,7 @@ def _retrieve_user(name, realm=None, **kwargs):
raise
-def _delete_user(name, **kwargs):
+def _delete_user(name: str, **kwargs):
con = get_connection()
try:
return con._http_request(method="DELETE", path="User/" + name, **kwargs).read()
@@ -173,10 +177,14 @@ def _delete_user(name, **kwargs):
raise
-def _update_user(name, realm=None, password=None, status=None,
- email=None, entity=None, **kwargs):
+def _update_user(name: str,
+ realm: Optional[str] = None,
+ password: Optional[str] = None,
+ status: Optional[str] = None,
+ email: Optional[str] = None,
+ entity: Optional[Entity] = None, **kwargs):
con = get_connection()
- params = {}
+ params: dict[str, Optional[str]] = {}
if password is not None:
params["password"] = password
@@ -204,9 +212,13 @@ def _update_user(name, realm=None, password=None, status=None,
raise
-def _insert_user(name, password=None, status=None, email=None, entity=None, **kwargs):
+def _insert_user(name: str,
+ password: Optional[str] = None,
+ status: Optional[str] = None,
+ email: Optional[str] = None,
+ entity: Optional[Entity] = None, **kwargs):
con = get_connection()
- params = {"username": name}
+ params: dict[str, Union[str, Entity]] = {"username": name}
if password is not None:
params["password"] = password
@@ -394,15 +406,15 @@ priority : bool, optional
"""
@staticmethod
- def _parse_boolean(bstr):
+ def _parse_boolean(bstr) -> bool:
return str(bstr) in ["True", "true", "TRUE", "yes"]
- def __init__(self, action, permission, priority=False):
+ def __init__(self, action: str, permission: str, priority: bool = False):
self._action = action
self._permission = permission
self._priority = PermissionRule._parse_boolean(priority)
- def _to_xml(self):
+ def _to_xml(self) -> etree._Element:
xml = etree.Element(self._action)
xml.set("permission", self._permission)
@@ -412,12 +424,15 @@ priority : bool, optional
return xml
@staticmethod
- def _parse_element(elem):
- return PermissionRule(elem.tag, elem.get(
- "permission"), elem.get("priority"))
+ def _parse_element(elem: etree._Element):
+ permission = elem.get("permission")
+ if permission is None:
+ raise ValueError(f"Permission is missing in PermissionRule xml: {elem}")
+ priority = PermissionRule._parse_boolean(elem.get("priority"))
+ return PermissionRule(elem.tag, permission, priority if priority is not None else False)
@staticmethod
- def _parse_body(body):
+ def _parse_body(body: str):
xml = etree.fromstring(body)
ret = set()
diff --git a/src/linkahead/common/datatype.py b/src/linkahead/common/datatype.py
index 65e6246c0287f0af07aa604f4bc18ce54615cae2..7afcb7a5beee26a99934640ac41ccf403f9325fe 100644
--- a/src/linkahead/common/datatype.py
+++ b/src/linkahead/common/datatype.py
@@ -22,12 +22,15 @@
#
# ** end header
#
-
+from __future__ import annotations
import re
-import sys
-if sys.version_info >= (3, 8):
- from typing import Literal
+from typing import TYPE_CHECKING
+if TYPE_CHECKING:
+ from typing import Literal, Union
+ from linkahead.common.models import Entity, Container
+ DATATYPE = Literal["DOUBLE", "REFERENCE", "TEXT", "DATETIME", "INTEGER", "FILE", "BOOLEAN"]
+
from ..exceptions import EmptyUniqueQueryError, QueryNotUniqueError
@@ -38,19 +41,16 @@ DATETIME = "DATETIME"
INTEGER = "INTEGER"
FILE = "FILE"
BOOLEAN = "BOOLEAN"
-if sys.version_info >= (3, 8):
- DATATYPE = Literal["DOUBLE", "REFERENCE", "TEXT", "DATETIME", "INTEGER", "FILE", "BOOLEAN"]
-def LIST(datatype):
+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) + ">"
-def get_list_datatype(datatype: str, strict: bool = False):
+def get_list_datatype(datatype: str, strict: bool = False) -> Union[str, None]:
"""Returns the datatype of the elements in the list. If it not a list, return None."""
# TODO Union[str, Entity]
if not isinstance(datatype, str) or not datatype.lower().startswith("list"):
@@ -74,13 +74,13 @@ def get_list_datatype(datatype: str, strict: bool = False):
return None
-def is_list_datatype(datatype):
+def is_list_datatype(datatype: str) -> bool:
""" returns whether the datatype is a list """
return get_list_datatype(datatype) is not None
-def is_reference(datatype):
+def is_reference(datatype: str) -> bool:
"""Returns whether the value is a reference
FILE and REFERENCE properties are examples, but also datatypes that are
@@ -105,12 +105,12 @@ def is_reference(datatype):
if datatype in [DOUBLE, BOOLEAN, INTEGER, TEXT, DATETIME]:
return False
elif is_list_datatype(datatype):
- return is_reference(get_list_datatype(datatype))
+ return is_reference(get_list_datatype(datatype)) # type: ignore
else:
return True
-def get_referenced_recordtype(datatype):
+def get_referenced_recordtype(datatype: str) -> str:
"""Return the record type of the referenced datatype.
Raises
@@ -134,7 +134,7 @@ def get_referenced_recordtype(datatype):
raise ValueError("datatype must be a reference")
if is_list_datatype(datatype):
- datatype = get_list_datatype(datatype)
+ datatype = get_list_datatype(datatype) # type: ignore
if datatype is None:
raise ValueError("list does not have a list datatype")
@@ -145,7 +145,7 @@ def get_referenced_recordtype(datatype):
return datatype
-def get_id_of_datatype(datatype):
+def get_id_of_datatype(datatype: str) -> int:
""" returns the id of a Record Type
This is not trivial, as queries may also return children. A check comparing
@@ -170,12 +170,14 @@ def get_id_of_datatype(datatype):
from .models import execute_query
if is_list_datatype(datatype):
- datatype = get_list_datatype(datatype)
+ datatype = get_list_datatype(datatype) # type: ignore
q = "FIND RECORDTYPE {}".format(datatype)
# we cannot use unique=True here, because there might be subtypes
- res = execute_query(q)
- res = [el for el in res if el.name.lower() == datatype.lower()]
+ 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
if len(res) > 1:
raise QueryNotUniqueError(
diff --git a/src/linkahead/common/models.py b/src/linkahead/common/models.py
index 970d373a6cda57e87bf5abfdf7362c24e911d42f..cf803d84232c7e85c6b42e88477327ee144fac00 100644
--- a/src/linkahead/common/models.py
+++ b/src/linkahead/common/models.py
@@ -34,8 +34,6 @@ transactions.
"""
from __future__ import annotations # Can be removed with 3.10.
-from __future__ import print_function, unicode_literals
-from enum import Enum
import re
import sys
@@ -49,13 +47,15 @@ from random import randint
from tempfile import NamedTemporaryFile
from typing import TYPE_CHECKING
+from typing import Any, Final, Literal, Optional, TextIO, Union
-if TYPE_CHECKING and sys.version_info > (3, 7):
+if TYPE_CHECKING:
from datetime import datetime
- from typing import Any, Dict, Optional, Type, Union, List, TextIO, Tuple, Literal
from .datatype import DATATYPE
from tempfile import _TemporaryFileWrapper
from io import BufferedWriter
+ from os import PathLike
+ QueryDict = dict[str, Optional[str]]
from warnings import warn
@@ -102,15 +102,17 @@ from .versioning import Version
_ENTITY_URI_SEGMENT = "Entity"
-OBLIGATORY = "OBLIGATORY"
-SUGGESTED = "SUGGESTED"
-RECOMMENDED = "RECOMMENDED"
-FIX = "FIX"
-ALL = "ALL"
-NONE = "NONE"
+OBLIGATORY: Final = "OBLIGATORY"
+SUGGESTED: Final = "SUGGESTED"
+RECOMMENDED: Final = "RECOMMENDED"
+FIX: Final = "FIX"
+ALL: Final = "ALL"
+NONE: Final = "NONE"
+
if TYPE_CHECKING:
INHERITANCE = Literal["OBLIGATORY", "SUGGESTED", "RECOMMENDED", "ALL", "NONE", "FIX"]
IMPORTANCE = Literal["OBLIGATORY", "RECOMMENDED", "SUGGESTED", "FIX", "NONE"]
+ ROLE = Literal["Entity", "Record", "RecordType", "Property", "File"]
SPECIAL_ATTRIBUTES = ["name", "role", "datatype", "description",
"id", "path", "checksum", "size", "value"]
@@ -139,7 +141,7 @@ class Entity:
**kwargs,
):
- self.__role = kwargs["role"] if "role" in kwargs else None
+ self.__role: Optional[ROLE] = kwargs["role"] if "role" in kwargs else None
self._checksum: Optional[str] = None
self._size = None
self._upload = None
@@ -148,7 +150,7 @@ class Entity:
self._wrapped_entity: Optional[Entity] = None
self._version: Optional[Version] = None
self._cuid: Optional[str] = None
- self._flags: Dict[str, str] = dict()
+ self._flags: dict[str, str] = dict()
self.__value = None
self.__datatype: Optional[DATATYPE] = None
self.datatype: Optional[DATATYPE] = datatype
@@ -168,7 +170,7 @@ class Entity:
self.id: Optional[int] = id
self.state: Optional[State] = None
- def copy(self):
+ def copy(self) -> Entity:
"""
Return a copy of entity.
@@ -179,6 +181,7 @@ class Entity:
Special attributes, as defined by the global variable SPECIAL_ATTRIBUTES and additionaly
the "value" are copied using setattr.
"""
+ new: Union[File, Property, RecordType, Record, Entity]
if self.role == "File":
new = File()
elif self.role == "Property":
@@ -243,7 +246,7 @@ class Entity:
return self._wrapped_entity.size
@property
- def id(self):
+ def id(self) -> Any:
if self.__id is not None:
return self.__id
@@ -253,9 +256,9 @@ class Entity:
return self._wrapped_entity.id
@id.setter
- def id(self, new_id):
+ def id(self, new_id) -> None:
if new_id is not None:
- self.__id = int(new_id)
+ self.__id: Optional[int] = int(new_id)
else:
self.__id = None
@@ -449,7 +452,8 @@ class Entity:
"""
# @review Florian Spreckelsen 2022-03-17
if self.acl is None:
- raise EntityHasNoAclError("This entity does not have an ACL (yet).")
+ raise EntityHasNoAclError(
+ "This entity does not have an ACL (yet).")
self.acl.deny(realm=realm, username=username, role=role,
permission=permission, priority=priority,
@@ -457,6 +461,8 @@ class Entity:
def revoke_denial(self, realm=None, username=None,
role=None, permission=None, priority=False):
+ if self.acl is None:
+ raise EntityHasNoAclError("This entity does not have an ACL (yet).")
self.acl.revoke_denial(
realm=realm,
username=username,
@@ -466,6 +472,8 @@ class Entity:
def revoke_grant(self, realm=None, username=None,
role=None, permission=None, priority=False):
+ if self.acl is None:
+ raise EntityHasNoAclError("This entity does not have an ACL (yet).")
self.acl.revoke_grant(
realm=realm,
username=username,
@@ -479,7 +487,8 @@ class Entity:
return permission in self.permissions
if self.acl is None:
- raise EntityHasNoAclError("This entity does not have an ACL (yet).")
+ raise EntityHasNoAclError(
+ "This entity does not have an ACL (yet).")
return self.acl.is_permitted(role=role, permission=permission)
def get_all_messages(self) -> Messages:
@@ -594,10 +603,10 @@ class Entity:
bool,
datetime,
Entity,
- List[int],
- List[str],
- List[bool],
- List[Entity],
+ list[int],
+ list[str],
+ list[bool],
+ list[Entity],
None,
] = None,
id: Optional[int] = None,
@@ -662,8 +671,8 @@ class Entity:
If the first parameter is an integer then it is interpreted as the id and id must be
undefined or None.
UserWarning
- If the first parameter is not None and neither an instance of Entity nor an integer it is
- interpreted as the name and name must be undefined or None.
+ If the first parameter is not None and neither an instance of Entity nor an integer it
+ is interpreted as the name and name must be undefined or None.
Raises
------
@@ -679,7 +688,8 @@ class Entity:
>>> import linkahead as db
>>> rec = db.Record(name="TestRec").add_parent(name="TestType")
- >>> rec.add_property("TestProp", value=27) # specified by name, you could equally use the property's id if it is known
+ >>> rec.add_property("TestProp", value=27) # specified by name, you could equally use the
+ >>> # property's id if it is known
You can also use the Python object:
@@ -702,10 +712,12 @@ class Entity:
Note that since `TestProp` is a scalar integer Property, the datatype
`LIST<INTEGER>` has to be specified explicitly.
- Finally, we can also add reference properties, specified by the RecordType of the referenced entity.
+ Finally, we can also add reference properties, specified by the RecordType of the referenced
+ entity.
>>> ref_rec = db.Record(name="ReferencedRecord").add_parent(name="OtherRT")
- >>> rec.add_property(name="OtherRT", value=ref_rec) # or value=ref_rec.id if ref_rec has one set by the server
+ >>> rec.add_property(name="OtherRT", value=ref_rec) # or value=ref_rec.id if ref_rec has
+ >>> # one set by the server
See more on adding properties and inserting data in
https://docs.indiscale.com/caosdb-pylib/tutorials/Data-Insertion.html.
@@ -725,12 +737,21 @@ class Entity:
abstract_property = property
elif isinstance(property, int):
if pid is not None:
- raise UserWarning("The first parameter was an integer which would normally be interpreted as the id of the property which is to be added. But you have also specified a parameter 'id' in the method call. This is ambiguous and cannot be processed.")
+ raise UserWarning(
+ "The first parameter was an integer which would normally be interpreted as the"
+ " id of the property which is to be added. But you have also specified a"
+ " parameter 'id' in the method call. This is ambiguous and cannot be processed."
+ )
pid = property
id = pid
elif property is not None:
if name is not None:
- raise UserWarning("The first parameter was neither an instance of Entity nor an integer. Therefore the string representation of your first parameter would normally be interpreted name of the property which is to be added. But you have also specified a parameter 'name' in the method call. This is ambiguous and cannot be processed.")
+ raise UserWarning(
+ "The first parameter was neither an instance of Entity nor an integer."
+ " Therefore the string representation of your first parameter would normally be"
+ " interpreted name of the property which is to be added. But you have also"
+ " specified a parameter 'name' in the method call. This is ambiguous and cannot"
+ " be processed.")
name = str(property)
if property is None and name is None and pid is None:
@@ -904,7 +925,7 @@ out: bool
return self.parents
- def get_parents_recursively(self, retrieve: bool = True) -> List[Entity]:
+ def get_parents_recursively(self, retrieve: bool = True) -> list[Entity]:
"""Get all ancestors of this entity.
Parameters
@@ -915,16 +936,16 @@ retrieve: bool, optional
Returns
-------
-out: List[Entity]
+out: list[Entity]
The parents of this Entity
"""
- all_parents: List[Entity] = []
+ all_parents: list[Entity] = []
self._get_parent_recursively(all_parents, retrieve=retrieve)
return all_parents
- def _get_parent_recursively(self, all_parents: List[Entity], retrieve: bool = True):
+ def _get_parent_recursively(self, all_parents: list[Entity], retrieve: bool = True):
"""Get all ancestors with a little helper.
As a side effect of this method, the ancestors are added to
@@ -1045,12 +1066,13 @@ out: List[Entity]
return p
else:
- raise ValueError("`pattern` argument should be an Entity, int or str.")
+ raise ValueError(
+ "`pattern` argument should be an Entity, int or str.")
return None
def _get_value_for_selector(
- self, selector: Union[str, List[str], Tuple[str]]
+ self, selector: Union[str, list[str], tuple[str]]
) -> Any:
"""return the value described by the selector
@@ -1144,7 +1166,7 @@ out: List[Entity]
row : tuple
A row-like representation of the entity's properties.
"""
- row = tuple()
+ row: tuple = tuple()
for selector in selectors:
val = self._get_value_for_selector(selector)
@@ -1192,7 +1214,7 @@ out: List[Entity]
return ret
- def get_errors_deep(self, roots=None) -> List[Tuple[str, List[Entity]]]:
+ def get_errors_deep(self, roots=None) -> list[tuple[str, list[Entity]]]:
"""Get all error messages of this entity and all sub-entities /
parents / properties.
@@ -1437,7 +1459,30 @@ out: List[Entity]
self.acl = Entity(name=self.name, id=self.id).retrieve(
flags={"ACL": None}).acl
- def update_acl(self):
+ def update_acl(self, **kwargs):
+ """Update this entity's ACL on the server.
+
+ A typical workflow is to first edit ``self.acl`` and then call this
+ method.
+
+ Note
+ ----
+ This overwrites any existing ACL, so you may want to run
+ ``retrieve_acl`` before updating the ACL in this entity.
+
+ Parameters
+ ----------
+ **kwargs : dict
+ Keyword arguments that are passed through to the
+ ``Entity.update`` method. Useful for e.g. ``unique=False`` in the
+ case of naming collisions.
+
+ Returns
+ -------
+ e : Entity
+ This entity after the update of the ACL.
+
+ """
if self.id is None:
c = Container().retrieve(query=self.name, sync=False)
@@ -1457,8 +1502,10 @@ out: List[Entity]
raise TransactionError(ae)
else:
e = Container().retrieve(query=self.id, sync=False)[0]
+ if self.acl is None:
+ raise EntityHasNoAclError("This entity does not have an ACL yet. Please set one first.")
e.acl = ACL(self.acl.to_xml())
- e.update()
+ e.update(**kwargs)
return e
@@ -1514,13 +1561,14 @@ out: List[Entity]
identified, retrieved, updated, and deleted via this ID until it has
been deleted.
- If the insertion fails, a LinkAheadException will be raised. The server will have returned at
- least one error-message describing the reason why it failed in that case (call
+ If the insertion fails, a LinkAheadException will be raised. The server will have returned
+ at least one error-message describing the reason why it failed in that case (call
<this_entity>.get_all_messages() in order to get these error-messages).
- Some insertions might cause warning-messages on the server-side, but the entities are inserted
- anyway. Set the flag 'strict' to True in order to force the server to take all warnings as errors.
- This prevents the server from inserting this entity if any warning occurs.
+ Some insertions might cause warning-messages on the server-side, but the entities are
+ inserted anyway. Set the flag 'strict' to True in order to force the server to take all
+ warnings as errors. This prevents the server from inserting this entity if any warning
+ occurs.
Parameters
----------
@@ -1558,21 +1606,22 @@ Second:
1) construct entity with id
2) call update method.
- For slight changes the second one it is more comfortable. Furthermore, it is possible to stay
- off-line until calling the update method. The name, description, unit, datatype, path,
- and value of an entity may be changed. Additionally, properties, parents and messages may be added.
+ For slight changes the second one it is more comfortable. Furthermore, it is possible to
+ stay off-line until calling the update method. The name, description, unit, datatype, path,
+ and value of an entity may be changed. Additionally, properties, parents and messages may be
+ added.
- However, the first one is more powerful: It is possible to delete and change properties, parents
- and attributes, which is not possible via the second one for internal reasons (which are reasons
- of definiteness).
+ However, the first one is more powerful: It is possible to delete and change properties,
+ parents and attributes, which is not possible via the second one for internal reasons (which
+ are reasons of definiteness).
If the update fails, a LinkAheadException will be raised. The server will have returned at
least one error message describing the reason why it failed in that case (call
<this_entity>.get_all_messages() in order to get these error-messages).
Some updates might cause warning messages on the server-side, but the updates are performed
- anyway. Set flag 'strict' to True in order to force the server to take all warnings as errors.
- This prevents the server from updating this entity if any warnings occur.
+ anyway. Set flag 'strict' to True in order to force the server to take all warnings as
+ errors. This prevents the server from updating this entity if any warnings occur.
@param strict=False: Flag for strict mode.
"""
@@ -1747,8 +1796,8 @@ class QueryTemplate():
raise_exception_on_error: bool = True,
unique: bool = True,
sync: bool = True,
- flags: Optional[Dict[str, Optional[str]]] = None,
- ):
+ flags: Optional[QueryDict] = None,
+ ) -> Container:
return Container().append(self).retrieve(
raise_exception_on_error=raise_exception_on_error,
@@ -1762,8 +1811,8 @@ class QueryTemplate():
raise_exception_on_error: bool = True,
unique: bool = True,
sync: bool = True,
- flags: Optional[Dict[str, Optional[str]]] = None,
- ):
+ flags: Optional[QueryDict] = None,
+ ) -> Container:
return Container().append(self).insert(
strict=strict,
@@ -1778,8 +1827,8 @@ class QueryTemplate():
raise_exception_on_error: bool = True,
unique: bool = True,
sync: bool = True,
- flags: Optional[Dict[str, Optional[str]]] = None,
- ):
+ flags: Optional[QueryDict] = None,
+ ) -> Container:
return Container().append(self).update(
strict=strict,
@@ -1795,7 +1844,7 @@ class QueryTemplate():
def __repr__(self):
return xml2str(self.to_xml())
- def to_xml(self, xml: Optional[etree._Element] = None):
+ def to_xml(self, xml: Optional[etree._Element] = None) -> etree._Element:
if xml is None:
xml = etree.Element("QueryTemplate")
@@ -1933,7 +1982,8 @@ class Property(Entity):
"""LinkAhead's Property object."""
- def add_property(self, property=None, value=None, id=None, name=None, description=None, datatype=None,
+ def add_property(self, property=None, value=None, id=None, name=None, description=None,
+ datatype=None,
unit=None, importance=FIX, inheritance=FIX): # @ReservedAssignment
"""See ``Entity.add_property``."""
@@ -1969,7 +2019,8 @@ class Property(Entity):
"""
- return super(Property, self).add_parent(parent=parent, id=id, name=name, inheritance=inheritance)
+ return super(Property, self).add_parent(parent=parent, id=id, name=name,
+ inheritance=inheritance)
def __init__(
self,
@@ -1999,13 +2050,14 @@ class Property(Entity):
local_serialization=local_serialization,
)
- def is_reference(self, server_retrieval=False):
+ def is_reference(self, server_retrieval: bool = False) -> Optional[bool]:
"""Returns whether this Property is a reference
Parameters
----------
server_retrieval : bool, optional
- If True and the datatype is not set, the Property is retrieved from the server, by default False
+ If True and the datatype is not set, the Property is retrieved from the server, by
+ default False
Returns
-------
@@ -2055,7 +2107,7 @@ class Message(object):
self.code = int(code) if code is not None else None
self.body = body
- def to_xml(self, xml: Optional[etree._Element] = None):
+ def to_xml(self, xml: Optional[etree._Element] = None) -> etree._Element:
if xml is None:
xml = etree.Element(str(self.type))
@@ -2075,21 +2127,23 @@ class Message(object):
def __eq__(self, obj):
if isinstance(obj, Message):
- return self.type == obj.type and self.code == obj.code and self.description == obj.description
+ return (self.type == obj.type and self.code == obj.code
+ and self.description == obj.description)
return False
- def get_code(self):
+ def get_code(self) -> Optional[int]:
warn(("get_code is deprecated and will be removed in future. "
"Use self.code instead."), DeprecationWarning)
- return int(self.code)
+ return int(self.code) if self.code is not None else None
class RecordType(Entity):
"""This class represents LinkAhead's RecordType entities."""
- def add_property(self, property=None, value=None, id=None, name=None, description=None, datatype=None,
+ def add_property(self, property=None, value=None, id=None, name=None, description=None,
+ datatype=None,
unit=None, importance=RECOMMENDED, inheritance=FIX): # @ReservedAssignment
"""See ``Entity.add_property``."""
@@ -2167,7 +2221,8 @@ class Record(Entity):
"""This class represents LinkAhead's Record entities."""
- def add_property(self, property=None, value=None, id=None, name=None, description=None, datatype=None,
+ def add_property(self, property=None, value=None, id=None, name=None, description=None,
+ datatype=None,
unit=None, importance=FIX, inheritance=FIX): # @ReservedAssignment
"""See ``Entity.add_property``."""
@@ -2266,7 +2321,7 @@ class File(Record):
xml: Optional[etree._Element] = None,
add_properties: INHERITANCE = "ALL",
local_serialization: bool = False,
- ):
+ ) -> etree._Element:
"""Convert this file to an xml element.
@return: xml element
@@ -2278,7 +2333,7 @@ class File(Record):
return Entity.to_xml(self, xml=xml, add_properties=add_properties,
local_serialization=local_serialization)
- def download(self, target: Optional[str] = None):
+ def download(self, target: Optional[str] = None) -> str:
"""Download this file-entity's actual file from the file server. It
will be stored to the target or will be hold as a temporary file.
@@ -2288,7 +2343,8 @@ class File(Record):
self.clear_server_messages()
if target:
- file_: Union[BufferedWriter, _TemporaryFileWrapper] = open(target, "wb")
+ file_: Union[BufferedWriter,
+ _TemporaryFileWrapper] = open(target, "wb")
else:
file_ = NamedTemporaryFile(mode='wb', delete=False)
checksum = File.download_from_path(file_, self.path)
@@ -2342,7 +2398,7 @@ class File(Record):
return File._get_checksum_single_file(files)
@staticmethod
- def _get_checksum_single_file(single_file):
+ def _get_checksum_single_file(single_file: Union[str, bytes, PathLike[str], PathLike[bytes]]):
_file = open(single_file, 'rb')
data = _file.read(1000)
checksum = sha512()
@@ -2368,10 +2424,10 @@ class _Properties(list):
def __init__(self):
list.__init__(self)
- self._importance: Dict[Entity, IMPORTANCE] = dict()
- self._inheritance: Dict[Entity, INHERITANCE] = dict()
- self._element_by_name: Dict[str, Entity] = dict()
- self._element_by_id: Dict[str, Entity] = dict()
+ self._importance: dict[Entity, IMPORTANCE] = dict()
+ self._inheritance: dict[Entity, INHERITANCE] = dict()
+ self._element_by_name: dict[str, Entity] = dict()
+ self._element_by_id: dict[str, Entity] = dict()
def get_importance(
self, property: Union[Property, Entity, str, None]
@@ -2404,7 +2460,7 @@ class _Properties(list):
def append(
self,
- property: Union[List[Entity], Entity, Property],
+ property: Union[list[Entity], Entity, Property],
importance: Optional[IMPORTANCE] = None,
inheritance: Optional[INHERITANCE] = None,
): # @ReservedAssignment
@@ -2435,6 +2491,7 @@ class _Properties(list):
return self
def to_xml(self, add_to_element: etree._Element, add_properties: INHERITANCE):
+ p: Property
for p in self:
importance = self._importance.get(p)
@@ -2562,7 +2619,7 @@ class _ParentList(list):
return self
- def to_xml(self, add_to_element):
+ def to_xml(self, add_to_element: etree._Element):
for p in self:
pelem = etree.Element("Parent")
@@ -2657,7 +2714,8 @@ class Messages(list):
<<< msgs = Messages()
<<< # create Message
- <<< msg = Message(type="HelloWorld", code=1, description="Greeting the world", body="Hello, world!")
+ <<< msg = Message(type="HelloWorld", code=1, description="Greeting the world",
+ ... body="Hello, world!")
<<< # append it to the Messages
<<< msgs.append(msg)
@@ -2731,11 +2789,12 @@ class Messages(list):
if isinstance(value, Message):
body = value.body
description = value.description
- m = Message
+ m = Message()
else:
body = value
description = None
- m = Message(type=type, code=code, description=description, body=body)
+ m = Message(type=type, code=code,
+ description=description, body=body)
if isinstance(key, int):
super().__setitem__(key, m)
else:
@@ -2743,7 +2802,8 @@ class Messages(list):
def __getitem__(self, key):
if not isinstance(key, int):
- warn("__getitem__ only supports integer keys in future.", DeprecationWarning)
+ warn("__getitem__ only supports integer keys in future.",
+ DeprecationWarning)
if isinstance(key, tuple):
if len(key) == 2:
type = key[0] # @ReservedAssignment
@@ -2769,7 +2829,8 @@ class Messages(list):
def __delitem__(self, key):
if isinstance(key, tuple):
- warn("__delitem__ only supports integer keys in future.", DeprecationWarning)
+ warn("__delitem__ only supports integer keys in future.",
+ DeprecationWarning)
if self.get(key[0], key[1]) is not None:
self.remove(self.get(key[0], key[1]))
else:
@@ -2822,7 +2883,7 @@ class Messages(list):
return default
- def to_xml(self, add_to_element):
+ def to_xml(self, add_to_element: etree._Element):
for m in self:
melem = m.to_xml()
add_to_element.append(melem)
@@ -2974,7 +3035,7 @@ class Container(list):
def __hash__(self):
return object.__hash__(self)
- def remove(self, entity):
+ def remove(self, entity: Entity):
"""Remove the first entity from this container which is equal to the
given entity. Raise a ValueError if there is no such entity.
@@ -3011,7 +3072,8 @@ class Container(list):
return e
raise KeyError("No entity with such cuid (" + str(cuid) + ")!")
- def get_entity_by_id(self, id): # @ReservedAssignment
+ # @ReservedAssignment
+ def get_entity_by_id(self, id: Union[int, str]) -> Entity:
"""Get the first entity which has the given id. Note: If several
entities are in this list which have the same id, this method will only
return the first and ignore the others.
@@ -3044,7 +3106,7 @@ class Container(list):
return error_list
- def get_entity_by_name(self, name: str, case_sensitive: bool = True):
+ def get_entity_by_name(self, name: str, case_sensitive: bool = True) -> Entity:
"""Get the first entity which has the given name. Note: If several
entities are in this list which have the same name, this method will
only return the first and ignore the others.
@@ -3123,11 +3185,13 @@ class Container(list):
return self
- def to_xml(self, add_to_element=None, local_serialization=False):
+ def to_xml(self, add_to_element: Optional[etree._Element] = None,
+ local_serialization: bool = False) -> etree._Element:
"""Get an xml tree representing this Container or append all entities
to the given xml element.
- @param add_to_element=None: optional element to which all entities of this container is to be appended.
+ @param add_to_element=None: optional element to which all entities of this container is to
+ be appended.
@return xml element
"""
tmpid = 0
@@ -3277,7 +3341,7 @@ class Container(list):
if isinstance(e, Message):
c.messages.append(e)
elif isinstance(e, Query):
- c.query = e
+ c.query = e # type: ignore
if e.messages is not None:
c.messages.extend(e.messages)
@@ -3300,7 +3364,8 @@ class Container(list):
return c
else:
raise LinkAheadException(
- "The server's response didn't contain the expected elements. The configuration of this client might be invalid (especially the url).")
+ "The server's response didn't contain the expected elements. The configuration of"
+ " this client might be invalid (especially the url).")
def _sync(
self,
@@ -3366,7 +3431,8 @@ class Container(list):
# which is to be synced with which:
# sync_dict[local_entity]=sync_remote_enities
- sync_dict: Dict[Union[Container, Entity], Optional[List[Entity]]] = dict()
+ sync_dict: dict[Union[Container, Entity],
+ Optional[list[Entity]]] = dict()
# list of remote entities which already have a local equivalent
used_remote_entities = []
@@ -3395,7 +3461,8 @@ class Container(list):
msg = "Request was not unique. CUID " + \
str(local_entity._cuid) + " was found " + \
str(len(sync_remote_entities)) + " times."
- local_entity.add_message(Message(description=msg, type="Error"))
+ local_entity.add_message(
+ Message(description=msg, type="Error"))
if raise_exception_on_error:
raise MismatchingEntitiesError(msg)
@@ -3420,7 +3487,8 @@ class Container(list):
msg = "Request was not unique. ID " + \
str(local_entity.id) + " was found " + \
str(len(sync_remote_entities)) + " times."
- local_entity.add_message(Message(description=msg, type="Error"))
+ local_entity.add_message(
+ Message(description=msg, type="Error"))
if raise_exception_on_error:
raise MismatchingEntitiesError(msg)
@@ -3450,7 +3518,8 @@ class Container(list):
msg = "Request was not unique. Path " + \
str(local_entity.path) + " was found " + \
str(len(sync_remote_entities)) + " times."
- local_entity.add_message(Message(description=msg, type="Error"))
+ local_entity.add_message(
+ Message(description=msg, type="Error"))
if raise_exception_on_error:
raise MismatchingEntitiesError(msg)
@@ -3480,7 +3549,8 @@ class Container(list):
msg = "Request was not unique. Name " + \
str(local_entity.name) + " was found " + \
str(len(sync_remote_entities)) + " times."
- local_entity.add_message(Message(description=msg, type="Error"))
+ local_entity.add_message(
+ Message(description=msg, type="Error"))
if raise_exception_on_error:
raise MismatchingEntitiesError(msg)
@@ -3493,13 +3563,15 @@ class Container(list):
sync_remote_entities.append(remote_entity)
if len(sync_remote_entities) > 0:
- sync_dict[self] = sync_remote_entities # FIXME: How is this supposed to work?
+ # FIXME: How is this supposed to work?
+ sync_dict[self] = sync_remote_entities
if unique and len(sync_remote_entities) != 0:
msg = "Request was not unique. There are " + \
str(len(sync_remote_entities)) + \
" entities which could not be matched to one of the requested ones."
- remote_container.add_message(Message(description=msg, type="Error"))
+ remote_container.add_message(
+ Message(description=msg, type="Error"))
if raise_exception_on_error:
raise MismatchingEntitiesError(msg)
@@ -3529,14 +3601,16 @@ class Container(list):
dependent_references = set()
dependencies = set()
+ container_item: Entity
for container_item in container:
item_id.add(container_item.id)
for parents in container_item.get_parents():
is_parent.add(parents.id)
+ prop: Property
for prop in container_item.get_properties():
- prop_dt = prop.datatype
+ prop_dt: Union[DATATYPE, str, None] = prop.datatype
if prop_dt is not None and is_reference(prop_dt):
# add only if it is a reference, not a simple property
# Step 1: look for prop.value
@@ -3559,7 +3633,8 @@ class Container(list):
if is_list_datatype(prop_dt):
ref_name = get_list_datatype(prop_dt)
try:
- is_being_referenced.add(container.get_entity_by_name(ref_name).id)
+ is_being_referenced.add(
+ container.get_entity_by_name(ref_name).id) # type: ignore
except KeyError:
pass
elif isinstance(prop_dt, str):
@@ -3592,7 +3667,8 @@ class Container(list):
return dependencies
- def delete(self, raise_exception_on_error=True, flags=None, chunk_size=100):
+ def delete(self, raise_exception_on_error: bool = True,
+ flags: Optional[QueryDict] = None, chunk_size: int = 100):
"""Delete all entities in this container.
Entities are identified via their id if present and via their
@@ -3604,7 +3680,8 @@ class Container(list):
"""
item_count = len(self)
- # Split Container in 'chunk_size'-sized containers (if necessary) to avoid error 414 Request-URI Too Long
+ # Split Container in 'chunk_size'-sized containers (if necessary) to avoid error 414
+ # Request-URI Too Long
if item_count > chunk_size:
dependencies = Container._find_dependencies_in_container(self)
@@ -3615,7 +3692,8 @@ class Container(list):
if len(dependencies) == item_count:
if raise_exception_on_error:
te = TransactionError(
- msg="The container is too large and with too many dependencies within to be deleted.",
+ msg=("The container is too large and with too many dependencies within to"
+ " be deleted."),
container=self)
raise te
@@ -3706,7 +3784,7 @@ class Container(list):
unique: bool = True,
raise_exception_on_error: bool = True,
sync: bool = True,
- flags=None,
+ flags: Optional[QueryDict] = None,
):
"""Retrieve all entities in this container identified via their id if
present and via their name otherwise. Any locally already existing
@@ -3716,9 +3794,9 @@ class Container(list):
If any entity has no id and no name a LinkAheadException will be raised.
- Note: If only a name is given this could lead to ambiguities. All entities with the name in question
- will be returned. Therefore, the container could contain more elements after the retrieval than
- before.
+ Note: If only a name is given this could lead to ambiguities. All entities with the name in
+ question will be returned. Therefore, the container could contain more elements after the
+ retrieval than before.
"""
if isinstance(query, list):
@@ -3782,7 +3860,7 @@ class Container(list):
return (entities[0:hl], entities[hl:len(entities)])
- def _retrieve(self, entities, flags: Dict[str, Optional[str]]):
+ def _retrieve(self, entities, flags: Optional[QueryDict]):
c = get_connection()
try:
_log_request("GET: " + _ENTITY_URI_SEGMENT + str(entities) +
@@ -3815,7 +3893,8 @@ class Container(list):
return self
@staticmethod
- def _dir_to_http_parts(root, d, upload): # @ReservedAssignment
+ # @ReservedAssignment
+ def _dir_to_http_parts(root: str, d: Optional[str], upload: str):
ret = []
x = (root + '/' + d if d is not None else root)
@@ -3842,7 +3921,7 @@ class Container(list):
raise_exception_on_error: bool = True,
unique: bool = True,
sync: bool = True,
- flags: Optional[Dict[str, Any]] = None,
+ flags: Optional[dict[str, Any]] = None,
):
"""Update these entites."""
@@ -3854,7 +3933,7 @@ class Container(list):
self.clear_server_messages()
insert_xml = etree.Element("Update")
- http_parts: List[MultipartParam] = []
+ http_parts: list[MultipartParam] = []
if flags is None:
flags = {}
@@ -3925,7 +4004,7 @@ class Container(list):
@staticmethod
def _process_file_if_present_and_add_to_http_parts(
- http_parts: List[MultipartParam], entity: Union[File, Entity]
+ http_parts: list[MultipartParam], entity: Union[File, Entity]
):
if isinstance(entity, File) and hasattr(
entity, 'file') and entity.file is not None:
@@ -3975,7 +4054,7 @@ class Container(list):
raise_exception_on_error: bool = True,
unique: bool = True,
sync: bool = True,
- flags: Optional[Dict[str, Optional[str]]] = None,
+ flags: Optional[QueryDict] = None,
):
"""Insert this file entity into LinkAhead. A successful insertion will
generate a new persistent ID for this entity. This entity can be
@@ -4003,7 +4082,7 @@ class Container(list):
self.clear_server_messages()
insert_xml = etree.Element("Insert")
- http_parts: List[MultipartParam] = []
+ http_parts: list[MultipartParam] = []
if flags is None:
flags = {}
@@ -4057,7 +4136,8 @@ class Container(list):
if len(self) > 0 and len(insert_xml) < 1:
te = TransactionError(
- msg="There are no entities to be inserted. This container contains existent entities only.",
+ msg=("There are no entities to be inserted. This container contains existent"
+ " entities only."),
container=self)
raise te
_log_request("POST: " + _ENTITY_URI_SEGMENT +
@@ -4164,8 +4244,8 @@ class Container(list):
return self
def get_property_values(
- self, *selectors: Union[str, Tuple[str]]
- ) -> List[Tuple[str]]:
+ self, *selectors: Union[str, tuple[str]]
+ ) -> list[tuple[str]]:
""" Return a list of tuples with values of the given selectors.
I.e. a tabular representation of the container's content.
@@ -4221,7 +4301,8 @@ def sync_global_acl():
ACL.global_acl = ACL(xml=pelem)
else:
raise LinkAheadException(
- "The server's response didn't contain the expected elements. The configuration of this client might be invalid (especially the url).")
+ "The server's response didn't contain the expected elements. The configuration of this"
+ " client might be invalid (especially the url).")
def get_known_permissions():
@@ -4257,11 +4338,13 @@ class ACI():
return hash(self.__repr__())
def __eq__(self, other):
- return isinstance(other, ACI) and (self.role is None and self.username == other.username and self.realm ==
- other.realm) or self.role == other.role and self.permission == other.permission
+ return (isinstance(other, ACI) and
+ (self.role is None and self.username == other.username
+ and self.realm == other.realm)
+ or self.role == other.role and self.permission == other.permission)
def __repr__(self):
- return str(self.realm) + ":" + str(self.username) + ":" + str(self.role) + ":" + str(self.permission)
+ return ":".join([str(self.realm), str(self.username), str(self.role), str(self.permission)])
def add_to_element(self, e: etree._Element):
if self.role is not None:
@@ -4340,7 +4423,7 @@ class ACL():
permission=permission, priority=priority,
revoke_grant=False)
- def combine(self, other: ACL):
+ def combine(self, other: ACL) -> ACL:
""" Combine and return new instance."""
result = ACL()
result._grants.update(other._grants)
@@ -4355,7 +4438,11 @@ class ACL():
return result
def __eq__(self, other):
- return isinstance(other, ACL) and other._grants == self._grants and self._denials == other._denials and self._priority_grants == other._priority_grants and self._priority_denials == other._priority_denials
+ return (isinstance(other, ACL)
+ and other._grants == self._grants
+ and self._denials == other._denials
+ and self._priority_grants == other._priority_grants
+ and self._priority_denials == other._priority_denials)
def is_empty(self):
return len(self._grants) + len(self._priority_grants) + \
@@ -4682,7 +4769,7 @@ class Query():
return self.flags.get(key)
def __init__(self, q: Union[str, etree._Element]):
- self.flags: Dict[str, Optional[str]] = dict()
+ self.flags: QueryDict = dict()
self.messages = Messages()
self.cached: Optional[bool] = None
self.etag = None
@@ -4708,7 +4795,7 @@ class Query():
else:
self.q = q
- def _query_request(self, query_dict: Dict[str, Optional[str]]):
+ def _query_request(self, query_dict: QueryDict):
"""Used internally to execute the query request..."""
_log_request("GET Entity?" + str(query_dict), None)
connection = get_connection()
@@ -4721,7 +4808,7 @@ class Query():
def _paging_generator(
self,
first_page: Container,
- query_dict: Dict[str, Optional[str]],
+ query_dict: QueryDict,
page_length: int,
):
"""Used internally to create a generator of pages instead instead of a
@@ -4838,9 +4925,9 @@ def execute_query(
unique: bool = False,
raise_exception_on_error: bool = True,
cache: bool = True,
- flags: Optional[Dict[str, Optional[str]]] = None,
+ flags: Optional[QueryDict] = None,
page_length: Optional[int] = None,
-) -> Union[Container, int]:
+) -> Union[Container, Entity, int]:
"""Execute a query (via a server-requests) and return the results.
Parameters
@@ -4879,7 +4966,7 @@ def execute_query(
Returns
-------
- results : Container or integer
+ results : Container or Entity or integer
Returns an integer when it was a `COUNT` query. Otherwise, returns a
Container with the resulting entities.
"""
@@ -4895,8 +4982,7 @@ def execute_query(
class DropOffBox(list):
def __init__(self, *args, **kwargs):
- warn(DeprecationWarning(
- "The DropOffBox is deprecated and will be removed in future."))
+ warn(DeprecationWarning("The DropOffBox is deprecated and will be removed in future."))
super().__init__(*args, **kwargs)
path = None
@@ -5006,7 +5092,7 @@ class Permission():
class Permissions():
- known_permissions: Optional[List[Permissions]] = None
+ known_permissions: Optional[Permissions] = None
def __init__(self, xml: etree._Element):
self.parse_xml(xml)
@@ -5121,7 +5207,8 @@ def _parse_single_xml_element(elem: etree._Element):
)
-def _evaluate_and_add_error(parent_error: TransactionError, ent: Union[Entity, QueryTemplate, Container]):
+def _evaluate_and_add_error(parent_error: TransactionError,
+ ent: Union[Entity, QueryTemplate, Container]):
"""Evaluate the error message(s) attached to entity and add a
corresponding exception to parent_error.
@@ -5251,7 +5338,7 @@ def raise_errors(arg0: Union[Entity, QueryTemplate, Container]):
raise transaction_error
-def delete(ids: Union[List[int], range], raise_exception_on_error: bool = True):
+def delete(ids: Union[list[int], range], raise_exception_on_error: bool = True):
c = Container()
if isinstance(ids, list) or isinstance(ids, range):
diff --git a/src/linkahead/common/state.py b/src/linkahead/common/state.py
index 82f314e80191163f14a5c4babdd749f977f2901b..e352f82d9820620d1692cb6337eb218210e799e6 100644
--- a/src/linkahead/common/state.py
+++ b/src/linkahead/common/state.py
@@ -19,11 +19,19 @@
#
# ** end header
+from __future__ import annotations # Can be removed with 3.10.
import copy
from lxml import etree
+from typing import TYPE_CHECKING
+import sys
-def _translate_to_state_acis(acis):
+if TYPE_CHECKING:
+ from typing import Optional
+ from linkahead.common.models import ACL, ACI
+
+
+def _translate_to_state_acis(acis: set[ACI]) -> set[ACI]:
result = set()
for aci in acis:
aci = copy.copy(aci)
@@ -50,7 +58,13 @@ class Transition:
A state name
"""
- def __init__(self, name, from_state, to_state, description=None):
+ def __init__(
+ self,
+ name: Optional[str],
+ from_state: Optional[str],
+ to_state: Optional[str],
+ description: Optional[str] = None,
+ ):
self._name = name
self._from_state = from_state
self._to_state = to_state
@@ -76,25 +90,29 @@ class Transition:
return f'Transition(name="{self.name}", from_state="{self.from_state}", to_state="{self.to_state}", description="{self.description}")'
def __eq__(self, other):
- return (isinstance(other, Transition)
- and other.name == self.name
- and other.to_state == self.to_state
- and other.from_state == self.from_state)
+ return (
+ isinstance(other, Transition)
+ and other.name == self.name
+ and other.to_state == self.to_state
+ and other.from_state == self.from_state
+ )
def __hash__(self):
return 23472 + hash(self.name) + hash(self.from_state) + hash(self.to_state)
@staticmethod
- def from_xml(xml):
- to_state = [to.get("name") for to in xml
- if to.tag.lower() == "tostate"]
- from_state = [from_.get("name") for from_ in xml
- if from_.tag.lower() == "fromstate"]
- result = Transition(name=xml.get("name"),
- description=xml.get("description"),
- from_state=from_state[0] if from_state else None,
- to_state=to_state[0] if to_state else None)
- return result
+ def from_xml(xml: etree._Element) -> "Transition":
+ to_state = [to.get("name")
+ for to in xml if to.tag.lower() == "tostate"]
+ from_state = [
+ from_.get("name") for from_ in xml if from_.tag.lower() == "fromstate"
+ ]
+ return Transition(
+ name=xml.get("name"),
+ description=xml.get("description"),
+ from_state=from_state[0] if from_state else None,
+ to_state=to_state[0] if to_state else None,
+ )
class State:
@@ -119,12 +137,12 @@ class State:
All transitions which are available from this state (read-only)
"""
- def __init__(self, model, name):
+ def __init__(self, model: Optional[str], name: Optional[str]):
self.name = name
self.model = model
- self._id = None
- self._description = None
- self._transitions = None
+ self._id: Optional[str] = None
+ self._description: Optional[str] = None
+ self._transitions: Optional[set[Transition]] = None
@property
def id(self):
@@ -139,9 +157,11 @@ class State:
return self._transitions
def __eq__(self, other):
- return (isinstance(other, State)
- and self.name == other.name
- and self.model == other.model)
+ return (
+ isinstance(other, State)
+ and self.name == other.name
+ and self.model == other.model
+ )
def __hash__(self):
return hash(self.name) + hash(self.model)
@@ -164,7 +184,7 @@ class State:
return xml
@staticmethod
- def from_xml(xml):
+ def from_xml(xml: etree._Element):
"""Create a new State instance from an xml Element.
Parameters
@@ -175,24 +195,26 @@ class State:
-------
state : State
"""
- name = xml.get("name")
- model = xml.get("model")
- result = State(name=name, model=model)
+ result = State(name=xml.get("name"), model=xml.get("model"))
result._id = xml.get("id")
result._description = xml.get("description")
- transitions = [Transition.from_xml(t) for t in xml if t.tag.lower() ==
- "transition"]
+ transitions = [
+ Transition.from_xml(t) for t in xml if t.tag.lower() == "transition"
+ ]
if transitions:
result._transitions = set(transitions)
return result
@staticmethod
- def create_state_acl(acl):
+ def create_state_acl(acl: ACL):
from .models import ACL
+
state_acl = ACL()
state_acl._grants = _translate_to_state_acis(acl._grants)
state_acl._denials = _translate_to_state_acis(acl._denials)
- state_acl._priority_grants = _translate_to_state_acis(acl._priority_grants)
- state_acl._priority_denials = _translate_to_state_acis(acl._priority_denials)
+ state_acl._priority_grants = _translate_to_state_acis(
+ acl._priority_grants)
+ state_acl._priority_denials = _translate_to_state_acis(
+ acl._priority_denials)
return state_acl
diff --git a/src/linkahead/common/timezone.py b/src/linkahead/common/timezone.py
index 8fc5e710d3cbf6f20cf81397573f972db3b22f12..9ccb433c49b07e73a90bdad3ae6caaf2017a9c1d 100644
--- a/src/linkahead/common/timezone.py
+++ b/src/linkahead/common/timezone.py
@@ -1,3 +1,7 @@
+from __future__ import annotations
+from typing import Optional
+
+
class TimeZone():
"""
TimeZone, e.g. CEST, Europe/Berlin, UTC+4.
@@ -7,13 +11,13 @@ class TimeZone():
----------
zone_id : string
ID of the time zone.
- offset : int
- Offset to UTC in seconds.
+ offset : str
+ Offset to UTC, e.g. "+1400"
display_name : string
A human-friendly name of the time zone:
"""
- def __init__(self, zone_id, offset, display_name):
+ def __init__(self, zone_id: Optional[str], offset: Optional[str], display_name: str):
self.zone_id = zone_id
self.offset = offset
self.display_name = display_name
diff --git a/src/linkahead/common/versioning.py b/src/linkahead/common/versioning.py
index facfbc488e413e090ea1a856501ccd96334f8354..2e292e6bb031725fbd6da618c4b888c05072c46b 100644
--- a/src/linkahead/common/versioning.py
+++ b/src/linkahead/common/versioning.py
@@ -26,10 +26,14 @@
Currently this module defines nothing but a single class, `Version`.
"""
-from __future__ import absolute_import
+from __future__ import annotations
from .utils import xml2str
from lxml import etree
+from typing import TYPE_CHECKING
+if TYPE_CHECKING:
+ from typing import Optional, List, Union
+
class Version():
"""The version of an entity.
@@ -95,9 +99,11 @@ class Version():
"""
# pylint: disable=redefined-builtin
- def __init__(self, id=None, date=None, username=None, realm=None,
- predecessors=None, successors=None, is_head=False,
- is_complete_history=False):
+ def __init__(self, id: Optional[str] = None, date: Optional[str] = None,
+ username: Optional[str] = None, realm: Optional[str] = None,
+ predecessors: Optional[List[Version]] = None, successors: Optional[List[Version]] = None,
+ is_head: Union[bool, str, None] = False,
+ is_complete_history: Union[bool, str, None] = False):
"""Typically the `predecessors` or `successors` should not "link back" to an existing Version
object."""
self.id = id
@@ -109,7 +115,7 @@ object."""
self.is_head = str(is_head).lower() == "true"
self.is_complete_history = str(is_complete_history).lower() == "true"
- def get_history(self):
+ def get_history(self) -> List[Version]:
""" Returns a flat list of Version instances representing the history
of the entity.
@@ -126,7 +132,7 @@ object."""
-------
list of Version
"""
- versions = []
+ versions: List[Version] = []
for p in self.predecessors:
# assuming that predecessors don't have any successors
versions = p.get_history()
@@ -137,7 +143,7 @@ object."""
versions.extend(s.get_history())
return versions
- def to_xml(self, tag="Version"):
+ def to_xml(self, tag: str = "Version") -> etree._Element:
"""Serialize this version to xml.
The tag name is 'Version' per default. But since this method is called
@@ -184,7 +190,7 @@ object."""
return xml2str(self.to_xml())
@staticmethod
- def from_xml(xml):
+ def from_xml(xml: etree._Element) -> Version:
"""Parse a version object from a 'Version' xml element.
Parameters
diff --git a/src/linkahead/configuration.py b/src/linkahead/configuration.py
index 55a400459b1f3c9ea93d10ddf8c78765aa323eaa..b020467c8c53e26d464a6a2fb473cc912b0e0612 100644
--- a/src/linkahead/configuration.py
+++ b/src/linkahead/configuration.py
@@ -21,14 +21,14 @@
#
# ** end header
#
-
+from __future__ import annotations
import os
import warnings
import yaml
try:
- optional_jsonschema_validate = None
+ optional_jsonschema_validate: Optional[Callable] = None
from jsonschema import validate as optional_jsonschema_validate
except ImportError:
pass
@@ -37,13 +37,17 @@ from configparser import ConfigParser
from os import environ, getcwd
from os.path import expanduser, isfile, join
+from typing import Union, Callable, Optional
+
+_pycaosdbconf = ConfigParser(allow_no_value=False)
+
def _reset_config():
global _pycaosdbconf
_pycaosdbconf = ConfigParser(allow_no_value=False)
-def configure(inifile):
+def configure(inifile: str) -> list[str]:
"""read config from file.
Return a list of files which have successfully been parsed.
@@ -61,15 +65,15 @@ def configure(inifile):
return read_config
-def get_config():
+def get_config() -> ConfigParser:
global _pycaosdbconf
if ("_pycaosdbconf" not in globals() or _pycaosdbconf is None):
_reset_config()
return _pycaosdbconf
-def config_to_yaml(config):
- valobj = {}
+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():
@@ -84,7 +88,7 @@ def config_to_yaml(config):
return valobj
-def validate_yaml_schema(valobj):
+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)
@@ -95,7 +99,7 @@ def validate_yaml_schema(valobj):
""")
-def _read_config_files():
+def _read_config_files() -> list[str]:
"""Read config files from different paths.
Read the config from either ``$PYCAOSDBINI`` or home directory (``~/.pylinkahead.ini``), and
diff --git a/src/linkahead/connection/authentication/external_credentials_provider.py b/src/linkahead/connection/authentication/external_credentials_provider.py
index 3d1b8afa17f58a87f09afba90c4bc7ae6dcba693..8e22c5c5796603dcc6acfbe3eb9345b8fb2e2f4e 100644
--- a/src/linkahead/connection/authentication/external_credentials_provider.py
+++ b/src/linkahead/connection/authentication/external_credentials_provider.py
@@ -23,13 +23,10 @@
#
"""external_credentials_provider."""
from __future__ import absolute_import, unicode_literals
-from abc import ABCMeta
+from abc import ABC
import logging
from .plain import PlainTextCredentialsProvider
-# meta class compatible with Python 2 *and* 3:
-ABC = ABCMeta(str('ABC'), (object, ), {str('__slots__'): ()})
-
class ExternalCredentialsProvider(PlainTextCredentialsProvider, ABC):
"""ExternalCredentialsProvider.
diff --git a/src/linkahead/connection/authentication/input.py b/src/linkahead/connection/authentication/input.py
index 2799207354b3949063461229d7d465e8a83c83ae..14e8196e72a9e6266631511cf36d5e8c0c5c68c4 100644
--- a/src/linkahead/connection/authentication/input.py
+++ b/src/linkahead/connection/authentication/input.py
@@ -25,13 +25,13 @@
A CredentialsProvider which reads the password from the input line.
"""
-from __future__ import absolute_import, unicode_literals, print_function
+from __future__ import annotations
from .interface import CredentialsProvider, CredentialsAuthenticator
-
+from typing import Optional
import getpass
-def get_authentication_provider():
+def get_authentication_provider() -> CredentialsAuthenticator:
"""get_authentication_provider.
Return an authenticator which uses the input for username/password credentials.
@@ -61,8 +61,8 @@ class InputCredentialsProvider(CredentialsProvider):
def __init__(self):
super(InputCredentialsProvider, self).__init__()
- self._password = None
- self._username = None
+ self._password: Optional[str] = None
+ self._username: Optional[str] = None
def configure(self, **config):
"""configure.
diff --git a/src/linkahead/connection/authentication/interface.py b/src/linkahead/connection/authentication/interface.py
index 6de43b81f441ab60401c1c01885eaa514790d3de..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,23 +19,25 @@
#
# 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 abc import ABCMeta, abstractmethod, abstractproperty
+
+from __future__ import annotations
+from abc import ABC, abstractmethod
import logging
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, Optional
+if TYPE_CHECKING:
+ from ..interface import CaosDBHTTPResponse
+ QueryDict = dict[str, Optional[str]]
-# meta class compatible with Python 2 *and* 3:
-ABC = ABCMeta('ABC', (object, ), {'__slots__': ()})
_LOGGER = logging.getLogger(__name__)
@@ -108,7 +111,7 @@ class AbstractAuthenticator(ABC):
"""
pass
- def on_response(self, response):
+ def on_response(self, response: CaosDBHTTPResponse):
"""on_response.
A call-back with is to be called by the connection after each
@@ -125,7 +128,7 @@ class AbstractAuthenticator(ABC):
self.auth_token = parse_auth_token(
response.getheader("Set-Cookie"))
- def on_request(self, method, path, headers, **kwargs):
+ def on_request(self, method: str, path: str, headers: QueryDict, **kwargs):
# pylint: disable=unused-argument
"""on_request.
@@ -262,10 +265,12 @@ class CredentialsProvider(ABC):
None
"""
- @abstractproperty
+ @property
+ @abstractmethod
def password(self):
"""password."""
- @abstractproperty
+ @property
+ @abstractmethod
def username(self):
"""username."""
diff --git a/src/linkahead/connection/authentication/keyring.py b/src/linkahead/connection/authentication/keyring.py
index 202520bbab7e940ccce6517e640eff5904039553..dad6ed9b4ad77175db6b75e30b70152878da487d 100644
--- a/src/linkahead/connection/authentication/keyring.py
+++ b/src/linkahead/connection/authentication/keyring.py
@@ -35,7 +35,7 @@ from .external_credentials_provider import ExternalCredentialsProvider
from .interface import CredentialsAuthenticator
-def get_authentication_provider():
+def get_authentication_provider() -> CredentialsAuthenticator:
"""get_authentication_provider.
Return an authenticator which uses plain text username/password credentials.
diff --git a/src/linkahead/connection/authentication/pass.py b/src/linkahead/connection/authentication/pass.py
index bec307401f945a6cd2e223195e0cce2396602061..81a901523fcad65680d34947cd9cc741e06a0352 100644
--- a/src/linkahead/connection/authentication/pass.py
+++ b/src/linkahead/connection/authentication/pass.py
@@ -33,7 +33,7 @@ from .interface import CredentialsAuthenticator
from .external_credentials_provider import ExternalCredentialsProvider
-def get_authentication_provider():
+def get_authentication_provider() -> CredentialsAuthenticator:
"""get_authentication_provider.
Return an authenticator which uses plain text username/password credentials.
diff --git a/src/linkahead/connection/connection.py b/src/linkahead/connection/connection.py
index 91b4a01da455d0f365e39b0b0f7359e07096e707..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
+from __future__ import annotations
import logging
import ssl
@@ -32,8 +32,7 @@ import warnings
from builtins import str # pylint: disable=redefined-builtin
from errno import EPIPE as BrokenPipe
from socket import error as SocketError
-from urllib.parse import quote, urlparse
-from warnings import warn
+from urllib.parse import ParseResult, quote, urlparse
from requests import Session as HTTPSession
from requests.adapters import HTTPAdapter
@@ -52,22 +51,28 @@ 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:
+ from typing import Optional, Any, Iterator, Union
+ from requests.models import Response
+ from ssl import _SSLMethod
+ from .authentication.interface import AbstractAuthenticator, CredentialsAuthenticator
+
_LOGGER = logging.getLogger(__name__)
class _WrappedHTTPResponse(CaosDBHTTPResponse):
- def __init__(self, response):
- self.response = response
- self._generator = None
- self._buffer = b''
- self._stream_consumed = False
+ def __init__(self, response: Response):
+ self.response: Response = response
+ self._generator: Optional[Iterator[Any]] = None
+ self._buffer: Optional[bytes] = b''
+ self._stream_consumed: bool = False
@property
def reason(self):
@@ -77,7 +82,7 @@ class _WrappedHTTPResponse(CaosDBHTTPResponse):
def status(self):
return self.response.status_code
- def read(self, size=None):
+ def read(self, size: Optional[int] = None):
if self._stream_consumed is True:
raise RuntimeError("Stream is consumed")
@@ -91,8 +96,13 @@ class _WrappedHTTPResponse(CaosDBHTTPResponse):
self._stream_consumed = True
return self.response.content
+ if size is None or size == 0:
+ raise RuntimeError(
+ "size parameter should not be None if the stream is not consumed yet")
+
if len(self._buffer) >= size:
# still enough bytes in the buffer
+ # FIXME: `chunk`` is used before definition
result = chunk[:size]
self._buffer = chunk[size:]
return result
@@ -117,7 +127,7 @@ class _WrappedHTTPResponse(CaosDBHTTPResponse):
self._buffer = None
return result
- def getheader(self, name, default=None):
+ def getheader(self, name: str, default=None):
return self.response.headers[name] if name in self.response.headers else default
def getheaders(self):
@@ -130,7 +140,7 @@ class _WrappedHTTPResponse(CaosDBHTTPResponse):
class _SSLAdapter(HTTPAdapter):
"""Transport adapter that allows us to use different SSL versions."""
- def __init__(self, ssl_version):
+ def __init__(self, ssl_version: _SSLMethod):
self.ssl_version = ssl_version
super().__init__()
@@ -156,7 +166,11 @@ class _DefaultCaosDBServerConnection(CaosDBServerConnection):
self._session = None
self._timeout = None
- def request(self, method, path, headers=None, body=None):
+ def request(self,
+ method: str, path: str,
+ headers: Optional[dict[str, str]] = None,
+ body: Union[str, bytes, None] = None,
+ **kwargs) -> _WrappedHTTPResponse:
"""request.
Send a HTTP request to the server.
@@ -169,14 +183,14 @@ class _DefaultCaosDBServerConnection(CaosDBServerConnection):
An URI path segment (without the 'scheme://host:port/' parts),
including query and frament segments.
headers : dict of str -> str, optional
- HTTP request headers. (Defautl: None)
+ HTTP request headers. (Default: None)
body : str or bytes or readable, optional
The body of the HTTP request. Bytes should be a utf-8 encoded
string.
Returns
-------
- response : CaosDBHTTPResponse
+ response : _WrappedHTTPResponse
"""
if headers is None:
@@ -232,14 +246,16 @@ class _DefaultCaosDBServerConnection(CaosDBServerConnection):
"No connection url specified. Please "
"do so via linkahead.configure_connection(...) or in a config "
"file.")
- if (not config["url"].lower().startswith("https://") and not config["url"].lower().startswith("http://")):
+ url_string: str = config["url"]
+ 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 "
"(i.e. start with https:// or "
"http://).")
- url = urlparse(config["url"])
+ url: ParseResult = urlparse(url=url_string)
path = url.path.strip("/")
if len(path) > 0:
path = path + "/"
@@ -271,7 +287,7 @@ class _DefaultCaosDBServerConnection(CaosDBServerConnection):
if "timeout" in config:
self._timeout = config["timeout"]
- def _setup_ssl(self, config):
+ 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:
@@ -325,7 +341,7 @@ _DEFAULT_CONF = {
}
-def _get_authenticator(**config):
+def _get_authenticator(**config) -> AbstractAuthenticator:
"""_get_authenticator.
Import and configure the password_method.
@@ -337,7 +353,7 @@ def _get_authenticator(**config):
Currently, there are four valid values for this parameter: 'plain',
'pass', 'keyring' and 'auth_token'.
**config :
- Any other keyword arguments are passed the configre method of the
+ Any other keyword arguments are passed the configure method of the
password_method.
Returns
@@ -413,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`,
@@ -534,8 +552,8 @@ class _Connection(object): # pylint: disable=useless-object-inheritance
__instance = None
def __init__(self):
- self._delegate_connection = None
- self._authenticator = None
+ self._delegate_connection: Optional[CaosDBServerConnection] = None
+ self._authenticator: Optional[CredentialsAuthenticator] = None
self.is_configured = False
@classmethod
@@ -553,7 +571,8 @@ class _Connection(object): # pylint: disable=useless-object-inheritance
"Missing CaosDBServerConnection implementation. You did not "
"specify an `implementation` for the connection.")
try:
- self._delegate_connection = config["implementation"]()
+ self._delegate_connection: CaosDBServerConnection = config["implementation"](
+ )
if not isinstance(self._delegate_connection,
CaosDBServerConnection):
@@ -579,14 +598,19 @@ class _Connection(object): # pylint: disable=useless-object-inheritance
return self
- def retrieve(self, entity_uri_segments=None, query_dict=None, **kwargs):
+ def retrieve(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(method="GET", path=path, **kwargs)
return http_response
- def delete(self, entity_uri_segments=None, query_dict=None, **kwargs):
+ 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(
@@ -594,15 +618,18 @@ class _Connection(object): # pylint: disable=useless-object-inheritance
return http_response
- def update(self, entity_uri_segment, query_dict=None, **kwargs):
+ 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)
return http_response
- def activate_user(self, link):
- self._authenticator.logout()
+ def activate_user(self, link: str) -> CaosDBHTTPResponse:
+ if self._authenticator is not None:
+ self._authenticator.logout()
fullurl = urlparse(link)
path = fullurl.path
query = fullurl.query
@@ -611,17 +638,19 @@ class _Connection(object): # pylint: disable=useless-object-inheritance
return http_response
- def put_form_data(self, entity_uri_segment, params):
+ def put_form_data(self, entity_uri_segment: str, params) -> CaosDBHTTPResponse:
return self._form_data_request(
method="PUT", path=entity_uri_segment, params=params)
- def post_form_data(self, entity_uri_segment, params):
+ 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, path, params):
+ 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"
@@ -633,7 +662,9 @@ class _Connection(object): # pylint: disable=useless-object-inheritance
return response
- def insert(self, entity_uri_segment, query_dict=None, body=None, **kwargs):
+ 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)
http_response = self._http_request(
@@ -641,7 +672,7 @@ class _Connection(object): # pylint: disable=useless-object-inheritance
return http_response
- def download_file(self, path):
+ def download_file(self, path: str):
"""This function downloads a file via HTTP from the LinkAhead file
system."""
try:
@@ -658,7 +689,9 @@ class _Connection(object): # pylint: disable=useless-object-inheritance
def _logout(self):
self._authenticator.logout()
- def _http_request(self, method, path, headers=None, body=None, **kwargs):
+ def _http_request(self, method: str, path: str,
+ headers: Optional[dict["str", Any]] = None,
+ body: Union[str, bytes, None] = None, **kwargs):
try:
return self._retry_http_request(method=method, path=path,
headers=headers, body=body,
@@ -681,16 +714,30 @@ class _Connection(object): # pylint: disable=useless-object-inheritance
**kwargs)
raise
- def _retry_http_request(self, method, path, headers, body, **kwargs):
+ def _retry_http_request(self,
+ method: str,
+ path: str,
+ headers: Optional[dict["str", Any]],
+ body: Union[str, bytes, None], **kwargs) -> CaosDBHTTPResponse:
- if hasattr(body, "encode"):
+ if hasattr(body, "encode") and body is not None:
# python3
body = body.encode("utf-8")
if headers is None:
headers = {}
+
+ if self._authenticator is None:
+ raise ValueError(
+ "No authenticator set. Please call configure_connection() first.")
+
self._authenticator.on_request(method=method, path=path,
headers=headers)
+
+ if self._delegate_connection is None:
+ raise ValueError(
+ "No connection set. Please call configure_connection() first.")
+
_LOGGER.debug("request: %s %s %s", method, path, str(headers))
http_response = self._delegate_connection.request(
method=method,
@@ -704,10 +751,16 @@ class _Connection(object): # pylint: disable=useless-object-inheritance
return http_response
- def get_username(self):
+ def get_username(self) -> str:
"""
Return the username of the current connection.
Shortcut for: get_connection()._authenticator._credentials_provider.username
"""
+ if self._authenticator is None:
+ raise ValueError(
+ "No authenticator set. Please call configure_connection() first.")
+ if self._authenticator._credentials_provider is None:
+ raise ValueError(
+ "No credentials provider set. Please call configure_connection() first.")
return self._authenticator._credentials_provider.username
diff --git a/src/linkahead/connection/encode.py b/src/linkahead/connection/encode.py
index 6b328285e97e4dce2483ddd955134ee64cd3ce84..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
@@ -48,6 +50,7 @@ as multipart/form-data suitable for a HTTP POST or PUT request.
multipart/form-data is the standard way to upload files over HTTP
"""
+from __future__ import annotations
__all__ = [
'gen_boundary', 'encode_and_quote', 'MultipartParam', 'encode_string',
@@ -61,6 +64,10 @@ import re
import os
import mimetypes
from email.header import Header
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+ from typing import Optional
def gen_boundary():
@@ -68,7 +75,7 @@ def gen_boundary():
return uuid.uuid4().hex
-def encode_and_quote(data):
+def encode_and_quote(data: Optional[str]) -> Optional[str]:
"""If ``data`` is unicode, return urllib.quote_plus(data.encode("utf-8"))
otherwise return urllib.quote_plus(data)"""
if data is None:
@@ -111,7 +118,7 @@ class MultipartParam(object):
"""
def __init__(self,
- name,
+ name: str,
value=None,
filename=None,
filetype=None,
@@ -147,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 d63dbeb8cc4cd59e056823440948aa54906dd47c..fc22577dffb4f2e0d30924324cd7a4901d2c8b1a 100644
--- a/src/linkahead/connection/interface.py
+++ b/src/linkahead/connection/interface.py
@@ -22,11 +22,14 @@
# ** end header
#
"""This module defines the CaosDBServerConnection interface."""
-from abc import ABCMeta, abstractmethod, abstractproperty
+from __future__ import annotations
+from abc import ABCMeta, abstractmethod, ABC
from warnings import warn
-# meta class compatible with Python 2 *and* 3:
-ABC = ABCMeta('ABC', (object, ), {'__slots__': ()})
+
+from typing import TYPE_CHECKING
+if TYPE_CHECKING:
+ from typing import Optional, Union
class CaosDBHTTPResponse(ABC):
@@ -34,14 +37,14 @@ class CaosDBHTTPResponse(ABC):
LinkAheadServer."""
@abstractmethod
- def read(self, size=-1):
+ def read(self, size: Optional[int] = -1):
"""Read up to *size* bytes from the response body.
If size is unspecified or -1, all bytes until EOF are returned.
"""
@abstractmethod
- def getheader(self, name, default=None):
+ def getheader(self, name: str, default=None):
"""Return the value of the header *name* or the value of *default* if
there is no such header.
@@ -50,12 +53,13 @@ class CaosDBHTTPResponse(ABC):
are returned likewise.
"""
- @abstractproperty
- def status(self):
+ @property
+ @abstractmethod
+ def status(self) -> int:
"""Status code of the response."""
@abstractmethod
- def getheaders(self):
+ def getheaders(self) -> dict[str, str]:
"""Return all headers."""
def __enter__(self):
@@ -78,7 +82,12 @@ class CaosDBServerConnection(ABC):
LinkAhead server."""
@abstractmethod
- def request(self, method, path, headers=None, body=None, **kwargs):
+ def request(self,
+ method: str,
+ path: str,
+ headers: Optional[dict[str, str]] = None,
+ body: Union[str, bytes, None] = None,
+ **kwargs) -> CaosDBHTTPResponse:
"""Abstract method. Implement this method for HTTP requests to the
LinkAhead server.
diff --git a/src/linkahead/connection/utils.py b/src/linkahead/connection/utils.py
index 90ec6b5ba6789747f5d4452a1260306b716b1f7e..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,14 +24,18 @@
# ** end header
#
"""Utility functions for the connection module."""
-from __future__ import unicode_literals, print_function
-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
+if TYPE_CHECKING:
+ from typing import Optional
-def urlencode(query):
+
+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
@@ -79,7 +85,8 @@ modules when they are called with only the query parameter.
}))
-def make_uri_path(segments=None, query=None):
+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
@@ -105,22 +112,25 @@ def make_uri_path(segments=None, query=None):
"""
path_no_query = ("/".join([quote(segment) for segment in segments])
if segments else "")
- return str(path_no_query if query is None else "?".join([
+ if query is None:
+ return str(path_no_query)
+
+ return str("?".join([
path_no_query, "&".join([
quote(key) + "=" +
- (quote(query[key]) if query[key] is not None else "")
+ (quote(query[key]) if query[key] is not None else "") # type: ignore
for key in query
])
]))
-def quote(string):
+def quote(string: str) -> str:
enc = string.encode('utf-8')
return _quote(enc).replace('/', '%2F')
-def parse_url(url):
- fullurl = urlparse(url)
+def parse_url(url: str):
+ fullurl = urlparse(url=url)
# make sure the path ends with a slash
if not fullurl.path.endswith("/"):
parse_result = list(fullurl)
@@ -132,19 +142,7 @@ def parse_url(url):
_PATTERN = re.compile(r"^SessionToken=([^;]*);.*$")
-def unquote(string):
- """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):
+def parse_auth_token(cookie: Optional[str]) -> Optional[str]:
"""parse_auth_token.
Parse an auth token from a cookie.
@@ -165,7 +163,7 @@ def parse_auth_token(cookie):
return auth_token
-def auth_token_to_cookie(auth_token):
+def auth_token_to_cookie(auth_token: str) -> str:
"""auth_token_to_cookie.
Urlencode an auth token string and format it as a cookie.
diff --git a/src/linkahead/high_level_api.py b/src/linkahead/high_level_api.py
index 70f1be36283b706f8d38d450d937ab13a9b9e699..18d219c732672d16d0ab43e562cfe73d682614fe 100644
--- a/src/linkahead/high_level_api.py
+++ b/src/linkahead/high_level_api.py
@@ -23,7 +23,7 @@
#
# ** end header
#
-
+# type: ignore
"""
A high level API for accessing LinkAhead entities from within python.
diff --git a/src/linkahead/utils/get_entity.py b/src/linkahead/utils/get_entity.py
index 282f7c86e10571d0e0d62b93da7f61bba5205cba..ce74089784525b1a38d9c5c5bbf3193a9ba94575 100644
--- a/src/linkahead/utils/get_entity.py
+++ b/src/linkahead/utils/get_entity.py
@@ -33,7 +33,8 @@ def get_entity_by_name(name: str) -> Entity:
Submits the query "FIND ENTITY WITH name='{name}'".
"""
name = escape_squoted_text(name)
- return execute_query(f"FIND ENTITY WITH name='{name}'", unique=True)
+ # type hint can be ignored, it's a unique query, so never Container or int
+ return execute_query(f"FIND ENTITY WITH name='{name}'", unique=True) # type: ignore
def get_entity_by_id(eid: Union[str, int]) -> Entity:
@@ -41,7 +42,8 @@ def get_entity_by_id(eid: Union[str, int]) -> Entity:
Submits the query "FIND ENTITY WITH id='{eid}'".
"""
- return execute_query(f"FIND ENTITY WITH id='{eid}'", unique=True)
+ # type hint can be ignored, it's a unique query
+ return execute_query(f"FIND ENTITY WITH id='{eid}'", unique=True) # type: ignore
def get_entity_by_path(path: str) -> Entity:
@@ -49,4 +51,5 @@ def get_entity_by_path(path: str) -> Entity:
Submits the query "FIND FILE WHICH IS STORED AT '{path}'".
"""
- return execute_query(f"FIND FILE WHICH IS STORED AT '{path}'", unique=True)
+ # type hint can be ignored, it's a unique query
+ return execute_query(f"FIND FILE WHICH IS STORED AT '{path}'", unique=True) # type: ignore
diff --git a/unittests/test_high_level_api.py b/unittests/test_high_level_api.py
index be57e3c747619852c7cec2002eac6928c7d77702..82c1a5caf0f0719b5946ecd6749b4079bb6794bc 100644
--- a/unittests/test_high_level_api.py
+++ b/unittests/test_high_level_api.py
@@ -21,7 +21,7 @@
#
# Test high level api module
# A. Schlemmer, 02/2022
-
+# type: ignore
import linkahead as db
from linkahead.high_level_api import (convert_to_entity, convert_to_python_object,