Add typing to module froide.helper.

froide/helper/api_utils.py

@@ -1,7 +1,9 @@
 import json
 from collections import OrderedDict
+from typing import Dict, List, Union
 
 from django.conf import settings
+from django.db.models.query import QuerySet
 from django.utils.html import format_html
 
 from elasticsearch_dsl.query import Q
@@ -13,11 +15,14 @@
 from rest_framework.reverse import reverse
 from rest_framework.serializers import ListSerializer
 from rest_framework.test import APIRequestFactory
-from rest_framework.utils.serializer_helpers import ReturnDict
+from rest_framework.utils.serializer_helpers import ReturnDict, ReturnList
 from rest_framework_jsonp.renderers import JSONPRenderer
 
+from froide.helper.search.queryset import SearchQuerySetWrapper
+from froide.publicbody.models import PublicBody
 
-def get_fake_api_context(url="/"):
+
+def get_fake_api_context(url: str = "/") -> Dict[str, Request]:
     factory = APIRequestFactory()
     host = settings.SITE_URL.split("/")[-1]
     request = factory.get(url, HTTP_HOST=host)
@@ -29,11 +34,13 @@ def get_fake_api_context(url="/"):
 
 class SearchFacetListSerializer(ListSerializer):
     @property
-    def data(self):
+    def data(self) -> ReturnDict:
         ret = super(ListSerializer, self).data
         return ReturnDict(ret, serializer=self)
 
-    def to_representation(self, instance):
+    def to_representation(
+        self, instance: Union[QuerySet, List[PublicBody]]
+    ) -> OrderedDict:
         ret = super().to_representation(instance)
 
         # Return both objects and facets by wrapping them in dict
@@ -50,7 +57,7 @@ def to_representation(self, instance):
 class CustomLimitOffsetPagination(LimitOffsetPagination):
     max_limit = 50
 
-    def get_paginated_response(self, data):
+    def get_paginated_response(self, data: Union[ReturnDict, ReturnList]) -> Response:
         if "facets" in data:
             # Split out facets into root object
             result = [("objects", data["objects"]), ("facets", data["facets"])]
@@ -78,7 +85,9 @@ def get_paginated_response(self, data):
 
 
 class ElasticLimitOffsetPagination(CustomLimitOffsetPagination):
-    def paginate_queryset(self, queryset, request, view=None):
+    def paginate_queryset(
+        self, queryset: SearchQuerySetWrapper, request: Request, view=None
+    ) -> None:
         self.limit = self.get_limit(request)
         if self.limit is None:
             return None

froide/helper/auth.py

@@ -1,8 +1,19 @@
 from functools import lru_cache
+from typing import Optional, Type, Union
 
 from django.contrib.auth import get_permission_codename
-from django.db.models import Q
+from django.db.models.query import QuerySet
+from django.db.models.query_utils import Q
+from django.http import HttpRequest
+from django.utils.functional import SimpleLazyObject
 
+from rest_framework.request import Request
+
+from froide.account.models import User
+
+# from froide.document.models import Document, DocumentCollection
+from froide.foirequest.models.project import FoiProject
+from froide.foirequest.models.request import FoiRequest
 from froide.team.models import Team
 
 AUTH_MAPPING = {
@@ -11,7 +22,10 @@
 }
 
 
-def check_permission(obj, request, verb):
+Obj = Optional[Union[Type[FoiRequest], FoiRequest]]
+
+
+def check_permission(obj: Obj, request: Union[HttpRequest, Request], verb: str) -> bool:
     user = request.user
     opts = obj._meta
 
@@ -28,7 +42,12 @@ def check_permission(obj, request, verb):
     return False
 
 
-def has_authenticated_access(obj, request, verb="write", scope=None):
+def has_authenticated_access(
+    obj,
+    request: Union[HttpRequest, Request],
+    verb: str = "write",
+    scope: Optional[str] = None,
+) -> bool:
     user = request.user
     if not user.is_authenticated:
         # No authentication, no access
@@ -56,8 +75,13 @@ def has_authenticated_access(obj, request, verb="write", scope=None):
     return False
 
 
+Request_opt = Optional[Union[HttpRequest, Request]]
+
+
 @lru_cache()
-def can_read_object(obj, request=None):
+def can_read_object(
+    obj: Union[FoiProject, FoiRequest], request: Request_opt = None
+) -> bool:
     if hasattr(obj, "is_public") and obj.is_public():
         return True
     if request is None:
@@ -66,27 +90,27 @@ def can_read_object(obj, request=None):
 
 
 @lru_cache()
-def can_read_object_authenticated(obj, request=None):
+def can_read_object_authenticated(obj, request: Request_opt = None) -> bool:
     if request is None:
         return False
     return has_authenticated_access(obj, request, verb="read")
 
 
 @lru_cache()
-def can_write_object(obj, request):
+def can_write_object(obj, request: HttpRequest) -> bool:
     return has_authenticated_access(obj, request)
 
 
 @lru_cache()
-def can_manage_object(obj, request):
+def can_manage_object(obj: FoiRequest, request: HttpRequest) -> bool:
     """
     Team owner permission
     """
     return has_authenticated_access(obj, request, "manage")
 
 
 @lru_cache()
-def can_moderate_object(obj, request):
+def can_moderate_object(obj: FoiRequest, request: Union[HttpRequest, Request]) -> bool:
     return check_permission(obj, request, "moderate")
 
 
@@ -107,14 +131,14 @@ def can_access_object(verb, obj, request):
 
 
 def get_read_queryset(
-    qs,
-    request,
-    has_team=False,
-    public_field=None,
-    public_q=None,
-    scope=None,
-    fk_path=None,
-):
+    qs: QuerySet,
+    request: Union[HttpRequest, Request],
+    has_team: bool = False,
+    public_field: Optional[str] = None,
+    public_q: Optional[Q] = None,
+    scope: Optional[str] = None,
+    fk_path: Optional[str] = None,
+) -> QuerySet:
     user = request.user
     filters = None
     if public_field is not None:
@@ -196,14 +220,22 @@ def get_write_queryset(
     return qs.filter(filters)
 
 
-def make_q(lookup, value, fk_path=None):
+def make_q(
+    lookup: str,
+    value: Union[SimpleLazyObject, User, QuerySet],
+    fk_path: Optional[str] = None,
+) -> Q:
     path = lookup
     if fk_path is not None:
         path = "{}__{}".format(fk_path, lookup)
     return Q(**{path: value})
 
 
-def get_user_filter(request, teams=None, fk_path=None):
+def get_user_filter(
+    request: Union[HttpRequest, Request],
+    teams: Optional[QuerySet] = None,
+    fk_path: Optional[str] = None,
+) -> Q:
     user = request.user
     filter_arg = make_q("user", user, fk_path=fk_path)
     if teams:
@@ -212,6 +244,6 @@ def get_user_filter(request, teams=None, fk_path=None):
     return filter_arg
 
 
-def clear_lru_caches():
+def clear_lru_caches() -> None:
     for f in ACCESS_MAPPING.values():
         f.cache_clear()

froide/helper/cache.py

@@ -1,9 +1,16 @@
+from typing import Callable, Union
+
 from django.contrib.messages import get_messages
+from django.http import HttpRequest
+from django.http.response import HttpResponse
 from django.middleware.cache import CacheMiddleware
 from django.utils.decorators import decorator_from_middleware_with_args
 
+from rest_framework.request import Request
+from rest_framework.response import Response
+
 
-def cache_anonymous_page(time, **cache_kwargs):
+def cache_anonymous_page(time: int, **cache_kwargs) -> Callable:
     """
     Like cache_page decorator, but only for not authenticated users
     """
@@ -20,14 +27,18 @@ def _cache_page(request, *args, **kwargs):
 
 
 class MessageAwareCacheMiddleware(CacheMiddleware):
-    def _should_update_cache(self, request, response):
+    def _should_update_cache(
+        self,
+        request: Union[HttpRequest, Request],
+        response: Union[Response, HttpResponse],
+    ) -> bool:
         should = super(MessageAwareCacheMiddleware, self)._should_update_cache(
             request, response
         )
         return should and not get_messages(request)
 
 
-def cache_page(timeout, cache=None, key_prefix=None):
+def cache_page(timeout: int, cache: None = None, key_prefix: None = None) -> Callable:
     return decorator_from_middleware_with_args(MessageAwareCacheMiddleware)(
         cache_timeout=timeout, cache_alias=cache, key_prefix=key_prefix
     )

froide/helper/content_urls.py

@@ -1,7 +1,10 @@
+from typing import Union
+
 from django.conf import settings
+from django.utils.safestring import SafeString
 
 CONTENT_URLS = settings.FROIDE_CONFIG.get("content_urls", {})
 
 
-def get_content_url(name):
+def get_content_url(name: Union[SafeString, str]) -> str:
     return CONTENT_URLS.get(name, "/")