diff --git a/lib/rucio/core/importer.py b/lib/rucio/core/importer.py index ced3ed7a66..53c772b448 100644 --- a/lib/rucio/core/importer.py +++ b/lib/rucio/core/importer.py @@ -13,7 +13,8 @@ # See the License for the specific language governing permissions and # limitations under the License. -from typing import TYPE_CHECKING +from collections.abc import Iterable +from typing import TYPE_CHECKING, Any from rucio.common.config import config_get from rucio.common.exception import RSEOperationNotSupported @@ -28,7 +29,7 @@ @transactional_session -def import_rses(rses, rse_sync_method='edit', attr_sync_method='edit', protocol_sync_method='edit', vo='def', *, session: "Session"): +def import_rses(rses: dict[str, dict[str, Any]], rse_sync_method: str = 'edit', attr_sync_method: str = 'edit', protocol_sync_method: str = 'edit', vo: str = 'def', *, session: "Session") -> None: new_rses = [] for rse_name in rses: rse = rses[rse_name] @@ -136,7 +137,7 @@ def import_rses(rses, rse_sync_method='edit', attr_sync_method='edit', protocol_ @transactional_session -def import_distances(distances, vo='def', *, session: "Session"): +def import_distances(distances, vo: str = 'def', *, session: "Session") -> None: for src_rse_name in distances: src = rse_module.get_rse_id(rse=src_rse_name, vo=vo, session=session) for dest_rse_name in distances[src_rse_name]: @@ -156,7 +157,7 @@ def import_distances(distances, vo='def', *, session: "Session"): @transactional_session -def import_identities(identities, account_name, old_identities, old_identity_account, account_email, *, session: "Session"): +def import_identities(identities: Iterable[dict[str, Any]], account_name: str, old_identities: Iterable[tuple], old_identity_account: tuple[str, str, str], account_email: str, *, session: "Session") -> None: for identity in identities: identity['type'] = IdentityType[identity['type'].upper()] @@ -185,7 +186,7 @@ def import_identities(identities, account_name, old_identities, old_identity_acc @transactional_session -def import_accounts(accounts, vo='def', *, session: "Session"): +def import_accounts(accounts: Iterable[dict[str, Any]], vo: str = 'def', *, session: "Session") -> None: vo_filter = {'account': InternalAccount(account='*', vo=vo)} old_accounts = {account['account']: account for account in account_module.list_accounts(filter_=vo_filter, session=session)} missing_accounts = [account for account in accounts if account['account'] not in old_accounts] @@ -222,7 +223,7 @@ def import_accounts(accounts, vo='def', *, session: "Session"): @transactional_session -def import_data(data, vo='def', *, session: "Session"): +def import_data(data: dict[str, Any], vo: str = 'def', *, session: "Session") -> None: """ Import data to add and update records in Rucio.