diff --git a/src/linkahead/common/datatype.py b/src/linkahead/common/datatype.py
index c0c15feca240112f1f8e33a0cd37932151fcd9f0..4d09aa3031565cccab736b6ad37de10d0a7eed8a 100644
--- a/src/linkahead/common/datatype.py
+++ b/src/linkahead/common/datatype.py
@@ -24,6 +24,7 @@
#
import re
+from typing import Literal
from ..exceptions import EmptyUniqueQueryError, QueryNotUniqueError
@@ -34,6 +35,7 @@ DATETIME = "DATETIME"
INTEGER = "INTEGER"
FILE = "FILE"
BOOLEAN = "BOOLEAN"
+DATATYPE = Literal["DOUBLE", "REFERENCE", "TEXT", "DATETIME", "INTEGER", "FILE", "BOOLEAN"]
def LIST(datatype):
diff --git a/src/linkahead/common/models.py b/src/linkahead/common/models.py
index bcedfb7104d733bc1ac5248e8668a16549891fdc..157354b12608f8969d68739d8fefaba8e9921810 100644
--- a/src/linkahead/common/models.py
+++ b/src/linkahead/common/models.py
@@ -48,11 +48,11 @@ from os.path import isdir
from random import randint
from tempfile import NamedTemporaryFile
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Dict
if TYPE_CHECKING:
from datetime import datetime
- from typing import Any, Literal, Optional, Type, Union, List, TextIO
+ from typing import Any, Literal, Optional, Type, Union, List, TextIO, Tuple
from warnings import warn
@@ -71,8 +71,17 @@ from ..exceptions import (AmbiguousEntityError, AuthorizationError,
QueryNotUniqueError, TransactionError,
UniqueNamesError, UnqualifiedParentsError,
UnqualifiedPropertiesError)
-from .datatype import (BOOLEAN, DATETIME, DOUBLE, INTEGER, TEXT,
- get_list_datatype, is_list_datatype, is_reference)
+from .datatype import (
+ BOOLEAN,
+ DATETIME,
+ DOUBLE,
+ INTEGER,
+ TEXT,
+ DATATYPE,
+ get_list_datatype,
+ is_list_datatype,
+ is_reference,
+)
from .state import State
from .timezone import TimeZone
from .utils import uuid, xml2str
@@ -80,24 +89,14 @@ from .versioning import Version
_ENTITY_URI_SEGMENT = "Entity"
-
-# importances/inheritance
-class Inheritance(Enum):
- OBLIGATORY = "OBLIGATORY"
- SUGGESTED = "SUGGESTED"
- RECOMMENDED = "RECOMMENDED"
- FIX = "FIX"
- ALL = "ALL"
- NONE = "NONE"
-
-
OBLIGATORY = "OBLIGATORY"
SUGGESTED = "SUGGESTED"
RECOMMENDED = "RECOMMENDED"
FIX = "FIX"
ALL = "ALL"
NONE = "NONE"
-
+if TYPE_CHECKING:
+ INHERITANCE = Literal["OBLIGATORY", "SUGGESTED", "RECOMMENDED", "FIX", "ALL", "NONE"]
SPECIAL_ATTRIBUTES = ["name", "role", "datatype", "description",
"id", "path", "checksum", "size", "value"]
@@ -121,7 +120,7 @@ class Entity:
name: Optional[str] = None,
id: Optional[int] = None,
description: Optional[str] = None, # @ReservedAssignment
- datatype: Optional[str] = None,
+ datatype: Optional[DATATYPE] = None,
value=None,
**kwargs,
):
@@ -133,32 +132,33 @@ class Entity:
# If an entity is used (e.g. as parent), it is wrapped instead of being used directly.
# see Entity._wrap()
self._wrapped_entity: Optional[Entity] = None
- self._version = None
- self._cuid = None
- self._flags = dict()
+ self._version: Optional[Version] = None
+ self._cuid: Optional[str] = None
+ self._flags: Dict[str, str] = dict()
self.__value = None
- self.__datatype = None
- self.datatype = datatype
+ self.__datatype: Optional[DATATYPE] = None
+ self.datatype: Optional[DATATYPE] = datatype
self.value = value
self.messages = Messages()
self.properties = _Properties()
self.parents = _ParentList()
- self.path = None
- self.file = None
- self.unit = None
+ self.path: Optional[str] = None
+ self.file: Optional[File] = None
+ self.unit: Optional[str] = None
self.acl: Optional[ACL] = None
- self.permissions = None
+ self.permissions: Optional[Permissions] = None
self.is_valid = lambda: False
self.is_deleted = lambda: False
self.name = name
self.description = description
- self.id = id
- self.state = None
+ self.id: Optional[int] = id
+ self.state: Optional[State] = None
def copy(self):
"""
Return a copy of entity.
+ FIXME: This method doesn't have a deep keyword argument.
If deep == True return a deep copy, recursively copying all sub entities.
Standard properties are copied using add_property.
@@ -204,7 +204,7 @@ class Entity:
return self._wrapped_entity.version
@version.setter
- def version(self, version):
+ def version(self, version: Optional[Version]):
self._version = version
@property
@@ -276,14 +276,14 @@ class Entity:
return self._wrapped_entity.description
- @property
- def checksum(self):
- return self._checksum
-
@description.setter
def description(self, new_description):
self.__description = new_description
+ @property
+ def checksum(self):
+ return self._checksum
+
@property
def unit(self):
if self.__unit is not None or self._wrapped_entity is None:
@@ -355,7 +355,7 @@ class Entity:
realm: Optional[str] = None,
username: Optional[str] = None,
role: Optional[str] = None,
- permission: str = None,
+ permission: Optional[str] = None,
priority: bool = False,
revoke_denial: bool = True,
):
@@ -397,7 +397,7 @@ class Entity:
realm: Optional[str] = None,
username: Optional[str] = None,
role: Optional[str] = None,
- permission: str = None,
+ permission: Optional[str] = None,
priority: bool = False,
revoke_grant: bool = True,
):
@@ -452,7 +452,7 @@ class Entity:
permission=permission,
priority=priority)
- def is_permitted(self, permission: str, role: Optional[str] = None):
+ def is_permitted(self, permission: Permission, role: Optional[str] = None):
if role is None:
# pylint: disable=unsupported-membership-test
@@ -580,7 +580,7 @@ class Entity:
datatype: Optional[str] = None,
unit: Optional[str] = None,
importance: Optional[str] = None,
- inheritance: Union[str, Inheritance, None] = None,
+ inheritance: Union[str, INHERITANCE, None] = None,
) -> Entity: # @ReservedAssignment
"""Add a property to this entity.
@@ -756,7 +756,7 @@ class Entity:
parent: Union[Entity, int, str, None] = None,
id: Optional[int] = None,
name: Optional[str] = None,
- inheritance: Union[str, Inheritance, None] = None,
+ inheritance: Union[INHERITANCE, str, None] = None,
): # @ReservedAssignment
"""Add a parent to this entity.
@@ -771,7 +771,7 @@ class Entity:
name : str
Name of the parent entity. Ignored if `parent is not
none`.
- inheritance : str, Inheritance
+ inheritance : str, INHERITANCE
One of ``obligatory``, ``recommended``, ``suggested``, or ``fix``. Specifies the
minimum importance which parent properties need to have to be inherited by this
entity. If no `inheritance` is given, no properties will be inherited by the child.
@@ -978,7 +978,7 @@ out: List[Entity]
return self.properties
- def get_property(self, pattern):
+ def get_property(self, pattern: Union[int, str, Entity]) -> Union[Property, None]:
""" Return the first matching property or None.
Parameters
@@ -1023,7 +1023,9 @@ out: List[Entity]
return None
- def _get_value_for_selector(self, selector):
+ def _get_value_for_selector(
+ self, selector: Union[str, List[str], Tuple[str]]
+ ) -> Any:
"""return the value described by the selector
A selector is a list or a tuple of strings describing a path in an
@@ -1195,8 +1197,8 @@ out: List[Entity]
self,
xml: Optional[etree._Element] = None,
add_properties=ALL,
- local_serialization=False,
- ):
+ local_serialization: bool=False,
+ )->etree._Element:
"""Generate an xml representation of this entity. If the parameter xml
is given, all attributes, parents, properties, and messages of this
entity will be added to it instead of creating a new element.
@@ -1216,9 +1218,8 @@ out: List[Entity]
assert isinstance(xml, etree._Element)
# unwrap wrapped entity
-
if self._wrapped_entity is not None:
- xml: etree._Element = self._wrapped_entity.to_xml(xml, add_properties)
+ xml = self._wrapped_entity.to_xml(xml, add_properties)
if self.id is not None:
xml.set("id", str(self.id))
@@ -1960,8 +1961,8 @@ class Message(object):
self,
type: Optional[str] = None,
code: Optional[int] = None,
- description=None,
- body: str = None,
+ description: Optional[str]=None,
+ body: Optional[str] = None,
): # @ReservedAssignment
self.description = description
self.type = type if type is not None else "Info"
@@ -2010,7 +2011,13 @@ class RecordType(Entity):
property=property, id=id, name=name, description=description, datatype=datatype,
value=value, unit=unit, importance=importance, inheritance=inheritance)
- def add_parent(self, parent=None, id=None, name=None, inheritance:Union[str, Inheritance]=OBLIGATORY):
+ def add_parent(
+ self,
+ parent=None,
+ id=None,
+ name=None,
+ inheritance: Union[INHERITANCE, str, None] = OBLIGATORY,
+ ):
"""Add a parent to this RecordType
Parameters
@@ -2301,7 +2308,7 @@ class _Properties(list):
return self
def to_xml(
- self, add_to_element: etree._Element, add_properties: Union[str, Inheritance]
+ self, add_to_element: etree._Element, add_properties: Union[str, INHERITANCE]
):
for p in self:
importance = self._importance.get(p)
@@ -4191,10 +4198,10 @@ class ACL():
self._priority_grants: set[ACI] = set()
self._priority_denials: set[ACI] = set()
- def _get_boolean_priority(self, priority: str):
+ def _get_boolean_priority(self, priority: Any):
return str(priority).lower() in ["true", "1", "yes", "y"]
- def _remove_item(self, item, priority: str):
+ def _remove_item(self, item, priority: bool):
try:
self._denials.remove(item)
except KeyError:
@@ -4219,7 +4226,7 @@ class ACL():
username: Optional[str] = None,
realm: Optional[str] = None,
role: Optional[str] = None,
- permission: str = None,
+ permission: Optional[str] = None,
priority: Union[bool, str] = False,
):
priority = self._get_boolean_priority(priority)
@@ -4300,7 +4307,7 @@ class ACL():
username: Optional[str] = None,
realm: Optional[str] = None,
role: Optional[str] = None,
- permission: str = None,
+ permission: Optional[str] = None,
priority: bool = False,
revoke_grant: bool = True,
):
@@ -4499,10 +4506,10 @@ class Query():
def getFlag(self, key):
return self.flags.get(key)
- def __init__(self, q):
- self.flags = dict()
+ def __init__(self, q: Union[str, etree._Element]):
+ self.flags: Dict[str, str] = dict()
self.messages = Messages()
- self.cached = None
+ self.cached: Optional[bool] = None
self.etag = None
if isinstance(q, etree._Element):
@@ -4547,8 +4554,13 @@ class Query():
yield next_page
index += page_length
- def execute(self, unique=False, raise_exception_on_error=True, cache=True,
- page_length=None):
+ def execute(
+ self,
+ unique: bool = False,
+ raise_exception_on_error: bool = True,
+ cache: bool = True,
+ page_length: Optional[int] = None,
+ ):
"""Execute a query (via a server-requests) and return the results.
Parameters
@@ -4635,8 +4647,14 @@ class Query():
return cresp
-def execute_query(q, unique=False, raise_exception_on_error=True, cache=True,
- flags=None, page_length=None):
+def execute_query(
+ q: str,
+ unique: bool = False,
+ raise_exception_on_error: bool = True,
+ cache: bool = True,
+ flags: Optional[Dict[str, str]] = None,
+ page_length: Optional[int] = None,
+):
"""Execute a query (via a server-requests) and return the results.
Parameters
@@ -4728,7 +4746,7 @@ class DropOffBox(list):
class UserInfo():
- def __init__(self, xml):
+ def __init__(self, xml: etree._Element):
self.roles = [role.text for role in xml.findall("Roles/Role")]
self.name = xml.get("username")
self.realm = xml.get("realm")
@@ -4778,7 +4796,7 @@ class Info():
class Permission():
- def __init__(self, name, description=None):
+ def __init__(self, name: str, description: Optional[str] = None):
self.name = name
self.description = description
@@ -4802,13 +4820,13 @@ class Permissions():
known_permissions = None
- def __init__(self, xml):
+ def __init__(self, xml: etree._Element):
self.parse_xml(xml)
def clear(self):
self._perms = set()
- def parse_xml(self, xml):
+ def parse_xml(self, xml: etree._Element):
self.clear()
for e in xml:
@@ -4840,12 +4858,12 @@ def parse_xml(xml: Union[str, etree._Element]):
if isinstance(xml, etree._Element):
elem: etree._Element = xml
else:
- elem: etree._Element = etree.fromstring(xml)
+ elem = etree.fromstring(xml)
return _parse_single_xml_element(elem)
-def _parse_single_xml_element(elem):
+def _parse_single_xml_element(elem: etree._Element):
classmap = {
'record': Record,
'recordtype': RecordType,
@@ -4899,7 +4917,7 @@ def _parse_single_xml_element(elem):
"code"), description=elem.get("description"), body=elem.text)
-def _evaluate_and_add_error(parent_error: TransactionError, ent: Entity):
+def _evaluate_and_add_error(parent_error: TransactionError, ent: Union[Entity, Container]):
"""Evaluate the error message(s) attached to entity and add a
corresponding exception to parent_error.
@@ -4908,7 +4926,7 @@ def _evaluate_and_add_error(parent_error: TransactionError, ent: Entity):
parent_error : TransactionError
Parent error to which the new exception will be attached. This
exception will be a direct child.
- ent : Entity
+ ent : Entity or Container
Entity that caused the TransactionError. An exception is
created depending on its error message(s).
@@ -5029,7 +5047,7 @@ def raise_errors(arg0: Union[Entity, QueryTemplate, Container]):
raise transaction_error
-def delete(ids:Union[List[int], range], raise_exception_on_error=True):
+def delete(ids: Union[List[int], range], raise_exception_on_error=True):
c = Container()
if isinstance(ids, list) or isinstance(ids, range):