Switch ordering of classes to make types Python 3.8 compatible

Author: Grimpy101

flask_phpbb3/backends/base.py

@@ -1,83 +1,12 @@
 import abc
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, Union
 
 import cachelib
 
 
 ACL_OPTIONS_CACHE_TTL: int = 3600 * 1
 
 
-class BaseBackend:
-    KNOWN_OPERATIONS = (
-        'fetch',
-        'get',
-        'has',
-        'set',
-    )
-    KNOWN_DRIVERS = (
-        'psycopg2',
-    )
-
-    def __init__(
-        self,
-        cache: cachelib.BaseCache,
-        config: Dict[str, Any]
-    ):
-        self._functions: Dict[str, str] = {}
-        self._connection = None
-        self._cache = cache
-        self._config = config
-
-        self._prepare_statements()
-        custom_statements = self._config.get('CUSTOM_STATEMENTS', {})
-        if not isinstance(custom_statements, dict):
-            custom_statements = {}
-        self._functions.update(custom_statements)
-
-    @abc.abstractmethod
-    def _setup_connection(self) -> None:
-        raise NotImplementedError
-
-    @abc.abstractmethod
-    def _prepare_statements(self) -> None:
-        raise NotImplementedError
-
-    @property
-    @abc.abstractmethod
-    def _db(self) -> Any:
-        raise NotImplementedError
-
-    @abc.abstractmethod
-    def execute(
-        self,
-        command: str,
-        cache: bool = False,
-        cache_ttl: Optional[int] = None,
-        skip: int = 0,
-        limit: Optional[int] = 10,
-        **kwargs: int | str
-    ) -> Any:
-        raise NotImplementedError
-
-    @abc.abstractmethod
-    def close(self) -> None:
-        raise NotImplementedError
-
-    @property
-    @abc.abstractmethod
-    def is_closed(self) -> bool:
-        raise NotImplementedError
-
-    def get_user_acl(self, raw_user_permissions: str) -> UserAcl:
-        raw_acl_options = self.execute(
-            'fetch_acl_options',
-            cache=True,
-            cache_ttl=ACL_OPTIONS_CACHE_TTL,
-            limit=None,
-        )
-        return UserAcl(raw_acl_options, raw_user_permissions)
-
-
 class UserAcl:
     def __init__(self, raw_acl_options: List[Dict], raw_user_permissions: str):
         self._acl_options = self._parse_acl_options(raw_acl_options)
@@ -188,3 +117,74 @@ def has_privileges(self, *privileges: str, **kwargs: int) -> bool:
         for option in privileges:
             output |= self.has_privilege(option, forum_id=forum_id)
         return output
+
+
+class BaseBackend:
+    KNOWN_OPERATIONS = (
+        'fetch',
+        'get',
+        'has',
+        'set',
+    )
+    KNOWN_DRIVERS = (
+        'psycopg2',
+    )
+
+    def __init__(
+        self,
+        cache: cachelib.BaseCache,
+        config: Dict[str, Any]
+    ):
+        self._functions: Dict[str, str] = {}
+        self._connection = None
+        self._cache = cache
+        self._config = config
+
+        self._prepare_statements()
+        custom_statements = self._config.get('CUSTOM_STATEMENTS', {})
+        if not isinstance(custom_statements, dict):
+            custom_statements = {}
+        self._functions.update(custom_statements)
+
+    @abc.abstractmethod
+    def _setup_connection(self) -> None:
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def _prepare_statements(self) -> None:
+        raise NotImplementedError
+
+    @property
+    @abc.abstractmethod
+    def _db(self) -> Any:
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def execute(
+        self,
+        command: str,
+        cache: bool = False,
+        cache_ttl: Optional[int] = None,
+        skip: int = 0,
+        limit: Optional[int] = 10,
+        **kwargs: Union[int, str]
+    ) -> Any:
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def close(self) -> None:
+        raise NotImplementedError
+
+    @property
+    @abc.abstractmethod
+    def is_closed(self) -> bool:
+        raise NotImplementedError
+
+    def get_user_acl(self, raw_user_permissions: str) -> UserAcl:
+        raw_acl_options = self.execute(
+            'fetch_acl_options',
+            cache=True,
+            cache_ttl=ACL_OPTIONS_CACHE_TTL,
+            limit=None,
+        )
+        return UserAcl(raw_acl_options, raw_user_permissions)

flask_phpbb3/backends/psycopg2.py

@@ -1,5 +1,5 @@
 import json
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, Union
 
 try:
     import psycopg2
@@ -128,7 +128,7 @@ def _sql_query(
         cache_ttl: Optional[int] = None,
         skip: int = 0,
         limit: Optional[int] = 10,
-        **kwargs: int | str
+        **kwargs: Union[int, str]
     ) -> Any:
         """Executes a query with values in kwargs."""
         if operation not in self.KNOWN_OPERATIONS:
@@ -180,7 +180,7 @@ def _execute_operation(
         self,
         operation: str,
         query: str,
-        params: Dict[str, str | int]
+        params: Dict[str, Union[str, int]]
     ) -> Any:
         cursor = self._db.cursor()
 
@@ -214,7 +214,7 @@ def execute(
         cache_ttl: Optional[int] = None,
         skip: int = 0,
         limit: Optional[int] = 10,
-        **kwargs: int | str
+        **kwargs: Union[int, str]
     ) -> Any:
         cache_key_prefix: Optional[str] = None
         if cache:

flask_phpbb3/sessions.py

@@ -1,5 +1,7 @@
+from __future__ import annotations
+
 import json
-from typing import Any, Dict, Optional, Set
+from typing import Any, Dict, Optional, Set, Union
 
 import cachelib
 
@@ -30,7 +32,7 @@ def __init__(self) -> None:
         # request should not be executed multiple times
         self._request_cache: Dict[str, Any] = {}
 
-    def __setitem__(self, key: str, value: str | int) -> None:
+    def __setitem__(self, key: str, value: Union[str, int]) -> None:
         modified: bool = self.get(key) != value
         super(PhpBB3Session, self).__setitem__(key, value)
         if key not in self._read_only_properties:
@@ -60,7 +62,7 @@ def is_authenticated(self) -> bool:
         user_id = int(self.get('user_id', 1))
         return user_id > 1
 
-    def is_member(self, group: int | str) -> bool:
+    def is_member(self, group: Union[int, str]) -> bool:
         """Tests if user is a member of specified group."""
         if isinstance(group, int):
             # Try with default group