From 8df083c46167466d87f88c7807feeacf55d8bbc3 Mon Sep 17 00:00:00 2001 From: serhiy Date: Fri, 28 Jul 2023 13:20:16 +0100 Subject: [PATCH] Remove __future__ usage - the python committee has decided that __annotations__ is not the correct way to deal with cyclical dependencies --- archivist/access_policies.py | 40 ++++++++-------- archivist/appidp.py | 4 +- archivist/applications.py | 26 +++++------ archivist/archivist.py | 38 ++++++++------- archivist/archivistpublic.py | 48 ++++++++++--------- archivist/asset.py | 6 +-- archivist/assetattachments.py | 13 +++--- archivist/assets.py | 41 +++++++++-------- archivist/attachments.py | 19 ++++---- archivist/cmds/runner/run.py | 6 +-- archivist/compliance.py | 11 ++--- archivist/compliance_policies.py | 27 ++++++----- archivist/compliance_policy_requests.py | 5 +- archivist/composite.py | 3 +- archivist/confirmer.py | 32 ++++++------- archivist/dictmerge.py | 10 ++-- archivist/errors.py | 3 +- archivist/events.py | 61 ++++++++++++------------- archivist/headers.py | 4 +- archivist/locations.py | 31 ++++++------- archivist/or_dict.py | 5 +- archivist/runner.py | 27 ++++++----- archivist/sboms.py | 3 +- archivist/subjects.py | 40 ++++++++-------- archivist/subjects_confirmer.py | 7 +-- archivist/tenancies.py | 3 +- archivist/utils.py | 7 ++- 27 files changed, 242 insertions(+), 278 deletions(-) diff --git a/archivist/access_policies.py b/archivist/access_policies.py index bbb6dfe4..a2e1b594 100644 --- a/archivist/access_policies.py +++ b/archivist/access_policies.py @@ -21,11 +21,9 @@ """ -from __future__ import annotations - from copy import deepcopy from logging import getLogger -from typing import TYPE_CHECKING, Any, Generator, Optional +from typing import TYPE_CHECKING, Any, Generator # pylint:disable=cyclic-import # but pylint doesn't understand this feature @@ -47,7 +45,7 @@ class AccessPolicy(dict): """AccessPolicy object""" @property - def name(self) -> str | None: + def name(self) -> "str | None": """str: name of the access policy""" return self.get("display_name") @@ -63,7 +61,7 @@ class _AccessPoliciesClient: """ - def __init__(self, archivist_instance: Archivist): + def __init__(self, archivist_instance: "Archivist"): self._archivist = archivist_instance self._subpath = f"{archivist_instance.root}/{ACCESS_POLICIES_SUBPATH}" self._label = f"{self._subpath}/{ACCESS_POLICIES_LABEL}" @@ -73,9 +71,9 @@ def __str__(self) -> str: def create( self, - props: dict[str, Any], - filters: list[dict[str, Any]], - access_permissions: list[dict[str, Any]], + props: "dict[str, Any]", + filters: "list[dict[str, Any]]", + access_permissions: "list[dict[str, Any]]", ) -> AccessPolicy: """Create access policy @@ -97,7 +95,7 @@ def create( ), ) - def create_from_data(self, data: dict[str, Any]) -> AccessPolicy: + def create_from_data(self, data: "dict[str, Any]") -> AccessPolicy: """Create access policy Creates access policy with request body from data stream. @@ -135,9 +133,9 @@ def update( self, identity, *, - props: Optional[dict[str, Any]] = None, - filters: Optional[list[dict]] = None, - access_permissions: Optional[list[dict]] = None, + props: "dict[str, Any] | None " = None, + filters: "list[dict] | None " = None, + access_permissions: "list[dict] | None " = None, ) -> AccessPolicy: """Update Access Policy @@ -162,7 +160,7 @@ def update( ) ) - def delete(self, identity: str) -> dict[str, Any]: + def delete(self, identity: str) -> "dict[str, Any]": """Delete Access Policy Deletes access policy. @@ -178,11 +176,11 @@ def delete(self, identity: str) -> dict[str, Any]: def __params( self, - props: Optional[dict[str, Any]], + props: "dict[str, Any] | None", *, - filters: list[dict] | None = None, - access_permissions: list[dict] | None = None, - ) -> dict[str, Any]: + filters: "list[dict] | None" = None, + access_permissions: "list[dict] | None" = None, + ) -> "dict[str, Any]": params = deepcopy(props) if props else {} if filters is not None: params["filters"] = filters @@ -192,7 +190,7 @@ def __params( return _deepmerge(self._archivist.fixtures.get(ACCESS_POLICIES_LABEL), params) - def count(self, *, display_name: Optional[str] = None) -> int: + def count(self, *, display_name: "str | None" = None) -> int: """Count access policies. Counts number of access policies that match criteria. @@ -208,7 +206,7 @@ def count(self, *, display_name: Optional[str] = None) -> int: return self._archivist.count(self._label, params=params) def list( - self, *, page_size: Optional[int] = None, display_name: Optional[str] = None + self, *, page_size: "int|None" = None, display_name: "str|None" = None ) -> Generator[AccessPolicy, None, None]: """List access policies. @@ -235,7 +233,7 @@ def list( # additional queries on different endpoints def list_matching_assets( - self, access_policy_id: str, *, page_size: Optional[int] = None + self, access_policy_id: str, *, page_size: "int|None" = None ) -> Generator[Asset, None, None]: """List matching assets. @@ -259,7 +257,7 @@ def list_matching_assets( ) def list_matching_access_policies( - self, asset_id: str, *, page_size: Optional[int] = None + self, asset_id: str, *, page_size: "int|None" = None ) -> Generator[AccessPolicy, None, None]: """List matching access policies. diff --git a/archivist/appidp.py b/archivist/appidp.py index f8100d17..cfc41a83 100644 --- a/archivist/appidp.py +++ b/archivist/appidp.py @@ -21,8 +21,6 @@ """ -from __future__ import annotations - from logging import getLogger from typing import TYPE_CHECKING @@ -57,7 +55,7 @@ class _AppIDPClient: """ - def __init__(self, archivist_instance: Archivist): + def __init__(self, archivist_instance: "Archivist"): self._archivist = archivist_instance self._subpath = f"{archivist_instance.root}/{APPIDP_SUBPATH}" self._label = f"{self._subpath}/{APPIDP_LABEL}" diff --git a/archivist/applications.py b/archivist/applications.py index 98589e83..d49d8102 100644 --- a/archivist/applications.py +++ b/archivist/applications.py @@ -21,10 +21,8 @@ """ -from __future__ import annotations - from logging import getLogger -from typing import TYPE_CHECKING, Any, Optional +from typing import TYPE_CHECKING, Any # pylint:disable=cyclic-import # but pylint doesn't understand this feature if TYPE_CHECKING: @@ -55,7 +53,7 @@ class _ApplicationsClient: """ - def __init__(self, archivist_instance: Archivist): + def __init__(self, archivist_instance: "Archivist"): self._archivist = archivist_instance self._subpath = f"{archivist_instance.root}/{APPLICATIONS_SUBPATH}" self._label = f"{self._subpath}/{APPLICATIONS_LABEL}" @@ -63,7 +61,7 @@ def __init__(self, archivist_instance: Archivist): def __str__(self) -> str: return f"ApplicationsClient({self._archivist.url})" - def create(self, display_name: str, custom_claims: dict[str, str]) -> Application: + def create(self, display_name: str, custom_claims: "dict[str, str]") -> Application: """Create application Creates application with defined attributes. @@ -84,7 +82,7 @@ def create(self, display_name: str, custom_claims: dict[str, str]) -> Applicatio ), ) - def create_from_data(self, data: dict[str, Any]) -> Application: + def create_from_data(self, data: "dict[str, Any]") -> Application: """Create application Creates application with request body from data stream. @@ -117,8 +115,8 @@ def update( self, identity: str, *, - display_name: Optional[str] = None, - custom_claims: Optional[dict[str, str]] = None, + display_name: "str|None" = None, + custom_claims: "dict[str,str]|None" = None, ) -> Application: """Update Application @@ -143,7 +141,7 @@ def update( ) ) - def delete(self, identity: str) -> dict[str, Any]: + def delete(self, identity: str) -> "dict[str, Any]": """Delete Application Deletes application. @@ -160,9 +158,9 @@ def delete(self, identity: str) -> dict[str, Any]: def __params( self, *, - display_name: Optional[str] = None, - custom_claims: Optional[dict[str, str]] = None, - ) -> dict[str, Any]: + display_name: "str|None" = None, + custom_claims: "dict[str,str] | None" = None, + ) -> "dict[str, Any]": params = {} if display_name is not None: @@ -176,8 +174,8 @@ def __params( def list( self, *, - page_size: Optional[int] = None, - display_name: Optional[str] = None, + page_size: "int|None" = None, + display_name: "str|None" = None, ): """List applications. diff --git a/archivist/archivist.py b/archivist/archivist.py index 73844d7a..440c1dab 100644 --- a/archivist/archivist.py +++ b/archivist/archivist.py @@ -29,12 +29,10 @@ attachments, IAM subjects and IAM access policies documented elsewhere. """ -from __future__ import annotations - from copy import deepcopy from logging import getLogger from time import time -from typing import Any, BinaryIO, Optional, Tuple +from typing import Any, BinaryIO from requests_toolbelt.multipart.encoder import MultipartEncoder @@ -104,9 +102,9 @@ class Archivist(ArchivistPublic): # pylint: disable=too-many-instance-attribute def __init__( self, url: str, - auth: str | Tuple[str, str] | None, + auth: "str|tuple[str,str]|None", *, - fixtures: Optional[dict[str, dict[Any, Any]]] = None, + fixtures: "dict[str,dict[Any,Any]]|None" = None, verify: bool = True, max_time: float = MAX_TIME, ): @@ -183,7 +181,7 @@ def root(self) -> str: return self._root @property - def auth(self) -> str | None: + def auth(self) -> "str | None": """str: authorization token""" if self._auth is None and self._machine_auth is None: @@ -208,7 +206,7 @@ def Public(self) -> ArchivistPublic: # pylint: disable=invalid-name max_time=self._max_time, ) - def __copy__(self) -> Archivist: + def __copy__(self) -> "Archivist": return Archivist( self._url, self.auth, @@ -217,7 +215,7 @@ def __copy__(self) -> Archivist: max_time=self._max_time, ) - def _add_headers(self, headers: dict[str, str] | None) -> dict[str, Any]: + def _add_headers(self, headers: "dict[str,str]|None") -> "dict[str,Any]": newheaders = {**headers} if isinstance(headers, dict) else {} auth = self.auth # this may trigger a refetch so only do it once here @@ -233,11 +231,11 @@ def _add_headers(self, headers: dict[str, str] | None) -> dict[str, Any]: def post( self, url: str, - request: dict[str, Any] | None, + request: "dict[str,Any]|None", *, - headers: Optional[dict[str, Any]] = None, - data: dict[str, Any] | bool = False, - ) -> dict[str, Any]: + headers: "dict[str,Any]|None" = None, + data: "dict[str, Any] | bool" = False, + ) -> "dict[str, Any]": """POST method (REST) Creates an entity @@ -276,11 +274,11 @@ def post_file( self, url: str, fd: BinaryIO, - mtype: str | None, + mtype: "str|None", *, form: str = "file", - params: Optional[dict] = None, - ) -> dict[str, Any]: + params: "dict[str, Any]|None" = None, + ) -> "dict[str, Any]": """POST method (REST) - upload binary Uploads a file to an endpoint @@ -321,8 +319,8 @@ def post_file( @retry_429 def delete( - self, url: str, *, headers: Optional[dict[str, Any]] = None - ) -> dict[str, Any]: + self, url: str, *, headers: "dict[str, Any]|None" = None + ) -> "dict[str, Any]": """DELETE method (REST) Deletes an entity @@ -352,10 +350,10 @@ def delete( def patch( self, url: str, - request: dict[str, Any], + request: "dict[str, Any]", *, - headers: Optional[dict[str, Any]] = None, - ) -> dict[str, Any]: + headers: "dict[str, Any]| None" = None, + ) -> "dict[str, Any]": """PATCH method (REST) Updates the specified entity. diff --git a/archivist/archivistpublic.py b/archivist/archivistpublic.py index b35aa0b6..b63ba940 100644 --- a/archivist/archivistpublic.py +++ b/archivist/archivistpublic.py @@ -22,12 +22,10 @@ """ -from __future__ import annotations - from collections import deque from copy import deepcopy from logging import getLogger -from typing import TYPE_CHECKING, Any, BinaryIO, Optional +from typing import TYPE_CHECKING, Any, BinaryIO import requests @@ -81,7 +79,7 @@ class ArchivistPublic: # pylint: disable=too-many-instance-attributes def __init__( self, *, - fixtures: Optional[dict[str, Any]] = None, + fixtures: "dict[str, Any]|None" = None, verify: bool = True, max_time: float = MAX_TIME, ): @@ -153,12 +151,12 @@ def max_time(self) -> float: return self._max_time @property - def fixtures(self) -> dict[str, Any]: + def fixtures(self) -> "dict[str, Any]": """dict: Contains predefined attributes for each endpoint""" return self._fixtures @fixtures.setter - def fixtures(self, fixtures: dict[str, Any]): + def fixtures(self, fixtures: "dict[str, Any]"): """dict: Contains predefined attributes for each endpoint""" self._fixtures = _deepmerge(self._fixtures, fixtures) @@ -169,7 +167,7 @@ def __copy__(self): max_time=self._max_time, ) - def _add_headers(self, headers: Optional[dict]) -> dict[str, str]: + def _add_headers(self, headers: "dict[str, str]|None") -> "dict[str, str]": newheaders = {**headers} if headers is not None else {} return newheaders @@ -182,9 +180,9 @@ def get( self, url: str, *, - headers: Optional[dict[str, str]] = None, - params: Optional[dict[str, Any]] = None, - ) -> dict[str, Any]: + headers: "dict[str, str]|None" = None, + params: "dict[str, Any]|None" = None, + ) -> "dict[str, Any]": """GET method (REST) Args: @@ -217,9 +215,9 @@ def get_file( url: str, fd: BinaryIO, *, - headers: Optional[dict[str, str]] = None, - params: Optional[dict[str, Any]] = None, - ) -> Response: + headers: "dict[str, str]|None" = None, + params: "dict[str, Any]|None" = None, + ) -> "Response": """GET method (REST) - chunked Downloads a binary object from upstream storage. @@ -260,11 +258,11 @@ def get_file( def __list( self, url: str, - params: Optional[dict[str, Any]], + params: "dict[str, Any]|None", *, - page_size: Optional[int] = None, - headers: Optional[dict[str, str]] = None, - ) -> Response: + page_size: "int|None" = None, + headers: "dict[str, str]|None" = None, + ) -> "Response": if page_size is not None: if params is not None: params["page_size"] = page_size @@ -286,7 +284,7 @@ def __list( return response - def last_response(self, *, responses: int = 1) -> list[Response]: + def last_response(self, *, responses: int = 1) -> "list[Response]": """Returns the requested number of response objects from the response ring buffer Args: @@ -303,10 +301,10 @@ def get_by_signature( self, url: str, field: str, - params: dict[str, Any], + params: "dict[str, Any]", *, - headers: Optional[dict[str, str]] = None, - ) -> dict[str, Any]: + headers: "dict[str, str]|None" = None, + ) -> "dict[str, Any]": """GET method (REST) with params string Reads an entity indirectly by searching for its signature @@ -352,7 +350,7 @@ def get_by_signature( return records[0] - def count(self, url: str, *, params: Optional[dict[str, Any]] = None) -> int: + def count(self, url: str, *, params: "dict[str, Any]|None" = None) -> int: """GET method (REST) with params string Returns the count of objects that match params @@ -388,9 +386,9 @@ def list( url: str, field: str, *, - page_size: Optional[int] = None, - params: Optional[dict[str, Any]] = None, - headers: Optional[dict[str, str]] = None, + page_size: "int|None" = None, + params: "dict[str, Any]|None" = None, + headers: "dict[str, str]|None" = None, ): """GET method (REST) with params string diff --git a/archivist/asset.py b/archivist/asset.py index 88f078f2..0e2ad473 100644 --- a/archivist/asset.py +++ b/archivist/asset.py @@ -1,8 +1,6 @@ """Asset data class """ -from __future__ import annotations - from contextlib import suppress @@ -14,7 +12,7 @@ class Asset(dict): """ @property - def primary_image(self) -> str | None: + def primary_image(self) -> "str|None": """Primary Image Attachment that is the primary image of the asset. @@ -49,7 +47,7 @@ def primary_image(self) -> str | None: return None @property - def name(self) -> str | None: + def name(self) -> "str|None": """str: name of the asset""" name = None with suppress(KeyError, TypeError): diff --git a/archivist/assetattachments.py b/archivist/assetattachments.py index 1558d304..82767d7b 100644 --- a/archivist/assetattachments.py +++ b/archivist/assetattachments.py @@ -24,11 +24,10 @@ # pylint:disable=too-few-public-methods -from __future__ import annotations from copy import deepcopy from logging import getLogger -from typing import TYPE_CHECKING, Any, BinaryIO, Optional +from typing import TYPE_CHECKING, Any, BinaryIO from urllib.parse import urlparse if TYPE_CHECKING: @@ -59,7 +58,7 @@ class _AssetAttachmentsClient: """ - def __init__(self, archivist_instance: Archivist): + def __init__(self, archivist_instance: "Archivist"): self._archivist = archivist_instance self._public = archivist_instance.public self._subpath = f"{archivist_instance.root}/{ASSETATTACHMENTS_SUBPATH}" @@ -99,7 +98,7 @@ def _identity(self, identity: str, attachment_id: str) -> str: return f"{self._label}/{identity}/{uuid}" - def __params(self, params: Optional[dict[str, Any]]) -> dict[str, Any]: + def __params(self, params: "dict[str, Any]|None") -> "dict[str, Any]": params = deepcopy(params) if params else {} # pylint: disable=protected-access return _deepmerge(self._archivist.fixtures.get(ATTACHMENTS_LABEL), params) @@ -110,8 +109,8 @@ def download( attachment_id: str, fd: BinaryIO, *, - params: Optional[dict[str, Any]] = None, - ) -> Response: + params: "dict[str, Any]|None" = None, + ) -> "Response": """Read attachment Reads attachment into data sink (usually a file opened for write). @@ -145,7 +144,7 @@ def info( self, identity: str, attachment_id: str, - ) -> dict[str, Any]: + ) -> "dict[str, Any]": """Read asset attachment info Reads asset attachment info diff --git a/archivist/assets.py b/archivist/assets.py index e86ebbdf..f00c0240 100644 --- a/archivist/assets.py +++ b/archivist/assets.py @@ -21,11 +21,10 @@ """ -from __future__ import annotations from copy import deepcopy from logging import getLogger -from typing import TYPE_CHECKING, Any, Optional, Tuple +from typing import TYPE_CHECKING, Any # pylint:disable=cyclic-import # but pylint doesn't understand this feature from . import confirmer @@ -57,7 +56,7 @@ class _AssetsPublic: """ - def __init__(self, archivist_instance: Archivist): + def __init__(self, archivist_instance: "Archivist"): self._archivist = archivist_instance self._public = archivist_instance.public self._subpath = f"{archivist_instance.root}/{ASSETS_SUBPATH}" @@ -100,7 +99,7 @@ class _AssetsRestricted(_AssetsPublic): """ - def __init__(self, archivist_instance: Archivist): + def __init__(self, archivist_instance: "Archivist"): super().__init__(archivist_instance) self._label = f"{self._subpath}/{ASSETS_LABEL}" self.pending_count: int = 0 @@ -109,8 +108,8 @@ def __str__(self) -> str: return f"AssetsRestricted({self._archivist.url})" def __params( - self, props: Optional[dict[str, Any]], attrs: Optional[dict[str, Any]] - ) -> dict[str, Any]: + self, props: "dict[str, Any]|None", attrs: "dict[str, Any]|None" + ) -> "dict[str, Any]": params = deepcopy(props) if props else {} if attrs: params["attributes"] = attrs @@ -120,8 +119,8 @@ def __params( def create( self, *, - props: Optional[dict[str, Any]] = None, - attrs: Optional[dict[str, Any]] = None, + props: "dict[str, Any]|None" = None, + attrs: "dict[str, Any]|None" = None, confirm: bool = True, ) -> Asset: """Create asset @@ -144,7 +143,9 @@ def create( data = self.__params(newprops, attrs) return self.create_from_data(data, confirm=confirm) - def create_from_data(self, data: dict[str, Any], *, confirm: bool = True) -> Asset: + def create_from_data( + self, data: "dict[str, Any]", *, confirm: bool = True + ) -> Asset: """Create asset Creates asset with request body from data stream. @@ -165,8 +166,8 @@ def create_from_data(self, data: dict[str, Any], *, confirm: bool = True) -> Ass return self.wait_for_confirmation(asset["identity"]) def create_if_not_exists( - self, data: dict[str, Any], *, confirm: bool = True - ) -> Tuple[Asset, bool]: + self, data: "dict[str, Any]", *, confirm: bool = True + ) -> "tuple[Asset, bool]": """ Creates an asset and associated locations and attachments if asset does not already exist. @@ -289,8 +290,8 @@ def wait_for_confirmation(self, identity: str) -> Asset: def wait_for_confirmed( self, *, - props: Optional[dict[str, Any]] = None, - attrs: Optional[dict[str, Any]] = None, + props: "dict[str, Any]|None" = None, + attrs: "dict[str, Any]|None" = None, ) -> bool: """Wait for assets to be confirmed. @@ -320,8 +321,8 @@ def wait_for_confirmed( def count( self, *, - props: Optional[dict[str, Any]] = None, - attrs: Optional[dict[str, Any]] = None, + props: "dict[str, Any]|None" = None, + attrs: "dict[str, Any]|None" = None, ) -> int: """Count assets. @@ -340,9 +341,9 @@ def count( def list( self, *, - page_size: Optional[int] = None, - props: Optional[dict[str, Any]] = None, - attrs: Optional[dict[str, Any]] = None, + page_size: "int|None" = None, + props: "dict[str, Any]|None" = None, + attrs: "dict[str, Any]|None" = None, ): """List assets. @@ -370,8 +371,8 @@ def list( def read_by_signature( self, *, - props: Optional[dict[str, Any]] = None, - attrs: Optional[dict[str, Any]] = None, + props: "dict[str, Any]|None" = None, + attrs: "dict[str, Any]|None" = None, ) -> Asset: """Read Asset by signature. diff --git a/archivist/attachments.py b/archivist/attachments.py index 894562f6..c11b88e6 100644 --- a/archivist/attachments.py +++ b/archivist/attachments.py @@ -24,13 +24,12 @@ # pylint:disable=too-few-public-methods -from __future__ import annotations from copy import deepcopy from io import BytesIO from logging import getLogger from os import path -from typing import TYPE_CHECKING, Any, BinaryIO, Optional +from typing import TYPE_CHECKING, Any, BinaryIO if TYPE_CHECKING: from requests.models import Response @@ -67,7 +66,7 @@ class _AttachmentsClient: """ - def __init__(self, archivist_instance: Archivist): + def __init__(self, archivist_instance: "Archivist"): self._archivist = archivist_instance self._subpath = f"{archivist_instance.root}/{ATTACHMENTS_SUBPATH}" self._label = f"{self._subpath}/{ATTACHMENTS_LABEL}" @@ -75,7 +74,7 @@ def __init__(self, archivist_instance: Archivist): def __str__(self) -> str: return f"AttachmentsClient({self._archivist.url})" - def get_default_key(self, data: dict[str, str]) -> str: + def get_default_key(self, data: "dict[str, str]") -> str: """ Return a key to use if no key was provided either use filename or url as one of them is required @@ -87,7 +86,7 @@ def get_default_key(self, data: dict[str, str]) -> str: ) return attachment_key.replace(".", "_") - def create(self, data: dict[str, Any]) -> dict[str, Any]: # pragma: no cover + def create(self, data: "dict[str, Any]") -> "dict[str, Any]": # pragma: no cover """ Create an attachment and return struct suitable for use in an asset or event creation. @@ -159,7 +158,7 @@ def create(self, data: dict[str, Any]) -> dict[str, Any]: # pragma: no cover return result - def upload(self, fd: BinaryIO, *, mtype: Optional[str] = None) -> Attachment: + def upload(self, fd: BinaryIO, *, mtype: "str|None" = None) -> Attachment: """Create attachment Creates attachment from opened file or other data source. @@ -182,7 +181,7 @@ def upload(self, fd: BinaryIO, *, mtype: Optional[str] = None) -> Attachment: ) ) - def __params(self, params: Optional[dict[str, Any]]) -> dict[str, Any]: + def __params(self, params: "dict[str, Any]|None") -> "dict[str, Any]": params = deepcopy(params) if params else {} # pylint: disable=protected-access return _deepmerge(self._archivist.fixtures.get(ATTACHMENTS_LABEL), params) @@ -192,8 +191,8 @@ def download( identity: str, fd: BinaryIO, *, - params: Optional[dict[str, Any]] = None, - ) -> Response: + params: "dict[str, Any]|None" = None, + ) -> "Response": """Read attachment Reads attachment into data sink (usually a file opened for write).. @@ -218,7 +217,7 @@ def download( def info( self, identity: str, - ) -> dict[str, Any]: + ) -> "dict[str, Any]": """Read attachment info Reads attachment info diff --git a/archivist/cmds/runner/run.py b/archivist/cmds/runner/run.py index b57e9d7e..2c6eba84 100644 --- a/archivist/cmds/runner/run.py +++ b/archivist/cmds/runner/run.py @@ -1,6 +1,5 @@ # pylint: disable=missing-docstring -from __future__ import annotations from logging import getLogger from os import environ @@ -13,13 +12,12 @@ from ... import about if TYPE_CHECKING: - from ..archivist import Archivist - + from ...archivist import Archivist LOGGER = getLogger(__name__) -def run(arch: Archivist, args): +def run(arch: "Archivist", args): LOGGER.info("Using version %s of rkvst-archivist", about.__version__) LOGGER.info("Namespace %s", args.namespace) diff --git a/archivist/compliance.py b/archivist/compliance.py index fa6e3387..a8ae7b1b 100644 --- a/archivist/compliance.py +++ b/archivist/compliance.py @@ -21,10 +21,9 @@ """ -from __future__ import annotations from logging import getLogger -from typing import TYPE_CHECKING, Any, Optional +from typing import TYPE_CHECKING, Any if TYPE_CHECKING: # pylint:disable=cyclic-import # but pylint doesn't understand this feature @@ -58,7 +57,7 @@ class _ComplianceClient: # pylint: disable=too-few-public-methods """ - def __init__(self, archivist_instance: Archivist): + def __init__(self, archivist_instance: "Archivist"): self._archivist = archivist_instance self._subpath = f"{archivist_instance.root}/{COMPLIANCE_SUBPATH}" self._label = f"{self._subpath}/{COMPLIANCE_LABEL}" @@ -70,8 +69,8 @@ def compliant_at( self, asset_id, *, - compliant_at: Optional[bool] = None, - report: Optional[str] = None, + compliant_at: "bool|None" = None, + report: "str|None" = None, ) -> Compliance: """ Reads compliance of a particular asset. @@ -97,7 +96,7 @@ def compliant_at( self.compliant_at_report(response) return Compliance(**response) - def compliant_at_report(self, compliance: dict[str, Any]): + def compliant_at_report(self, compliance: "dict[str, Any]"): """ Prints report of compliance_at request diff --git a/archivist/compliance_policies.py b/archivist/compliance_policies.py index ac6cfe75..337e32e4 100644 --- a/archivist/compliance_policies.py +++ b/archivist/compliance_policies.py @@ -25,11 +25,10 @@ """ -from __future__ import annotations from copy import deepcopy from logging import getLogger -from typing import TYPE_CHECKING, Any, Optional, Union +from typing import TYPE_CHECKING, Any, Union if TYPE_CHECKING: # pylint:disable=cyclic-import # but pylint doesn't understand this feature @@ -75,7 +74,7 @@ class _CompliancePoliciesClient: """ - def __init__(self, archivist_instance: Archivist): + def __init__(self, archivist_instance: "Archivist"): self._archivist = archivist_instance self._subpath = f"{archivist_instance.root}/{COMPLIANCE_POLICIES_SUBPATH}" self._label = f"{self._subpath}/{COMPLIANCE_POLICIES_LABEL}" @@ -86,11 +85,11 @@ def __str__(self) -> str: def create( self, policy: Union[ - CompliancePolicySince, - CompliancePolicyCurrentOutstanding, - CompliancePolicyPeriodOutstanding, - CompliancePolicyDynamicTolerance, - CompliancePolicyRichness, + "CompliancePolicySince", + "CompliancePolicyCurrentOutstanding", + "CompliancePolicyPeriodOutstanding", + "CompliancePolicyDynamicTolerance", + "CompliancePolicyRichness", ], ) -> CompliancePolicy: """Create A compliance policy @@ -110,7 +109,7 @@ def create( """ return self.create_from_data(policy.dict()) - def create_from_data(self, data: dict[str, Any]) -> CompliancePolicy: + def create_from_data(self, data: "dict[str, Any]") -> "CompliancePolicy": """Create compliance_policy Creates compliance_policy with request body from data stream. @@ -140,7 +139,7 @@ def read(self, identity: str) -> CompliancePolicy: """ return CompliancePolicy(**self._archivist.get(f"{self._subpath}/{identity}")) - def delete(self, identity: str) -> dict[str, Any]: + def delete(self, identity: str) -> "dict[str, Any]": """Delete Compliance Policy Deletes compliance policy. @@ -155,14 +154,14 @@ def delete(self, identity: str) -> dict[str, Any]: """ return self._archivist.delete(f"{self._subpath}/{identity}") - def __params(self, props: Optional[dict[str, Any]]) -> dict[str, Any]: + def __params(self, props: "dict[str, Any]|None") -> "dict[str, Any]": params = deepcopy(props) if props else {} # pylint: disable=protected-access return _deepmerge( self._archivist.fixtures.get(COMPLIANCE_POLICIES_LABEL), params ) - def count(self, *, props: Optional[dict[str, Any]] = None) -> int: + def count(self, *, props: "dict[str, Any]|None" = None) -> int: """Count compliance policies. Counts number of compliance policies that match criteria. @@ -180,7 +179,7 @@ def count(self, *, props: Optional[dict[str, Any]] = None) -> int: ) def list( - self, *, page_size: Optional[int] = None, props: Optional[dict[str, Any]] = None + self, *, page_size: "int|None" = None, props: "dict[str, Any]|None" = None ): """List compliance policies. @@ -204,7 +203,7 @@ def list( ) ) - def read_by_signature(self, *, props: Optional[dict[str, Any]] = None): + def read_by_signature(self, *, props: "dict[str, Any]|None" = None): """Read compliance policy by signature. Reads compliance policy that meets criteria. Only one compliance policy is expected. diff --git a/archivist/compliance_policy_requests.py b/archivist/compliance_policy_requests.py index 1e022aa9..1518749f 100644 --- a/archivist/compliance_policy_requests.py +++ b/archivist/compliance_policy_requests.py @@ -4,7 +4,6 @@ """ -from __future__ import annotations from dataclasses import asdict, dataclass @@ -24,7 +23,7 @@ class CompliancePolicyBase: description: str display_name: str - asset_filter: list[list] + asset_filter: "list[list]" def dict(self): """Emit dictionary representation""" @@ -90,7 +89,7 @@ class CompliancePolicyRichness(CompliancePolicyBase): complies with a set of assertions. """ - richness_assertions: list[list] + richness_assertions: "list[list]" compliance_type: str = CompliancePolicyType.COMPLIANCE_RICHNESS.name def dict(self): diff --git a/archivist/composite.py b/archivist/composite.py index cdc48624..4da25e35 100644 --- a/archivist/composite.py +++ b/archivist/composite.py @@ -21,7 +21,6 @@ """ -from __future__ import annotations from logging import getLogger from typing import TYPE_CHECKING @@ -46,7 +45,7 @@ class _CompositeClient: These mthods are not unittested and provided as a convenience. """ - def __init__(self, archivist_instance: Archivist): + def __init__(self, archivist_instance: "Archivist"): self._archivist = archivist_instance def __str__(self) -> str: diff --git a/archivist/confirmer.py b/archivist/confirmer.py index 7a16f09c..a19d9d84 100644 --- a/archivist/confirmer.py +++ b/archivist/confirmer.py @@ -1,11 +1,10 @@ """assets confirmer interface """ -from __future__ import annotations from copy import deepcopy from logging import getLogger -from typing import TYPE_CHECKING, Any, Optional, Union, overload +from typing import TYPE_CHECKING, Any, Union, overload import backoff @@ -13,7 +12,8 @@ # pylint:disable=cyclic-import # but pylint doesn't understand this feature from backoff._typing import Details - from . import assets, events + from .assets import Asset, _AssetsPublic, _AssetsRestricted + from .events import Event, _EventsPublic, _EventsRestricted from .constants import ( @@ -28,13 +28,11 @@ MAX_TIME = 1200 LOGGER = getLogger(__name__) -if TYPE_CHECKING: - # pylint: disable=protected-access - PublicManagers = Union[assets._AssetsPublic, events._EventsPublic] - PrivateManagers = Union[assets._AssetsRestricted, events._EventsRestricted] - Managers = Union[PublicManagers, PrivateManagers] - - ReturnTypes = Union[assets.Asset, events.Event] +# pylint: disable=protected-access +PublicManagers = Union["_AssetsPublic", "_EventsPublic"] +PrivateManagers = Union["_AssetsRestricted", "_EventsRestricted"] +Managers = Union[PublicManagers, PrivateManagers] +ReturnTypes = Union["Asset", "Event"] def __lookup_max_time(): @@ -56,26 +54,22 @@ def __on_giveup_confirmation(details: "Details"): @overload -def _wait_for_confirmation( - self: assets._AssetsRestricted, identity: str -) -> assets.Asset: +def _wait_for_confirmation(self: "_AssetsRestricted", identity: str) -> "Asset": ... # pragma: no cover @overload -def _wait_for_confirmation(self: assets._AssetsPublic, identity: str) -> assets.Asset: +def _wait_for_confirmation(self: "_AssetsPublic", identity: str) -> "Asset": ... # pragma: no cover @overload -def _wait_for_confirmation( - self: events._EventsRestricted, identity: str -) -> events.Event: +def _wait_for_confirmation(self: "_EventsRestricted", identity: str) -> "Event": ... # pragma: no cover @overload -def _wait_for_confirmation(self: events._EventsPublic, identity: str) -> events.Event: +def _wait_for_confirmation(self: "_EventsPublic", identity: str) -> "Event": ... # pragma: no cover @@ -125,7 +119,7 @@ def __on_giveup_confirmed(details: "Details"): on_giveup=__on_giveup_confirmed, ) def _wait_for_confirmed( - self: PrivateManagers, *, props: Optional[dict[str, Any]] = None, **kwargs: Any + self: PrivateManagers, *, props: "dict[str, Any]|None" = None, **kwargs: Any ) -> bool: """Return False until all entities are confirmed""" diff --git a/archivist/dictmerge.py b/archivist/dictmerge.py index de837798..e17b9a69 100644 --- a/archivist/dictmerge.py +++ b/archivist/dictmerge.py @@ -1,16 +1,16 @@ """Archivist dict deep merge """ -from __future__ import annotations + from copy import deepcopy -from typing import Any, Optional +from typing import Any from flatten_dict import flatten, unflatten def _deepmerge( - dct1: Optional[dict[str, Any]], dct2: Optional[dict[str, Any]] -) -> dict[str, Any]: + dct1: "dict[str, Any]|None", dct2: "dict[str, Any]|None" +) -> "dict[str, Any]": """Deep merge 2 dictionaries The settings from dct2 overwrite or add to dct1 @@ -26,7 +26,7 @@ def _deepmerge( return unflatten({**flatten(dct1), **flatten(dct2)}) -def _dotdict(dct: Optional[dict[str, Any]]) -> dict[str, str] | None: +def _dotdict(dct: "dict[str, Any]|None") -> "dict[str, str] | None": """Emit nested dictionary as dot delimited dict with one level""" if dct is None: return None diff --git a/archivist/errors.py b/archivist/errors.py index 33573753..6fa05e1f 100644 --- a/archivist/errors.py +++ b/archivist/errors.py @@ -6,7 +6,6 @@ import json from logging import getLogger -from typing import Optional from requests import Response @@ -59,7 +58,7 @@ class ArchivistNotFoundError(ArchivistError): class ArchivistTooManyRequestsError(ArchivistError): """Too many requests in too short a time (429)""" - def __init__(self, retry: Optional[str], *args): + def __init__(self, retry: "str|None", *args): self.retry = float(retry) if retry is not None else 0 super().__init__(*args) diff --git a/archivist/events.py b/archivist/events.py index a852eb1f..9627eb04 100644 --- a/archivist/events.py +++ b/archivist/events.py @@ -22,11 +22,10 @@ """ -from __future__ import annotations from copy import deepcopy from logging import getLogger -from typing import TYPE_CHECKING, Any, Optional +from typing import TYPE_CHECKING, Any # pylint:disable=cyclic-import # but pylint doesn't understand this feature from . import confirmer @@ -55,7 +54,7 @@ class Event(dict): """ @property - def when(self) -> str | None: + def when(self) -> "str | None": """when Timestamp of event @@ -78,7 +77,7 @@ def when(self) -> str | None: return None @property - def who(self) -> str | None: + def who(self) -> "str | None": """who Principal identity. @@ -113,7 +112,7 @@ class _EventsPublic: """ - def __init__(self, archivist_instance: Archivist): + def __init__(self, archivist_instance: "Archivist"): self._archivist = archivist_instance self._public = archivist_instance.public self._subpath = f"{archivist_instance.root}/{ASSETS_SUBPATH}" @@ -147,10 +146,10 @@ def read(self, identity: str) -> Event: def _params( self, - props: Optional[dict[str, Any]], - attrs: Optional[dict[str, Any]], - asset_attrs: Optional[dict[str, Any]], - ) -> dict[str, Any]: + props: "dict[str, Any]|None", + attrs: "dict[str, Any]|None", + asset_attrs: "dict[str, Any]|None", + ) -> "dict[str, Any]": params = deepcopy(props) if props else {} if attrs: params["event_attributes"] = attrs @@ -162,10 +161,10 @@ def _params( def count( self, *, - asset_id: Optional[str] = None, - props: Optional[dict[str, Any]] = None, - attrs: Optional[dict[str, Any]] = None, - asset_attrs: Optional[dict[str, Any]] = None, + asset_id: "str|None" = None, + props: "dict[str, Any]|None" = None, + attrs: "dict[str, Any]|None" = None, + asset_attrs: "dict[str, Any]|None" = None, ) -> int: """Count events. @@ -202,11 +201,11 @@ def count( def list( self, *, - asset_id: Optional[str] = None, - page_size: Optional[int] = None, - props: Optional[dict[str, Any]] = None, - attrs: Optional[dict[str, Any]] = None, - asset_attrs: Optional[dict[str, Any]] = None, + asset_id: "str|None" = None, + page_size: "int|None" = None, + props: "dict[str, Any]|None" = None, + attrs: "dict[str, Any]|None" = None, + asset_attrs: "dict[str, Any]|None" = None, ): """List events. @@ -241,10 +240,10 @@ def list( def read_by_signature( self, *, - asset_id: Optional[str] = None, - props: Optional[dict[str, Any]] = None, - attrs: Optional[dict[str, Any]] = None, - asset_attrs: Optional[dict[str, Any]] = None, + asset_id: "str|None" = None, + props: "dict[str, Any]|None" = None, + attrs: "dict[str, Any]|None" = None, + asset_attrs: "dict[str, Any]|None" = None, ) -> Event: """Read event by signature. @@ -285,7 +284,7 @@ class _EventsRestricted(_EventsPublic): """ - def __init__(self, archivist_instance: Archivist): + def __init__(self, archivist_instance: "Archivist"): super().__init__(archivist_instance) self.pending_count: int = 0 @@ -295,10 +294,10 @@ def __str__(self) -> str: def create( self, asset_id: str, - props: dict[str, Any], - attrs: dict[str, Any], + props: "dict[str, Any]", + attrs: "dict[str, Any]", *, - asset_attrs: Optional[dict[str, Any]] = None, + asset_attrs: "dict[str, Any]|None" = None, confirm: bool = True, ) -> Event: """Create event @@ -325,7 +324,7 @@ def create( ) def create_from_data( - self, asset_id: str, data: dict[str, Any], *, confirm: bool = True + self, asset_id: str, data: "dict[str, Any]", *, confirm: bool = True ) -> Event: """Create event @@ -402,10 +401,10 @@ def wait_for_confirmation(self, identity: str) -> Event: def wait_for_confirmed( self, *, - asset_id: Optional[str] = None, - props: Optional[dict[str, Any]] = None, - attrs: Optional[dict[str, Any]] = None, - asset_attrs: Optional[dict[str, Any]] = None, + asset_id: "str|None" = None, + props: "dict[str, Any]|None" = None, + attrs: "dict[str, Any]|None" = None, + asset_attrs: "dict[str, Any]|None" = None, ) -> bool: """Wait for events to be confirmed. diff --git a/archivist/headers.py b/archivist/headers.py index 53cb8c19..8ae55075 100644 --- a/archivist/headers.py +++ b/archivist/headers.py @@ -4,12 +4,10 @@ """ -from typing import Optional - from requests import models -def _headers_get(headers: models.CaseInsensitiveDict, key: str) -> Optional[str]: +def _headers_get(headers: models.CaseInsensitiveDict, key: str) -> "str|None": if headers is not None: ret = headers.get(key) if ret is not None: diff --git a/archivist/locations.py b/archivist/locations.py index 2d170bec..7b88c13d 100644 --- a/archivist/locations.py +++ b/archivist/locations.py @@ -22,12 +22,11 @@ """ -from __future__ import annotations from contextlib import suppress from copy import deepcopy from logging import getLogger -from typing import TYPE_CHECKING, Any, Optional, Tuple +from typing import TYPE_CHECKING, Any if TYPE_CHECKING: # pylint:disable=cyclic-import # but pylint doesn't understand this feature @@ -49,7 +48,7 @@ class Location(dict): """ @property - def name(self) -> str | None: + def name(self) -> "str | None": """str: name of the location""" name = None with suppress(KeyError): @@ -69,7 +68,7 @@ class _LocationsClient: """ - def __init__(self, archivist_instance: Archivist): + def __init__(self, archivist_instance: "Archivist"): self._archivist = archivist_instance self._subpath = f"{archivist_instance.root}/{LOCATIONS_SUBPATH}" self._label = f"{self._subpath}/{LOCATIONS_LABEL}" @@ -78,7 +77,7 @@ def __str__(self) -> str: return f"LocationsClient({self._archivist.url})" def create( - self, props: dict[str, Any], *, attrs: Optional[dict[str, Any]] = None + self, props: "dict[str, Any]", *, attrs: "dict[str, Any]|None" = None ) -> Location: """Create location @@ -95,7 +94,7 @@ def create( LOGGER.debug("Create Location %s", props) return self.create_from_data(self.__params(props, attrs)) - def create_from_data(self, data: dict[str, Any]) -> Location: + def create_from_data(self, data: "dict[str, Any]") -> Location: """Create location Creates location with request body from data stream. @@ -110,7 +109,7 @@ def create_from_data(self, data: dict[str, Any]) -> Location: """ return Location(**self._archivist.post(self._label, data)) - def create_if_not_exists(self, data: dict[str, Any]) -> Tuple[Location, bool]: + def create_if_not_exists(self, data: "dict[str, Any]") -> "tuple[Location, bool]": """ Create a location if not already exists @@ -172,8 +171,8 @@ def read(self, identity: str) -> Location: return Location(**self._archivist.get(f"{self._subpath}/{identity}")) def __params( - self, props: Optional[dict[str, Any]], attrs: Optional[dict[str, Any]] - ) -> dict[str, Any]: + self, props: "dict[str, Any]|None", attrs: "dict[str, Any]|None" + ) -> "dict[str, Any]": params = deepcopy(props) if props else {} if attrs: params["attributes"] = attrs @@ -183,8 +182,8 @@ def __params( def count( self, *, - props: Optional[dict[str, Any]] = None, - attrs: Optional[dict[str, Any]] = None, + props: "dict[str, Any]|None" = None, + attrs: "dict[str, Any]|None" = None, ) -> int: """Count locations. @@ -203,9 +202,9 @@ def count( def list( self, *, - page_size: Optional[int] = None, - props: Optional[dict[str, Any]] = None, - attrs: Optional[dict[str, Any]] = None, + page_size: "int|None" = None, + props: "dict[str, Any]|None" = None, + attrs: "dict[str, Any]|None" = None, ): """List locations. @@ -234,8 +233,8 @@ def list( def read_by_signature( self, *, - props: Optional[dict[str, Any]] = None, - attrs: Optional[dict[str, Any]] = None, + props: "dict[str, Any]|None" = None, + attrs: "dict[str, Any]|None" = None, ) -> Location: """Read location by signature. diff --git a/archivist/or_dict.py b/archivist/or_dict.py index 0690360f..bcea5bee 100644 --- a/archivist/or_dict.py +++ b/archivist/or_dict.py @@ -5,16 +5,15 @@ Dictionaries where key is always 'or' and value is a list of strings """ -from __future__ import annotations from typing import Any -def or_dict(list_: list) -> dict[str, list]: +def or_dict(list_: list) -> "dict[str, list]": """Construct a dictionary with key 'or'""" return {"or": list_} -def and_list(lists: list) -> list[dict[str, list[Any]]]: +def and_list(lists: list) -> "list[dict[str, list[Any]]]": """Construct a list of or dictionaries""" return [or_dict(j) for j in lists] diff --git a/archivist/runner.py b/archivist/runner.py index 4e5045a0..3e76e0a7 100644 --- a/archivist/runner.py +++ b/archivist/runner.py @@ -3,7 +3,6 @@ """ -from __future__ import annotations from collections import defaultdict from functools import partialmethod @@ -11,7 +10,7 @@ from logging import getLogger from time import sleep as time_sleep from types import GeneratorType -from typing import TYPE_CHECKING, Any, Callable, Optional, Tuple +from typing import TYPE_CHECKING, Any, Callable from uuid import UUID from .errors import ArchivistError, ArchivistInvalidOperationError @@ -50,7 +49,7 @@ class _ActionMap(dict): # a dictionary # similarly for location and subjects labels # - def __init__(self, archivist_instance: Archivist): + def __init__(self, archivist_instance: "Archivist"): super().__init__() self._archivist = archivist_instance @@ -192,7 +191,7 @@ def __init__(self, archivist_instance: Archivist): "use_subject_label": "add_arg_identity", } - def ops(self, action_name: str) -> dict[str, Any]: + def ops(self, action_name: str) -> "dict[str, Any]": """ Get valid entry in map """ @@ -208,7 +207,7 @@ def action(self, action_name: str) -> Callable: # if an exception occurs here then the dict initialized above is faulty. return self.ops(action_name).get("action") # pyright: ignore - def keywords(self, action_name: str) -> Tuple | None: + def keywords(self, action_name: str) -> "tuple | None": """ Get keywords in map """ @@ -228,11 +227,11 @@ def label(self, noun: str, endpoint: str, action_name: str) -> bool: class _Step(dict): # pylint:disable=too-many-instance-attributes - def __init__(self, archivist_instance: Archivist, **kwargs): + def __init__(self, archivist_instance: "Archivist", **kwargs): super().__init__(**kwargs) self._archivist = archivist_instance - self._args: list[Any] = [] - self._kwargs: dict[str, Any] = {} + self._args: "list[Any]" = [] + self._kwargs: "dict[str, Any]" = {} self._actions = None self._action = None self._action_name = None @@ -397,7 +396,7 @@ class _Runner: ArchivistRunner takes a url, token_file. """ - def __init__(self, archivist_instance: Archivist): + def __init__(self, archivist_instance: "Archivist"): self._archivist = archivist_instance self.entities: defaultdict self.deletions = {} @@ -405,7 +404,7 @@ def __init__(self, archivist_instance: Archivist): def __str__(self) -> str: return f"Runner({self._archivist.url})" - def __call__(self, config: dict[str, Any]): + def __call__(self, config: "dict[str, Any]"): """ The dict config contains a list of `steps` to be performed serially, e.g. @@ -441,7 +440,7 @@ def __call__(self, config: dict[str, Any]): except (ArchivistError, KeyError) as ex: LOGGER.info("Runner exception %s", ex) - def run_steps(self, config: dict[str, Any]): + def run_steps(self, config: "dict[str, Any]"): """Runs all defined steps in self.config.""" self.entities = tree() for step in config["steps"]: @@ -450,7 +449,7 @@ def run_steps(self, config: dict[str, Any]): self.delete() self._archivist.close() - def run_step(self, step: dict[str, Any]): + def run_step(self, step: "dict[str, Any]"): """Runs a step given parameters and the type of step. Args: @@ -480,7 +479,7 @@ def run_step(self, step: dict[str, Any]): if s.label("set", noun) and label is not None: self.entities[label] = response - def set_deletions(self, response: dict[str, Any], delete_method): + def set_deletions(self, response: "dict[str, Any]", delete_method): """sets entry to be deleted""" if delete_method is not None: @@ -493,7 +492,7 @@ def delete(self): LOGGER.info("Delete %s", identity) delete_method(identity) - def identity(self, name: str) -> Optional[str]: + def identity(self, name: str) -> "str|None": """Gets entity id""" identity = self.entities[name]["identity"] diff --git a/archivist/sboms.py b/archivist/sboms.py index e4b42ef1..8d030f5f 100644 --- a/archivist/sboms.py +++ b/archivist/sboms.py @@ -6,7 +6,6 @@ # pylint:disable=too-few-public-methods -from __future__ import annotations from io import BytesIO from logging import getLogger @@ -19,7 +18,7 @@ LOGGER = getLogger(__name__) -def sboms_parse(data: dict[str, Any]) -> dict[str, Any]: # pragma: no cover +def sboms_parse(data: "dict[str, Any]") -> "dict[str, Any]": # pragma: no cover """ parse the sbom and extract pertinent information diff --git a/archivist/subjects.py b/archivist/subjects.py index 40dd4043..9f699fd9 100644 --- a/archivist/subjects.py +++ b/archivist/subjects.py @@ -21,12 +21,11 @@ """ -from __future__ import annotations from base64 import b64decode from json import loads as json_loads from logging import getLogger -from typing import TYPE_CHECKING, Any, Optional, Tuple +from typing import TYPE_CHECKING, Any # pylint:disable=cyclic-import # but pylint doesn't understand this feature from . import subjects_confirmer @@ -60,7 +59,7 @@ class _SubjectsClient: maxDiff = None - def __init__(self, archivist_instance: Archivist): + def __init__(self, archivist_instance: "Archivist"): self._archivist = archivist_instance self._subpath = f"{archivist_instance.root}/{SUBJECTS_SUBPATH}" self._label = f"{self._subpath}/{SUBJECTS_LABEL}" @@ -69,7 +68,10 @@ def __str__(self) -> str: return f"SubjectsClient({self._archivist.url})" def create( - self, display_name: str, wallet_pub_key: list[str], tessera_pub_key: list[str] + self, + display_name: str, + wallet_pub_key: "list[str]", + tessera_pub_key: "list[str]", ) -> Subject: """Create subject @@ -94,8 +96,8 @@ def create( ) def share( - self, name: str, other_name: str, other_archivist: Archivist - ) -> Tuple[Subject, Subject]: + self, name: str, other_name: str, other_archivist: "Archivist" + ) -> "tuple[Subject, Subject]": """Import the self subjects from the foreign archivist connection from another organization - mutually share. @@ -137,7 +139,7 @@ def import_subject(self, display_name: str, subject: Subject) -> Subject: subject["tessera_pub_key"], ) - def create_from_data(self, data: dict[str, Any]) -> Subject: + def create_from_data(self, data: "dict[str, Any]") -> Subject: """Create subject Creates subject with request body from data stream. @@ -153,7 +155,7 @@ def create_from_data(self, data: dict[str, Any]) -> Subject: LOGGER.debug("Create Subject from data %s", data) return Subject(**self._archivist.post(self._label, data)) - def create_from_b64(self, data: dict[str, Any]) -> Subject: + def create_from_b64(self, data: "dict[str, Any]") -> Subject: """Create subject Creates subject with request body from b64 encoded string @@ -218,9 +220,9 @@ def update( self, identity: str, *, - display_name: Optional[str] = None, - wallet_pub_key: Optional[list[str]] = None, - tessera_pub_key: Optional[list[str]] = None, + display_name: "str|None" = None, + wallet_pub_key: "list[str]|None" = None, + tessera_pub_key: "list[str]|None" = None, ) -> Subject: """Update Subject @@ -247,7 +249,7 @@ def update( ) ) - def delete(self, identity: str) -> dict[str, Any]: + def delete(self, identity: str) -> "dict[str, Any]": """Delete Subject Deletes subject. @@ -264,10 +266,10 @@ def delete(self, identity: str) -> dict[str, Any]: def __params( self, *, - display_name: Optional[str] = None, - wallet_pub_key: Optional[list[str]] = None, - tessera_pub_key: Optional[list[str]] = None, - ) -> dict[str, Any]: + display_name: "str|None" = None, + wallet_pub_key: "list[str]|None" = None, + tessera_pub_key: "list[str]|None" = None, + ) -> "dict[str, Any]": params = {} if display_name is not None: @@ -281,7 +283,7 @@ def __params( return _deepmerge(self._archivist.fixtures.get(SUBJECTS_LABEL), params) - def count(self, *, display_name: Optional[str] = None) -> int: + def count(self, *, display_name: "str|None" = None) -> int: """Count subjects. Counts number of subjects that match criteria. @@ -301,8 +303,8 @@ def count(self, *, display_name: Optional[str] = None) -> int: def list( self, *, - page_size: Optional[int] = None, - display_name: Optional[str] = None, + page_size: "int|None" = None, + display_name: "str|None" = None, ): """List subjects. diff --git a/archivist/subjects_confirmer.py b/archivist/subjects_confirmer.py index c4797b08..059808f2 100644 --- a/archivist/subjects_confirmer.py +++ b/archivist/subjects_confirmer.py @@ -1,7 +1,6 @@ """assets confirmer interface """ -from __future__ import annotations from logging import getLogger from typing import TYPE_CHECKING @@ -9,7 +8,7 @@ import backoff if TYPE_CHECKING: - from . import subjects + from .subjects import Subject, _SubjectsClient from .constants import ( CONFIRMATION_CONFIRMED, @@ -44,9 +43,7 @@ def __on_giveup_confirmation(details): on_backoff=backoff_handler, on_giveup=__on_giveup_confirmation, ) -def _wait_for_confirmation( - self: subjects._SubjectsClient, identity: str -) -> subjects.Subject: +def _wait_for_confirmation(self: "_SubjectsClient", identity: str) -> "Subject": """Return None until subjects is confirmed""" subject = self.read(identity) if CONFIRMATION_STATUS not in subject: diff --git a/archivist/tenancies.py b/archivist/tenancies.py index 59c6f298..947f8d92 100644 --- a/archivist/tenancies.py +++ b/archivist/tenancies.py @@ -21,7 +21,6 @@ """ -from __future__ import annotations from logging import getLogger from typing import TYPE_CHECKING @@ -56,7 +55,7 @@ class _TenanciesClient: maxDiff = None - def __init__(self, archivist_instance: Archivist): + def __init__(self, archivist_instance: "Archivist"): self._archivist = archivist_instance self._subpath = f"{archivist_instance.root}/{TENANCIES_SUBPATH}" self._label = f"{self._subpath}/{TENANCIES_LABEL}" diff --git a/archivist/utils.py b/archivist/utils.py index 40b7f862..cf759be0 100644 --- a/archivist/utils.py +++ b/archivist/utils.py @@ -1,10 +1,9 @@ """Some convenience stuff """ -from __future__ import annotations from logging import getLogger -from typing import TYPE_CHECKING, Tuple +from typing import TYPE_CHECKING if TYPE_CHECKING: from io import BytesIO @@ -41,7 +40,7 @@ def backoff_handler(details): # download arbitrary files from a url. -def get_url(url: str, fd: BytesIO): # pragma: no cover +def get_url(url: str, fd: "BytesIO"): # pragma: no cover """GET method (REST) - chunked Downloads a binary object from upstream storage. @@ -90,7 +89,7 @@ def get_auth( return None -def selector_signature(selector: list, data: dict) -> Tuple[dict, dict | None]: +def selector_signature(selector: list, data: dict) -> "tuple[dict, dict | None]": """ Convert a selector to a signature for list and count methods