use less of Any as type

ruff.toml

@@ -3,6 +3,10 @@ line-length = 120
 target-version = "py310"
 
 [lint]
-select = ["E", "F", "W", "I", "ASYNC", "UP", "FLY", "PERF", "FURB", "ERA", "ANN0", "ANN2"]
+select = ["E", "F", "W", "I", "ASYNC", "UP", "FLY", "PERF", "FURB", "ERA", "ANN"]
 
-ignore = ["E501"]
+# ANN101 and ANN102 are depracated
+ignore = ["E501", "ANN1"]
+
+[lint.flake8-annotations]
+allow-star-arg-any = true

src/eduid/common/clients/amapi_client/amapi_client.py

@@ -9,6 +9,7 @@
 __author__ = "masv"
 
 from eduid.common.models.amapi_user import (
+    UserBaseRequest,
     UserUpdateEmailRequest,
     UserUpdateLanguageRequest,
     UserUpdateMetaCleanedRequest,
@@ -27,7 +28,7 @@ def __init__(self, amapi_url: str, auth_data: GNAPClientAuthData, verify_tls: bo
     def _users_base_url(self) -> str:
         return urlappend(self.amapi_url, "users")
 
-    def _put(self, base_path: str, user: str, endpoint: str, body: Any) -> httpx.Response:
+    def _put(self, base_path: str, user: str, endpoint: str, body: UserBaseRequest) -> httpx.Response:
         return self.put(url=urlappend(base_path, f"{user}/{endpoint}"), content=body.json())
 
     def update_user_email(self, user: str, body: UserUpdateEmailRequest) -> UserUpdateResponse:

src/eduid/common/decorators.py

@@ -29,7 +29,7 @@ def decorator(func1: Callable) -> Callable:
                 fmt1 = "Call to deprecated function {name} ({reason})."
 
             @wraps(func1)
-            def new_func1(*args: Any, **kwargs: Any) -> Any:
+            def new_func1(*args: Any, **kwargs: Any) -> Any:  # noqa: ANN401
                 warnings.simplefilter("always", DeprecationWarning)
                 warnings.warn(
                     fmt1.format(name=func1.__name__, reason=reason), category=DeprecationWarning, stacklevel=2
@@ -58,7 +58,7 @@ def new_func1(*args: Any, **kwargs: Any) -> Any:
             fmt2 = "Call to deprecated function {name}."
 
         @wraps(func2)
-        def new_func2(*args: Any, **kwargs: Any) -> Any:
+        def new_func2(*args: Any, **kwargs: Any) -> Any:  # noqa: ANN401
             warnings.simplefilter("always", DeprecationWarning)
             warnings.warn(fmt2.format(name=func2.__name__), category=DeprecationWarning, stacklevel=2)
             warnings.simplefilter("default", DeprecationWarning)

src/eduid/common/logging.py

@@ -139,7 +139,7 @@ def filter(self, record: logging.LogRecord) -> bool:
 def merge_config(base_config: dict[str, Any], new_config: dict[str, Any]) -> dict[str, Any]:
     """Recursively merge two dictConfig dicts."""
 
-    def merge(node: dict[str, Any], key: str, value: Any) -> None:
+    def merge(node: dict[str, Any], key: str, value: object) -> None:
         if isinstance(value, dict):
             for item in value:
                 if key in node:

src/eduid/common/misc/encoders.py

@@ -11,7 +11,7 @@
 class EduidJSONEncoder(json.JSONEncoder):
     # TODO: This enables us to serialise NameIDs into the stored sessions,
     #       but we don't seem to de-serialise them on load
-    def default(self, o: Any) -> str | Any:
+    def default(self, o: Any) -> str | Any:  # noqa: ANN401
         if isinstance(o, datetime):
             return o.isoformat()
         if isinstance(o, timedelta):

src/eduid/common/models/generic.py

@@ -16,7 +16,7 @@
 # https://docs.pydantic.dev/2.6/concepts/types/#handling-third-party-types
 class ObjectIdPydanticAnnotation:
     @classmethod
-    def __get_pydantic_core_schema__(cls, _source_type: Any, _handler: GetCoreSchemaHandler) -> core_schema.CoreSchema:
+    def __get_pydantic_core_schema__(cls, _source_type: Any, _handler: GetCoreSchemaHandler) -> core_schema.CoreSchema:  # noqa: ANN401
         """
         We return a pydantic_core.CoreSchema that behaves in the following ways:
 
@@ -58,7 +58,7 @@ def __get_pydantic_json_schema__(
 
 class JWKPydanticAnnotation:
     @classmethod
-    def __get_pydantic_core_schema__(cls, _source_type: Any, _handler: GetCoreSchemaHandler) -> core_schema.CoreSchema:
+    def __get_pydantic_core_schema__(cls, _source_type: Any, _handler: GetCoreSchemaHandler) -> core_schema.CoreSchema:  # noqa: ANN401
         """
         We return a pydantic_core.CoreSchema that behaves in the following ways:
 

src/eduid/common/models/scim_base.py

@@ -115,7 +115,7 @@ def is_group(self) -> bool:
         return self.ref is not None and "/Groups/" in self.ref
 
     @classmethod
-    def from_mapping(cls: type[TSubResource], data: Any) -> TSubResource:
+    def from_mapping(cls: type[TSubResource], data: object) -> TSubResource:
         return cls.model_validate(data)
 
 

src/eduid/common/rpc/lookup_mobile_relay.py

@@ -28,7 +28,7 @@ def find_nin_by_mobile(self, mobile_number: str) -> str | None:
             raise LookupMobileTaskFailed(f"find_nin_by_mobile task failed: {e}")
 
     @deprecated("This task seems unused")
-    def find_mobiles_by_nin(self, nin: str) -> Any:
+    def find_mobiles_by_nin(self, nin: str) -> Any:  # noqa: ANN401
         try:
             result = self._find_mobiles_by_NIN.delay(nin)
             result = result.get(timeout=10)  # Lower timeout than standard gunicorn worker timeout (25)

src/eduid/common/testing_base.py

@@ -4,28 +4,29 @@
 import logging.config
 import os
 import uuid
+from collections.abc import Iterable
 from datetime import datetime, timedelta, timezone
 from enum import Enum
 from typing import Any, TypeVar
 
 from bson import ObjectId
 
-from eduid.userdb.testing import MongoTestCase
+from eduid.userdb.testing import MongoTestCase, SetupConfig
 
 logger = logging.getLogger(__name__)
 
 
 class CommonTestCase(MongoTestCase):
     """Base Test case for eduID webapps and workers"""
 
-    def setUp(self, *args: Any, **kwargs: Any) -> None:
+    def setUp(self, config: SetupConfig | None = None) -> None:
         """
         set up tests
         """
         if "EDUID_CONFIG_YAML" not in os.environ:
             os.environ["EDUID_CONFIG_YAML"] = "YAML_CONFIG_NOT_USED"
 
-        super().setUp(*args, **kwargs)
+        super().setUp(config=config)
 
 
 SomeData = TypeVar("SomeData")
@@ -37,7 +38,7 @@ def normalised_data(
     """Utility function for normalising data before comparisons in test cases."""
 
     class NormaliseEncoder(json.JSONEncoder):
-        def default(self, o: Any) -> str | Any:
+        def default(self, o: object) -> Iterable:
             if isinstance(o, datetime):
                 if replace_datetime is not None:
                     return replace_datetime
@@ -66,7 +67,7 @@ class NormaliseDecoder(json.JSONDecoder):
         def __init__(self, *args: Any, **kwargs: Any) -> None:
             super().__init__(object_hook=self.object_hook, *args, **kwargs)
 
-        def object_hook(self, o: Any) -> dict[str, Any]:
+        def object_hook(self, o: dict) -> dict[str, Any]:
             """
             Decode any keys ending in _ts to datetime objects.
 

src/eduid/queue/decorators.py

@@ -25,7 +25,7 @@ def __call__(self, f: Callable[..., Any]) -> Callable[..., Any]:
         if not self.enabled:
             return f
 
-        def audit(*args: Any, **kwargs: Any) -> Any:
+        def audit(*args: Any, **kwargs: Any) -> Any:  # noqa: ANN401
             ret = f(*args, **kwargs)
             if not isclass(ret) and self.collection:  # we can't save class objects in mongodb
                 date = utc_now()
@@ -52,7 +52,7 @@ def disable(cls) -> None:
         cls.enabled = False
 
     @staticmethod
-    def _filter(func: str, data: Any, *args: Any, **kwargs: Any) -> Any:
+    def _filter(func: str, data: object, *args: Any, **kwargs: Any) -> Any:  # noqa: ANN401
         if data is False:
             return data
         if func == "_get_navet_data":

src/eduid/satosa/scimapi/stepup.py

@@ -469,7 +469,7 @@ def _handle_authn_response(self, context: satosa.context.Context, binding: SAMLB
             raise RuntimeError("Unexpected response type")
         return res
 
-    def _metadata_endpoint(self, context: satosa.context.Context, extra: Any) -> CallbackReturnType:
+    def _metadata_endpoint(self, context: satosa.context.Context, extra: object) -> CallbackReturnType:
         metadata_string = create_metadata_string(None, self.sp.config, 4, None, None, None, None, None).decode("utf-8")
         return Response(metadata_string, content="text/xml")
 

src/eduid/scimapi/testing.py

@@ -194,7 +194,7 @@ def _assertScimError(
         schemas: list[str] | None = None,
         status: int = 400,
         scim_type: str | None = None,
-        detail: Any | None = None,
+        detail: object | None = None,
         exclude_keys: list[str] | None = None,
     ) -> None:
         if schemas is None:

src/eduid/scimapi/utils.py

@@ -63,7 +63,7 @@ def load_jwks(config: ScimApiConfig) -> jwk.JWKSet:
 
 def retryable_db_write(func: Callable) -> Callable:
     @functools.wraps(func)
-    def wrapper_run_func(*args: Any, **kwargs: Any) -> Any:
+    def wrapper_run_func(*args: Any, **kwargs: Any) -> Any:  # noqa: ANN401
         max_retries = 10
         retry = 0
         while True:

src/eduid/userdb/admin/__init__.py

@@ -9,7 +9,7 @@
 import time
 from collections.abc import Generator
 from copy import deepcopy
-from typing import Any
+from typing import Any  # noqa: F401
 
 import bson
 import bson.json_util
@@ -51,7 +51,7 @@ def __init__(self, myname: str | None = None, backupbase: str = "/root/raw_db_ch
         self._backupbase: str = backupbase
         self._file_num: int = 0
 
-    def find(self, db: str, collection: str, search_filter: Any) -> Generator[RawData, None, None]:
+    def find(self, db: str, collection: str, search_filter: object) -> Generator[RawData, None, None]:
         """
         Look for documents matching search_filter in the specified database and collection.
 
@@ -72,7 +72,7 @@ def find(self, db: str, collection: str, search_filter: Any) -> Generator[RawDat
             )
             sys.exit(1)
 
-    def save_with_backup(self, raw: RawData, dry_run: bool = True) -> Any:
+    def save_with_backup(self, raw: RawData, dry_run: bool = True) -> str | None:
         """
         Save a mongodb document while trying to carefully make a backup of the document before, after and what changed.
 
@@ -113,7 +113,7 @@ def save_with_backup(self, raw: RawData, dry_run: bool = True) -> Any:
 
         if raw.before == raw.doc:
             sys.stderr.write(f"Document in {db_coll} with id {_id} not changed, aborting save_with_backup\n")
-            return
+            return None
 
         self._file_num = 0
         backup_dir = self._make_backupdir(db_coll, _id)
@@ -132,13 +132,13 @@ def save_with_backup(self, raw: RawData, dry_run: bool = True) -> Any:
         # Write changes.txt after saving, so it will also indicate a successful save
         return self._write_changes(raw, backup_dir, res)
 
-    def _write_changes(self, raw: RawData, backup_dir: str, res: Any) -> Any:
+    def _write_changes(self, raw: RawData, backup_dir: str, res: str) -> str:
         """
         Write a file with one line per change between the before-doc and current doc.
         The format is intended to be easy to grep through.
         """
 
-        def safe_encode(k2: Any, v2: Any) -> str:
+        def safe_encode(k2: object, v2: object) -> str:
             try:
                 return bson.json_util.dumps({k2: v2}, json_options=PYTHON_UUID_LEGACY_JSON_OPTIONS)
             except:

src/eduid/userdb/credentials/external.py

@@ -26,7 +26,7 @@ class ExternalCredential(Credential):
 
     @field_validator("credential_id", mode="before")
     @classmethod
-    def credential_id_objectid(cls, v: Any) -> str:
+    def credential_id_objectid(cls, v: object) -> str:
         """Turn ObjectId into string"""
         if isinstance(v, ObjectId):
             v = str(v)

src/eduid/userdb/credentials/password.py

@@ -1,7 +1,5 @@
 from __future__ import annotations
 
-from typing import Any
-
 from bson import ObjectId
 from pydantic import Field, field_validator
 
@@ -19,7 +17,7 @@ class Password(Credential):
 
     @field_validator("credential_id", mode="before")
     @classmethod
-    def credential_id_objectid(cls, v: Any) -> str:
+    def credential_id_objectid(cls, v: object) -> str:
         """Turn ObjectId into string"""
         if isinstance(v, ObjectId):
             v = str(v)

src/eduid/userdb/db/async_db.py

@@ -157,7 +157,7 @@ async def _drop_whole_collection(self) -> None:
         logger.warning(f"{self!s} Dropping collection {self._coll_name!r}")
         return await self._coll.drop()
 
-    async def _get_document_by_attr(self, attr: str, value: Any) -> Mapping[str, Any] | None:
+    async def _get_document_by_attr(self, attr: str, value: object) -> Mapping[str, Any] | None:
         """
         Return the document in the MongoDB matching field=value
 

src/eduid/userdb/db/sync_db.py

@@ -182,7 +182,7 @@ def _get_all_docs(self) -> pymongo.cursor.Cursor[TUserDbDocument]:
         """
         return self._coll.find({})
 
-    def _get_document_by_attr(self, attr: str, value: Any) -> TUserDbDocument | None:
+    def _get_document_by_attr(self, attr: str, value: Any) -> TUserDbDocument | None:  # noqa: ANN401
         """
         Return the document in the MongoDB matching field=value
 

src/eduid/userdb/element.py

@@ -245,7 +245,7 @@ class PrimaryElement(VerifiedElement, ABC):
 
     is_primary: bool = Field(default=False, alias="primary")  # primary is the old name
 
-    def __setattr__(self, key: str, value: Any) -> None:
+    def __setattr__(self, key: str, value: object) -> None:
         """
         raise PrimaryElementViolation when trying to set a primary element as unverified
         """

src/eduid/userdb/exceptions.py

@@ -2,8 +2,6 @@
 Exceptions thrown by the eduid.userdb database lookup functions.
 """
 
-from typing import Any
-
 
 class EduIDDBError(Exception):
     """
@@ -13,7 +11,7 @@ class EduIDDBError(Exception):
     :type reason: object
     """
 
-    def __init__(self, reason: Any) -> None:
+    def __init__(self, reason: object) -> None:
         Exception.__init__(self)
         self.reason = reason
 

src/eduid/userdb/mail.py

@@ -14,7 +14,7 @@ class MailAddress(PrimaryElement):
 
     @field_validator("email", mode="before")
     @classmethod
-    def validate_email(cls, v: Any) -> str:
+    def validate_email(cls, v: object) -> str:
         if not isinstance(v, str):
             raise ValueError("must be a string")
         return v.lower()

src/eduid/userdb/proofing/element.py

@@ -86,7 +86,7 @@ class EmailProofingElement(ProofingElement):
 
     @field_validator("email", mode="before")
     @classmethod
-    def validate_email(cls, v: Any) -> str:
+    def validate_email(cls, v: object) -> str:
         if not isinstance(v, str):
             raise ValueError("must be a string")
         return v.lower()

src/eduid/userdb/proofing/state.py

@@ -62,7 +62,7 @@ def _default_from_dict(cls: type[TProofingState], data: Mapping[str, Any], field
         return cls(**_data)
 
     @classmethod
-    def from_dict(cls, data: Mapping[str, Any]) -> Any:
+    def from_dict(cls, data: Mapping[str, Any]) -> Any:  # noqa: ANN401
         raise NotImplementedError(f"from_dict not implemented for class {cls.__name__}")
 
     def to_dict(self) -> TUserDbDocument:

src/eduid/userdb/support/models.py

@@ -11,7 +11,7 @@ class GenericFilterDict(dict):
     add_keys: list[str] | None = None
     remove_keys: list[str] | None = None
 
-    def __init__(self, data: Any | None) -> None:
+    def __init__(self, data: dict[str, Any] | None) -> None:
         """
         Create a filtered dict with white- or blacklisting of keys