Typing: Add type annotations to vo.py

lib/rucio/core/vo.py

@@ -14,7 +14,7 @@
 # limitations under the License.
 
 import re
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 from sqlalchemy.exc import DatabaseError, IntegrityError
 from sqlalchemy.orm.exc import NoResultFound
@@ -34,7 +34,7 @@
 
 
 @read_session
-def vo_exists(vo, *, session: "Session"):
+def vo_exists(vo: str, *, session: "Session") -> bool:
     """
     Verify that the vo exists.
 
@@ -47,7 +47,7 @@ def vo_exists(vo, *, session: "Session"):
 
 
 @transactional_session
-def add_vo(vo, description, email, *, session: "Session"):
+def add_vo(vo: str, description: str, email: str, *, session: "Session") -> None:
     """
     Add a VO and setup a new root user.
     New root user will have account name 'root' and a userpass identity with username: 'root@<vo>' and password: 'password'
@@ -89,7 +89,7 @@ def add_vo(vo, description, email, *, session: "Session"):
 
 
 @read_session
-def list_vos(*, session: "Session"):
+def list_vos(*, session: "Session") -> list[dict[str, Any]]:
     """
     List all the VOs in the db.
 
@@ -114,7 +114,7 @@ def list_vos(*, session: "Session"):
 
 
 @transactional_session
-def update_vo(vo, parameters, *, session: "Session"):
+def update_vo(vo: str, parameters: dict[str, Any], *, session: "Session") -> None:
     """
     Update VO properties (email, description).
 
@@ -136,7 +136,7 @@ def update_vo(vo, parameters, *, session: "Session"):
     query.update(param)
 
 
-def map_vo(vo):
+def map_vo(vo: str) -> str:
     """
     Converts a long VO name into the internal short (three letter)
     tag mapping.

lib/rucio/daemons/automatix/automatix.py

@@ -166,7 +166,7 @@ def run_once(heartbeat_handler: "HeartbeatHandler", inputfile: str, **_kwargs) -
     account = config_get("automatix", "account", raise_exception=False, default="root")
     scope = config_get("automatix", "scope", raise_exception=False, default="test")
     client = Client(account=account)
-    vo = map_vo(client.vo)
+    vo = map_vo(client.vo)  # type: ignore
     filters = {"scope": InternalScope("*", vo=vo)}
     scopes = list_scopes(filter_=filters)
     if InternalScope(scope, vo=vo) not in scopes: