refactor: fixed logical errors and modernised syntax

Signed-off-by: Vinay Singh <vinay@verid.id>

Author: vinaysingh8866

acapy_agent/database_manager/databases/postgresql_normalized/database.py

@@ -112,8 +112,13 @@ async def _monitor_active_sessions(self):
                                 await session.close()
                             except asyncio.CancelledError:
                                 raise
-                            except Exception:
-                                pass
+                            except Exception as e:
+                                LOGGER.warning(
+                                    "[monitor] Failed to close stale session %s: %s",
+                                    id(session),
+                                    str(e),
+                                    exc_info=True,
+                                )
 
     async def _get_profile_id(self, profile_name: str) -> int:
         conn = await self.pool.getconn()

acapy_agent/database_manager/databases/sqlite_normalized/test/test_sqlite_generic_with_wql.py

@@ -100,7 +100,7 @@ async def run_tests(store, db_path, is_encrypted=True):
             )
         )
         print(f"Scanned not male: {len(scanned_entries_not_male)} entries")
-        assert len(scanned_entries_not_male) == (2 if is_encrypted else 2), (
+        assert len(scanned_entries_not_male) == 2, (
             "Expected 2 not male "
             "(Alice, Charlie if encrypted; Eve, Frank if non-encrypted)"
         )

acapy_agent/database_manager/key.py

@@ -1,6 +1,6 @@
 """Handling of Key instances."""
 
-from typing import Union, Any
+from typing import Any
 from .db_types import KeyAlg, SeedMethod
 
 
@@ -12,39 +12,39 @@ def __init__(self, handle: Any):
         self._handle = handle
 
     @classmethod
-    def generate(cls, alg: Union[str, KeyAlg], *, ephemeral: bool = False) -> "Key":
+    def generate(cls, alg: str | KeyAlg, *, ephemeral: bool = False) -> "Key":
         """Raise an error as key generation requires bindings."""
         raise NotImplementedError("Key generation is not available without bindings.")
 
     @classmethod
     def from_seed(
         cls,
-        alg: Union[str, KeyAlg],
-        seed: Union[str, bytes],
+        alg: str | KeyAlg,
+        seed: str | bytes,
         *,
-        method: Union[str, SeedMethod] = None,
+        method: str | SeedMethod = None,
     ) -> "Key":
         """Raise an error as seed-based key creation requires bindings."""
         raise NotImplementedError(
             "Key creation from seed is not available without bindings."
         )
 
     @classmethod
-    def from_secret_bytes(cls, alg: Union[str, KeyAlg], secret: bytes) -> "Key":
+    def from_secret_bytes(cls, alg: str | KeyAlg, secret: bytes) -> "Key":
         """Raise an error as secret-based key creation requires bindings."""
         raise NotImplementedError(
             "Key creation from secret bytes is not available without bindings."
         )
 
     @classmethod
-    def from_public_bytes(cls, alg: Union[str, KeyAlg], public: bytes) -> "Key":
+    def from_public_bytes(cls, alg: str | KeyAlg, public: bytes) -> "Key":
         """Raise an error as public-based key creation requires bindings."""
         raise NotImplementedError(
             "Key creation from public bytes is not available without bindings."
         )
 
     @classmethod
-    def from_jwk(cls, jwk: Union[dict, str, bytes]) -> "Key":
+    def from_jwk(cls, jwk: dict | str | bytes) -> "Key":
         """Raise an error as JWK-based key creation requires bindings."""
         raise NotImplementedError(
             "Key creation from JWK is not available without bindings."
@@ -65,11 +65,11 @@ def ephemeral(self) -> bool:
         """Return a placeholder ephemeral flag since bindings is unavailable."""
         return False  # Placeholder value
 
-    def convert_key(self, alg: Union[str, KeyAlg]) -> "Key":
+    def convert_key(self, alg: str | KeyAlg) -> "Key":
         """Raise an error as key conversion requires bindings."""
         raise NotImplementedError("Key conversion is not available without bindings.")
 
-    def key_exchange(self, alg: Union[str, KeyAlg], pk: "Key") -> "Key":
+    def key_exchange(self, alg: str | KeyAlg, pk: "Key") -> "Key":
         """Raise an error as key exchange requires bindings."""
         raise NotImplementedError("Key exchange is not available without bindings.")
 
@@ -81,15 +81,15 @@ def get_secret_bytes(self) -> bytes:
         """Return placeholder secret bytes since bindings is unavailable."""
         return b"secret_bytes_placeholder"
 
-    def get_jwk_public(self, alg: Union[str, KeyAlg] = None) -> str:
+    def get_jwk_public(self, alg: str | KeyAlg = None) -> str:
         """Return placeholder public JWK since bindings is unavailable."""
         return "jwk_public_placeholder"
 
     def get_jwk_secret(self) -> bytes:
         """Return placeholder secret JWK since bindings is unavailable."""
         return b"jwk_secret_placeholder"
 
-    def get_jwk_thumbprint(self, alg: Union[str, KeyAlg] = None) -> str:
+    def get_jwk_thumbprint(self, alg: str | KeyAlg = None) -> str:
         """Return placeholder JWK thumbprint since bindings is unavailable."""
         return "jwk_thumbprint_placeholder"
 
@@ -102,7 +102,7 @@ def aead_random_nonce(self) -> bytes:
         return b"nonce_placeholder"
 
     def aead_encrypt(
-        self, message: Union[str, bytes], *, nonce: bytes = None, aad: bytes = None
+        self, message: str | bytes, *, nonce: bytes = None, aad: bytes = None
     ) -> str:
         """Return a placeholder for encrypted data."""
         return "Encrypted placeholder"
@@ -118,12 +118,12 @@ def aead_decrypt(
         """Return placeholder decrypted data."""
         return b"decrypted placeholder"
 
-    def sign_message(self, message: Union[str, bytes], sig_type: str = None) -> bytes:
+    def sign_message(self, message: str | bytes, sig_type: str = None) -> bytes:
         """Raise an error as signing requires bindings."""
         raise NotImplementedError("Message signing is not available without bindings.")
 
     def verify_signature(
-        self, message: Union[str, bytes], signature: bytes, sig_type: str = None
+        self, message: str | bytes, signature: bytes, sig_type: str = None
     ) -> bool:
         """Raise an error as verification requires bindings."""
         raise NotImplementedError(
@@ -136,7 +136,7 @@ def wrap_key(self, other: "Key", *, nonce: bytes = None) -> str:
 
     def unwrap_key(
         self,
-        alg: Union[str, KeyAlg],
+        alg: str | KeyAlg,
         ciphertext: bytes,
         *,
         nonce: bytes = None,

acapy_agent/database_manager/wql_normalized/encoders/sqlite_encoder.py

@@ -1,6 +1,6 @@
 """Module docstring."""
 
-from typing import List
+from typing import List, Tuple, cast
 from ..tags import TagQueryEncoder, TagName, CompareOp, ConjunctionOp, TagQuery
 import logging
 
@@ -101,9 +101,11 @@ def _encode_not(self, query: TagQuery) -> str:
         """Encode a NOT expression with special-cases for certain variants."""
         inner = query.data
         if inner.variant == "Exist":
-            return self.encode_exist(inner.data, True)
+            names = cast(List[TagName], inner.data)
+            return self.encode_exist(names, negate=True)
         if inner.variant == "In":
-            return self.encode_in(*inner.data, True)
+            name, values = cast(Tuple[TagName, List[str]], inner.data)
+            return self.encode_in(name, values, negate=True)
         if not self.normalized and inner.variant in [
             "Eq",
             "Neq",
@@ -113,7 +115,10 @@ def _encode_not(self, query: TagQuery) -> str:
             "Lte",
             "Like",
         ]:
-            return self.encode_op(getattr(CompareOp, inner.variant), *inner.data, True)
+            name, value = cast(Tuple[TagName, str], inner.data)
+            return self.encode_op(
+                getattr(CompareOp, inner.variant), name, value, negate=True
+            )
         subquery = self.encode_query(inner, False, top_level=False)
         if inner.variant in ["And", "Or"]:
             return f"NOT {subquery}"