AAP-68990 Switch from typeguard to mypy

.github/workflows/linting.yml

@@ -25,6 +25,8 @@ jobs:
             command: check_black
           - name: api-isort
             command: check_isort
+          - name: api-mypy
+            command: check_mypy
           - name: pure-python-imports
             command: check_pure_python_imports
     steps:
@@ -40,9 +42,17 @@ jobs:
         with:
           python-version: 3.12
 
+      - name: Install build requirements for mypy
+        if: matrix.tests.name == 'api-mypy'
+        run: sudo apt-get update && sudo apt-get install -y libsasl2-dev libldap2-dev libssl-dev libxmlsec1-dev
+
       - name: Install requirments
         run: pip3.12 install -r requirements/requirements_dev.txt
 
+      - name: Install all dependencies for mypy
+        if: matrix.tests.name == 'api-mypy'
+        run: pip3.12 install -r requirements/requirements_all.txt
+
       - name: Install test dependencies for pure-python-imports
         if: matrix.tests.name == 'pure-python-imports'
         run: |

Makefile

@@ -11,7 +11,7 @@ COMPOSE_UP_OPTS ?=
 DOCKER_COMPOSE ?= docker compose
 
 .PHONY: PYTHON_VERSION clean build\
-	check lint check_black check_flake8 check_isort git_hooks_config
+	check lint check_black check_flake8 check_isort check_mypy git_hooks_config
 
 PYTHON_VERSION:
 	@echo "$(subst python,,$(PYTHON))"
@@ -49,6 +49,10 @@ check_flake8:
 check_isort:
 	tox -e isort -- --check $(CHECK_SYNTAX_FILES)
 
+## Run mypy type checking
+check_mypy:
+	tox -e mypy
+
 
 ## Starts a postgres container in the background if one is not running
 # Options:

ansible_base/activitystream/serializers.py

@@ -32,10 +32,12 @@ class Meta:
     def get_content_type_model(self, obj) -> Optional[str]:
         if obj.content_type:
             return obj.content_type.model
+        return None
 
     def get_related_content_type_model(self, obj) -> Optional[str]:
         if obj.related_content_type:
             return obj.related_content_type.model
+        return None
 
     def _field_value_to_python(self, entry, field_name, value):
         model = entry.content_type.model_class()

ansible_base/authentication/authenticator_plugins/ldap.py

@@ -466,7 +466,7 @@ def __init__(self, prefix: str = 'AUTH_LDAP_', defaults: dict = {}):
 
         # Group type needs to be an object instead of a String so instantiate it
         group_type_class = find_class_in_modules(defaults["GROUP_TYPE"])
-        setattr(self, 'GROUP_TYPE', group_type_class(**defaults['GROUP_TYPE_PARAMS']))
+        setattr(self, 'GROUP_TYPE', group_type_class(**defaults['GROUP_TYPE_PARAMS']))  # type: ignore[operator]
 
 
 class AuthenticatorPlugin(LDAPBackend, AbstractAuthenticatorPlugin):
@@ -482,9 +482,9 @@ def __init__(self, database_instance=None, *args, **kwargs):
         self.configuration_encrypted_fields = ['BIND_PASSWORD']
         self.set_logger(logger)
 
-    def authenticate(self, request, username=None, password=None, **kwargs) -> (object, dict, list):
+    def authenticate(self, request, username=None, password=None, **kwargs):
         if not username or not password:
-            return
+            return None
         users_groups = []
 
         if not self.database_instance:

ansible_base/authentication/models/authenticator.py

@@ -13,7 +13,7 @@ def get_next_authenticator_order():
     return largest_order_authenticator['order'] + 1 if largest_order_authenticator else 1
 
 
-class Authenticator(UniqueNamedCommonModel):
+class Authenticator(UniqueNamedCommonModel):  # type: ignore[django-manager-missing]
     ignore_relations = ['authenticator_users']
     enabled = fields.BooleanField(default=False, help_text="Should this authenticator be enabled.")
     create_objects = fields.BooleanField(default=True, help_text="Allow authenticator to create objects (users, teams, organizations).")

ansible_base/authentication/models/authenticator_map.py

@@ -14,11 +14,11 @@ class Meta:
         constraints = [
             models.CheckConstraint(
                 name="%(app_label)s_%(class)s_require_org_team_if_team_map",
-                check=(~models.Q(map_type='team') | models.Q(team__isnull=False) & models.Q(organization__isnull=False)),
+                check=(~models.Q(map_type='team') | models.Q(team__isnull=False) & models.Q(organization__isnull=False)),  # type: ignore[call-arg]
             ),
             models.CheckConstraint(
                 name="%(app_label)s_%(class)s_require_org_if_org_map",
-                check=(~models.Q(map_type='organization') | models.Q(organization__isnull=False)),
+                check=(~models.Q(map_type='organization') | models.Q(organization__isnull=False)),  # type: ignore[call-arg]
             ),
         ]
         unique_together = ['name', 'authenticator']

ansible_base/authentication/models/authenticator_user.py

@@ -23,7 +23,7 @@ def b64_encode_binary_data_in_dict(obj: Any) -> Any:
     return obj
 
 
-class AuthenticatorUser(AbstractUserSocialAuth, AbstractCommonModel):
+class AuthenticatorUser(AbstractUserSocialAuth, AbstractCommonModel):  # type: ignore[django-manager-missing]
     """
     This appends extra information on the local user model that includes extra data returned by
     the authenticators and links the user to the authenticator that they used to login.

ansible_base/authentication/utils/authentication.py

@@ -71,7 +71,7 @@ def migrate_from_existing_authenticator(
     )
 
     if len(migrate_users) == 0:
-        return
+        return None
 
     try:
         main_user = AuthenticatorUser.objects.get(uid=uid, provider=authenticator).user

ansible_base/authentication/utils/authenticator_map.py

@@ -37,6 +37,7 @@ def check_expansion_syntax(value: Optional[str]) -> Optional[TranslatedString]:
 
     if has_expansion(value) and not _EXPANSION_RE.search(value):
         return _("Expansion only supports the format {% for_attr_value(attribute) %}")
+    return None
 
 
 def expand_syntax(attributes: dict, auth_map: AuthenticatorMap) -> list[Dict[str, str]]:
@@ -191,7 +192,7 @@ def check_role_type(map_type: Optional[str], role: Optional[str], org: Optional[
 
     if not _is_rbac_installed():
         errors['role'] = _("You specified a role without RBAC installed ")
-        return errors
+        return errors  # type: ignore[return-value]
 
     from ansible_base.rbac.models import RoleDefinition
 
@@ -201,7 +202,7 @@ def check_role_type(map_type: Optional[str], role: Optional[str], org: Optional[
 
         # system role is allowed for map type == role without further conditions
         if is_system_role and map_type == 'role':
-            return errors
+            return errors  # type: ignore[return-value]
 
         is_org_role, is_team_role = False, False
         if not is_system_role:
@@ -227,4 +228,4 @@ def check_role_type(map_type: Optional[str], role: Optional[str], org: Optional[
     except ObjectDoesNotExist:
         errors['role'] = _("RoleDefinition {role} doesn't exist").format(role=role)
 
-    return errors
+    return errors  # type: ignore[return-value]

ansible_base/authentication/utils/claims.py

@@ -240,7 +240,7 @@ def process_groups(trigger_condition: dict, groups: list, map_id: int, tracking_
         groups = [f"{group}".casefold() for group in groups]
         trigger_condition = _lowercase_group_triggers(trigger_condition)
 
-    invalid_conditions = set(trigger_condition.keys()) - set(TRIGGER_DEFINITION['groups']['keys'].keys())
+    invalid_conditions = set(trigger_condition.keys()) - set(TRIGGER_DEFINITION['groups']['keys'].keys())  # type: ignore[index]
     if invalid_conditions:
         logger.warning(f"[{tracking_id}] The conditions {', '.join(invalid_conditions)} for groups in mapping {map_id} are invalid and won't be processed")
 
@@ -283,6 +283,7 @@ def has_access_with_join(current_access: Optional[bool], new_access: bool, condi
 
     if condition == 'and':
         return current_access and new_access
+    return None
 
 
 def _lowercase_value(value: Any) -> Any:
@@ -353,7 +354,7 @@ def _validate_join_condition(join_condition, map_id: int, tracking_id: str) -> s
     Returns:
         Valid join condition ('or' or 'and')
     """
-    if join_condition not in TRIGGER_DEFINITION['attributes']['keys']['join_condition']['choices']:
+    if join_condition not in TRIGGER_DEFINITION['attributes']['keys']['join_condition']['choices']:  # type: ignore[index]
         logger.warning(f"[{tracking_id}] Trigger join_condition {join_condition} on authenticator map {map_id} is invalid and will be set to 'or'")
         return 'or'
     return join_condition
@@ -373,7 +374,7 @@ def _validate_attribute_conditions(attribute: str, condition: dict, map_id: int,
         True if conditions are valid and should be processed, False if should be skipped
     """
     # Warn if there are any invalid conditions, we are just going to ignore them
-    invalid_conditions = set(condition.keys()) - set(TRIGGER_DEFINITION['attributes']['keys']['*']['keys'].keys())
+    invalid_conditions = set(condition.keys()) - set(TRIGGER_DEFINITION['attributes']['keys']['*']['keys'].keys())  # type: ignore[index]
     if invalid_conditions:
         logger.warning(
             f"[{tracking_id}] The conditions {', '.join(invalid_conditions)} for attribute {attribute} "
@@ -567,7 +568,7 @@ def _process_user_value(
         user_str = f"{a_user_value}".casefold() if _is_case_insensitivity_enabled() else f"{a_user_value}"
 
         # Evaluate condition
-        result = evaluate_fn(user_str, trigger_value)
+        result = evaluate_fn(user_str, trigger_value)  # type: ignore[operator]
         has_access = has_access_with_join(has_access, result, join_condition)
 
         # Log result

ansible_base/jwt_consumer/common/util.py

@@ -61,7 +61,7 @@ def validate_x_trusted_proxy_header(header_value: str, ignore_cache=False) -> bo
             logger.warning(f"Timestamp {timestamp} was too old by {header_age_ms}ms to be valid-alter trusted_header_timeout if needed")
             return False
     except ValueError:
-        logger.warning(f"Unable to convert timestamp (base64) {b64encode(timestamp.encode('UTF-8'))} into an integer")
+        logger.warning(f"Unable to convert timestamp (base64) {b64encode(timestamp.encode('UTF-8')).decode()} into an integer")
         return False
 
     try:
@@ -71,7 +71,7 @@ def validate_x_trusted_proxy_header(header_value: str, ignore_cache=False) -> bo
         return False
 
     try:
-        public_key.verify(
+        public_key.verify(  # type: ignore[call-arg]
             signature_bytes,
             bytes(f'{_SHARED_SECRET}-{timestamp}', 'utf-8'),
             padding.PSS(mgf=padding.MGF1(hashes.SHA256()), salt_length=padding.PSS.MAX_LENGTH),

ansible_base/lib/channels/middleware.py

@@ -17,7 +17,7 @@
 def _get_authenticated_user(scope: dict):
     request = HttpRequest()
     request.META = {_http_key(k.decode()): v.decode() for (k, v) in scope["headers"]}
-    auth_classes = [auth() for auth in api_settings.DEFAULT_AUTHENTICATION_CLASSES]
+    auth_classes = [auth() for auth in api_settings.DEFAULT_AUTHENTICATION_CLASSES]  # type: ignore[operator]
     try:
         return Request(request, authenticators=auth_classes).user
     except Exception:

ansible_base/lib/redis/client.py

@@ -40,7 +40,7 @@ class DABRedisRespectACLFlushMixin:
     At some point in the future we may also want to override flushall as well
     """
 
-    def flushdb(self, asynchronous: Optional[bool] = None, **kwargs) -> ResponseT:
+    def flushdb(self, asynchronous: Optional[bool] = None, **kwargs) -> ResponseT:  # type: ignore[return]
         if asynchronous is not None:
             logger.warning("DABRedis clients implement an ACL friendly FLUSHDB which can not be async")
 
@@ -106,6 +106,11 @@ def connect(self, index: int = 0) -> Union[DABRedis, DABRedisCluster]:
 
 
 class RedisClientGetter:
+    url: str
+    connection_settings: dict
+    mode: str
+    redis_hosts: str | None
+
     def _redis_parse_url(self) -> None:
         if self.url == '':
             # If there is no URL we have nothing to do
@@ -190,7 +195,7 @@ def _get_hosts(self) -> None:
             raise ImproperlyConfigured(_('Unable to parse redis_hosts, see logs for more details'))
 
     def __init__(self, *args, **kwargs):
-        self.url = ''
+        self.url: str = ''
 
     def get_client(self, url: str = '', **kwargs) -> Union[DABRedis, DABRedisCluster]:
         # remove our settings which are invalid to the parent classes
@@ -265,7 +270,7 @@ def get_redis_status(url: str = '', timeout: int = _DEFAULT_STATUS_TIMEOUT_SEC,
             response['status'] = determine_cluster_node_status(response['cluster_nodes'])
             # Now our status should be STATUS_GOOD or STATUS_DEGRADED
             # There is one more check we need to do and that is if the cluster_info did not return ok we are in a bad state
-            if response['cluster_info']['cluster_state'] != _REDIS_CLUSTER_OK_STATUS:
+            if response['cluster_info']['cluster_state'] != _REDIS_CLUSTER_OK_STATUS:  # type: ignore[index]
                 response['status'] = STATUS_FAILED
     except Exception as e:
         response['status'] = STATUS_FAILED

ansible_base/lib/routers/association_resource_router.py

@@ -271,7 +271,7 @@ def __dir__(self):
         for method in exclude_list:
             setattr(AssociatedViewSetType, method, attribute_raiser)
 
-        class AssociatedViewSet(viewset, AssociationViewSetMethodsMixin, metaclass=AssociatedViewSetType):
+        class AssociatedViewSet(viewset, AssociationViewSetMethodsMixin, metaclass=AssociatedViewSetType):  # type: ignore[valid-type]
             """Adjusted version of given viewset for related endpoint with only list views"""
 
             pass

ansible_base/lib/utils/create_system_user.py

@@ -15,7 +15,7 @@
 """
 
 
-def create_system_user(user_model: Type[models.Model]) -> models.Model:
+def create_system_user(user_model: Type[models.Model]) -> Optional[models.Model]:
     # Note: We can't load models here so we can typecast to anything better than Model
     from ansible_base.lib.abstract_models.user import AbstractDABUser
 

ansible_base/lib/utils/db.py

@@ -223,7 +223,7 @@ def combine_settings_dict(settings_dict1: dj_db_dict, settings_dict2: dj_db_dict
     settings_dict['OPTIONS'].update(extra_options)
     # Apply overrides from nested OPTIONS for the listener connection
     for k, v in settings_dict2.get('OPTIONS', {}).items():
-        settings_dict['OPTIONS'][k] = v
+        settings_dict['OPTIONS'][k] = v  # type: ignore[index]
 
     return settings_dict
 

ansible_base/oauth2_provider/utils.py

@@ -7,7 +7,7 @@
 User = get_user_model()
 
 
-def is_external_account(user: User) -> Optional[Authenticator]:
+def is_external_account(user: "User") -> Optional[Authenticator]:  # type: ignore[valid-type]
     """
     Determines whether the user is associated with any external
     login source. If they are, return the source. Otherwise, None.

ansible_base/rbac/claims.py

@@ -237,7 +237,7 @@ def get_user_claims(user: Model) -> dict[str, Union[list[str], dict[str, Union[s
     global_roles = _get_user_global_roles(user)
 
     # Build final claims structure
-    return {'objects': object_arrays, 'object_roles': object_roles, 'global_roles': global_roles}
+    return {'objects': object_arrays, 'object_roles': object_roles, 'global_roles': global_roles}  # type: ignore[dict-item]
 
 
 # ---- for claims saving ----
@@ -413,9 +413,9 @@ def get_user_claims_hashable_form(claims: dict) -> dict[str, Union[list[str], di
                     ansible_ids.append(ansible_id)
 
         # Sort ansible_ids for deterministic ordering
-        hashable_claims['object_roles'][role_name] = sorted(ansible_ids)
+        hashable_claims['object_roles'][role_name] = sorted(ansible_ids)  # type: ignore[index]
 
-    return hashable_claims
+    return hashable_claims  # type: ignore[return-value]
 
 
 def get_claims_hash(hashable_claims: dict) -> str:

ansible_base/rbac/managed.py

@@ -33,7 +33,7 @@ def get_permissions(self, apps) -> set[str]:
         return self.permission_list
 
     def get_translated_name(self) -> str:
-        return _(self.name)
+        return _(self.name)  # type: ignore[return-value]
 
     def get_content_type(self, apps):
         model = self.get_model(apps)

ansible_base/rbac/models/content_type.py

@@ -40,7 +40,7 @@ def _add_to_cache(self, using: str, ct: django_models.Model) -> None:
     def _get_from_cache(self, opts: Options, service: str) -> django_models.Model:
         """Return a cached ``DABContentType`` for ``opts`` and ``service``."""
         key = (service, opts.app_label, opts.model_name)
-        return self._cache[self.db][key]
+        return self._cache[self.db][key]  # type: ignore[index]
 
     def _get_opts(self, model: Union[Type[django_models.Model], django_models.Model], for_concrete_model: bool) -> Options:
         """Return the ``Options`` object for ``model``."""
@@ -109,9 +109,9 @@ def get_for_models(
                 ct = self._get_from_cache(opts, model_service)
             except KeyError:
                 needed_models[(model_service, opts.app_label)].add(opts.model_name)
-                needed_opts[(model_service, opts.app_label, opts.model_name)].append(model)
+                needed_opts[(model_service, opts.app_label, opts.model_name)].append(model)  # type: ignore[index]
             else:
-                results[model] = ct
+                results[model] = ct  # type: ignore[index]
 
         if needed_opts:
             condition = django_models.Q(

ansible_base/rbac/models/role.py

@@ -80,12 +80,12 @@ def contribute_to_class(self, cls: Type[models.Model], name: str) -> None:
 
     def give_creator_permissions(self, user, obj) -> Optional['RoleUserAssignment']:
         if not permission_registry.is_registered(obj):
-            return  # Exit before getting content type, which will not exist
+            return None  # Exit before getting content type, which will not exist
 
         # If the user is a superuser, no need to bother giving the creator permissions
         for super_flag in settings.ANSIBLE_BASE_BYPASS_SUPERUSER_FLAGS:
             if getattr(user, super_flag):
-                return
+                return None
 
         needed_actions = settings.ANSIBLE_BASE_CREATOR_DEFAULTS
 
@@ -120,6 +120,7 @@ def give_creator_permissions(self, user, obj) -> Optional['RoleUserAssignment']:
             # reverse sync the assignment
             maybe_reverse_sync_assignment(assignment)
             return assignment
+        return None
 
     def get_or_create(self, permissions=(), defaults=None, **kwargs):
         """Override get_or_create to support lookup by permissions list.

ansible_base/rbac/permission_registry.py

@@ -27,7 +27,7 @@
 
 class PermissionRegistry:
     def __init__(self):
-        self._registry: set[Model] = set()  # model registry
+        self._registry: set[type[Model]] = set()  # model registry
         self._name_to_model = dict()
         self._parent_fields = dict()
         self._managed_roles = dict()  # code-defined role definitions, managed=True
@@ -92,6 +92,7 @@ def get_managed_role_constructor_by_name(self, name: str) -> Optional[ManagedRol
         for managed_role in self._managed_roles.values():
             if managed_role.name == name:
                 return managed_role
+        return None
 
     def register_managed_role_constructor(self, shortname: str, managed_role: ManagedRoleConstructor) -> None:
         """Add the given managed role to the managed role registry"""