Add/fix type annotation

lib/rucio/api/did.py

@@ -28,11 +28,9 @@
 from rucio.db.sqla.session import read_session, stream_session, transactional_session
 
 if TYPE_CHECKING:
-    from typing import Any, Dict, Optional, List
+    from typing import Any, Optional, Iterator
     from sqlalchemy.orm import Session
 
-    DIDListType = List[Dict]
-
 
 @stream_session
 def list_dids(scope, filters, did_type='collection', ignore_case=False, limit=None, offset=None, long=False, recursive=False, vo='def', *, session: "Session"):
@@ -330,11 +328,11 @@ def list_content_history(scope, name, vo='def', *, session: "Session"):
 
 
 @stream_session
-def bulk_list_files(dids: "DIDListType", long: bool = False, vo: str = 'def', *, session: "Session"):
+def bulk_list_files(dids: list[dict[str, Any]], long: bool = False, vo: str = 'def', *, session: "Session") -> Iterator[dict[str, Any]]:
     """
-    List file contents of a list of data identifier.
+    List file contents of a list of data identifiers.
 
-    :param dids:       A list of dids.
+    :param dids:       A list of DIDs.
     :param long:       A boolean to choose if more metadata are returned or not.
     :param vo:         The VO to act on.
     :param session:    The database session in use.

lib/rucio/client/didclient.py

@@ -15,6 +15,7 @@
 
 from datetime import datetime
 from json import dumps
+from typing import Optional, Any
 from urllib.parse import quote_plus
 
 from requests.status_codes import codes
@@ -342,7 +343,7 @@ def list_content_history(self, scope, name):
         exc_cls, exc_msg = self._get_exception(headers=r.headers, status_code=r.status_code, data=r.content)
         raise exc_cls(exc_msg)
 
-    def list_files(self, scope, name, long=None):
+    def list_files(self, scope: str, name: str, long: Optional[bool] = None):
         """
         List data identifier file contents.
 
@@ -364,11 +365,11 @@ def list_files(self, scope, name, long=None):
             exc_cls, exc_msg = self._get_exception(headers=r.headers, status_code=r.status_code, data=r.content)
             raise exc_cls(exc_msg)
 
-    def bulk_list_files(self, dids):
+    def bulk_list_files(self, dids: list[dict[str, Any]]):
         """
         List data identifier file contents.
 
-        :param dids: The list of dids.
+        :param dids: The list of DIDs.
         """
 
         data = {'dids': dids}

lib/rucio/core/did.py

@@ -42,14 +42,12 @@
 from rucio.db.sqla.util import temp_table_mngr
 
 if TYPE_CHECKING:
-    from collections.abc import Callable, Sequence
-    from typing import Any, Dict, Optional, Union, List
+    from collections.abc import Iterator, Sequence
+    from typing import Any, Optional, Union
     from sqlalchemy.orm import Session
     from sqlalchemy.schema import Table
-    from rucio.common.types import InternalAccount, InternalScope
+    from rucio.common.types import InternalAccount, InternalScope, LoggerFunction
 
-    LoggerFunction = Callable[..., Any]
-    DIDListType = List[Dict]
 
 METRICS = MetricManager(module=__name__)
 
@@ -1889,26 +1887,26 @@ def list_child_datasets(
 
 
 @stream_session
-def bulk_list_files(dids: "DIDListType", long: bool = False, *, session: "Session"):
+def bulk_list_files(dids: list[dict[str, Any]], long: bool = False, *, session: "Session") -> Optional[Iterator[dict[str, Any]]]:
     """
     List file contents of a list of data identifier.
 
-    :param dids:       A list of dids.
+    :param dids:       A list of DIDs.
     :param long:       A boolean to choose if more metadata are returned or not.
     :param session:    The database session in use.
     """
     for did in dids:
         try:
-            for file_ in list_files(scope=did['scope'], name=did['name'], long=long, session=session):
-                file_['parent_scope'] = did['scope']
-                file_['parent_name'] = did['name']
-                yield file_
+            for file_dict in list_files(scope=did['scope'], name=did['name'], long=long, session=session):
+                file_dict['parent_scope'] = did['scope']
+                file_dict['parent_name'] = did['name']
+                yield file_dict
         except exception.DataIdentifierNotFound:
             pass
 
 
 @stream_session
-def list_files(scope: "InternalScope", name: str, long: bool = False, *, session: "Session"):
+def list_files(scope: InternalScope, name: str, long: bool = False, *, session: "Session") -> Iterator[dict[str, Any]]:
     """
     List data identifier file contents.