Fix Python 3.8 compatibility issue and normalize string formatting in this PR

Author: ViktorSky

firebase_admin/__init__.py

@@ -13,7 +13,6 @@
 # limitations under the License.
 
 """Firebase Admin SDK for Python."""
-import datetime
 import json
 import os
 import threading
@@ -26,11 +25,10 @@
 from firebase_admin import _typing
 
 
-_T = typing.TypeVar("_T")
+_T = typing.TypeVar('_T')
 
-_apps: typing.Dict[str, "App"] = {}
+_apps: typing.Dict[str, 'App'] = {}
 _apps_lock = threading.RLock()
-_clock = lambda: datetime.datetime.now(datetime.timezone.utc)
 
 _DEFAULT_APP_NAME = '[DEFAULT]'
 _FIREBASE_CONFIG_ENV_VAR = 'FIREBASE_CONFIG'
@@ -40,8 +38,8 @@
 def initialize_app(
     credential: typing.Optional[_typing.CredentialLike] = None,
     options: typing.Optional[typing.Dict[str, typing.Any]] = None,
-    name: str = _DEFAULT_APP_NAME
-) -> "App":
+    name: str = _DEFAULT_APP_NAME,
+) -> 'App':
     """Initializes and returns a new App instance.
 
     Creates a new App instance using the specified options
@@ -94,7 +92,7 @@ def initialize_app(
         'you call initialize_app().').format(name))
 
 
-def delete_app(app: "App") -> None:
+def delete_app(app: 'App') -> None:
     """Gracefully deletes an App instance.
 
     Args:
@@ -122,7 +120,7 @@ def delete_app(app: "App") -> None:
          'second argument.').format(app.name))
 
 
-def get_app(name: str = _DEFAULT_APP_NAME) -> "App":
+def get_app(name: str = _DEFAULT_APP_NAME) -> 'App':
     """Retrieves an App instance by name.
 
     Args:
@@ -280,7 +278,7 @@ def _lookup_project_id(self) -> typing.Optional[str]:
         project_id: typing.Optional[str] = self._options.get('projectId')
         if not project_id:
             try:
-                project_id = getattr(self._credential, "project_id")
+                project_id = getattr(self._credential, 'project_id')
             except (AttributeError, DefaultCredentialsError):
                 pass
         if not project_id:

firebase_admin/_auth_providers.py

@@ -26,7 +26,7 @@
 from firebase_admin import _user_mgt
 
 
-_ProviderConfigT = typing_extensions.TypeVar("_ProviderConfigT", bound="ProviderConfig", default="ProviderConfig")
+_ProviderConfigT = typing_extensions.TypeVar('_ProviderConfigT', bound='ProviderConfig', default='ProviderConfig')
 
 
 MAX_LIST_CONFIGS_RESULTS = 100
@@ -119,7 +119,7 @@ def __init__(
         self,
         download: typing.Callable[[typing.Optional[str], int], typing.Dict[str, _typing.Json]],
         page_token: typing.Optional[str],
-        max_results: int
+        max_results: int,
     ) -> None:
         self._download = download
         self._max_results = max_results
@@ -151,7 +151,7 @@ def get_next_page(self) -> typing.Optional[typing_extensions.Self]:
             return self.__class__(self._download, self.next_page_token, self._max_results)
         return None
 
-    def iterate_all(self) -> "_ProviderConfigIterator[_ProviderConfigT]":
+    def iterate_all(self) -> '_ProviderConfigIterator[_ProviderConfigT]':
         """Retrieves an iterator for provider configs.
 
         Returned iterator will iterate through all the provider configs in the Firebase project

firebase_admin/_auth_utils.py

@@ -27,7 +27,7 @@
 from firebase_admin import _utils
 
 
-_PageT = typing.TypeVar("_PageT", bound=_typing.Page)
+_PageT = typing.TypeVar('_PageT', bound=_typing.Page)
 
 EMULATOR_HOST_ENV_VAR = 'FIREBASE_AUTH_EMULATOR_HOST'
 MAX_CLAIMS_PAYLOAD_SIZE = 1000
@@ -381,7 +381,7 @@ def __init__(
         self,
         message: str,
         cause: typing.Optional[Exception],
-        http_response: typing.Optional[requests.Response]
+        http_response: typing.Optional[requests.Response],
     ) -> None:
         exceptions.AlreadyExistsError.__init__(self, message, cause, http_response)
 
@@ -395,7 +395,7 @@ def __init__(
         self,
         message: str,
         cause: typing.Optional[Exception],
-        http_response: typing.Optional[requests.Response]
+        http_response: typing.Optional[requests.Response],
     ) -> None:
         exceptions.AlreadyExistsError.__init__(self, message, cause, http_response)
 
@@ -412,7 +412,7 @@ def __init__(
         self,
         message: str,
         cause: typing.Optional[Exception],
-        http_response: typing.Optional[requests.Response]
+        http_response: typing.Optional[requests.Response],
     ) -> None:
         exceptions.PermissionDeniedError.__init__(self, message, cause, http_response)
 
@@ -426,7 +426,7 @@ def __init__(
         self,
         message: str,
         cause: typing.Optional[Exception],
-        http_response: typing.Optional[requests.Response]
+        http_response: typing.Optional[requests.Response],
     ) -> None:
         exceptions.InvalidArgumentError.__init__(self, message, cause, http_response)
 
@@ -637,7 +637,7 @@ def _parse_error_body(response: requests.Response) -> typing.Tuple[typing.Option
 def _build_error_message(
     code: str,
     exc_type: typing.Optional[_typing.FirebaseErrorFactory],
-    custom_message: typing.Optional[str]
+    custom_message: typing.Optional[str],
 ) -> str:
     default_message: str = getattr(exc_type, 'default_message', 'Error while calling Auth service')
     ext = ' {0}'.format(custom_message) if custom_message else ''

firebase_admin/_gapic_utils.py

@@ -26,9 +26,10 @@
 from firebase_admin import _typing
 from firebase_admin import _utils
 
+
 def handle_platform_error_from_googleapiclient(
     error: Exception,
-    handle_func: typing.Optional[_typing.GoogleAPIErrorHandler] = None
+    handle_func: typing.Optional[_typing.GoogleAPIErrorHandler] = None,
 ) -> exceptions.FirebaseError:
     """Constructs a ``FirebaseError`` from the given googleapiclient error.
 
@@ -61,7 +62,7 @@ def _handle_func_googleapiclient(
     error: Exception,
     message: str,
     error_dict: typing.Dict[str, typing.Any],
-    http_response: requests.Response
+    http_response: requests.Response,
 ) -> exceptions.FirebaseError:
     """Constructs a ``FirebaseError`` from the given GCP error.
 

firebase_admin/_http_client.py

@@ -38,7 +38,7 @@
 
 logger = logging.getLogger(__name__)
 
-_AnyT = typing_extensions.TypeVar("_AnyT", default=typing.Any)
+_AnyT = typing_extensions.TypeVar('_AnyT', default=typing.Any)
 
 if hasattr(retry.Retry.DEFAULT, 'allowed_methods'):
     _ANY_METHOD: typing.Dict[str, typing.Any] = {'allowed_methods': None}
@@ -55,8 +55,6 @@
 DEFAULT_HTTPX_RETRY_CONFIG = _retry.HttpxRetry(
     max_retries=4, status_forcelist=[500, 503], backoff_factor=0.5)
 
-
-
 DEFAULT_TIMEOUT_SECONDS = 120
 
 METRICS_HEADERS = {
@@ -75,7 +73,7 @@ def __init__(
         credential: typing.Optional[google.auth.credentials.Credentials] = None,
         session: typing.Optional[requests.Session] = None,
         base_url: str = '',
-        headers: typing.Optional["_typing.HeadersLike"] = None,
+        headers: typing.Optional['_typing.HeadersLike'] = None,
         retries: retry.Retry = DEFAULT_RETRY_CONFIG,
         timeout: int = DEFAULT_TIMEOUT_SECONDS,
     ) -> None:
@@ -181,9 +179,9 @@ def close(self) -> None:
             self._session = None
 
 
-class JsonHttpClient(HttpClient[typing.Dict[str ,"_typing.Json"]]):
+class JsonHttpClient(HttpClient[typing.Dict[str, '_typing.Json']]):
     """An HTTP client that parses response messages as JSON."""
-    def parse_body(self, resp: requests.Response) -> typing.Dict[str ,"_typing.Json"]:
+    def parse_body(self, resp: requests.Response) -> typing.Dict[str, '_typing.Json']:
         return resp.json()
 
 

firebase_admin/_messaging_encoder.py

@@ -23,8 +23,8 @@
 
 from firebase_admin import _messaging_utils
 
-_K = typing.TypeVar("_K")
-_V = typing.TypeVar("_V")
+_K = typing.TypeVar('_K')
+_V = typing.TypeVar('_V')
 
 
 class Message:
@@ -537,7 +537,7 @@ def encode_webpush_notification(
     @classmethod
     def encode_webpush_notification_actions(
         cls,
-        actions: typing.Optional[typing.List[_messaging_utils.WebpushNotificationAction]]
+        actions: typing.Optional[typing.List[_messaging_utils.WebpushNotificationAction]],
     ) -> typing.Optional[typing.List[typing.Dict[str, str]]]:
         """Encodes a list of ``WebpushNotificationActions`` into JSON."""
         if actions is None:

firebase_admin/_messaging_utils.py

@@ -72,12 +72,12 @@ class AndroidConfig:
     def __init__(
         self,
         collapse_key: typing.Optional[str] = None,
-        priority: typing.Optional[typing.Literal["high", "normal"]] = None,
+        priority: typing.Optional[typing.Literal['high', 'normal']] = None,
         ttl: typing.Optional[typing.Union[numbers.Real, datetime.timedelta]] = None,
         restricted_package_name: typing.Optional[str] = None,
         data: typing.Optional[typing.Dict[str, str]] = None,
-        notification: typing.Optional["AndroidNotification"] = None,
-        fcm_options: typing.Optional["AndroidFCMOptions"] = None,
+        notification: typing.Optional['AndroidNotification'] = None,
+        fcm_options: typing.Optional['AndroidFCMOptions'] = None,
         direct_boot_ok: typing.Optional[bool] = None,
     ) -> None:
         self.collapse_key = collapse_key
@@ -197,15 +197,15 @@ def __init__(
         sticky: typing.Optional[bool] = None,
         event_timestamp: typing.Optional[datetime.datetime] = None,
         local_only: typing.Optional[Incomplete] = None,
-        priority: typing.Optional[typing.Literal["default", "min", "low", "high", "max", "normal"]] = None,
+        priority: typing.Optional[typing.Literal['default', 'min', 'low', 'high', 'max', 'normal']] = None,
         vibrate_timings_millis: typing.Optional[float] = None,
         default_vibrate_timings: typing.Optional[bool] = None,
         default_sound: typing.Optional[bool] = None,
-        light_settings: typing.Optional["LightSettings"] = None,
+        light_settings: typing.Optional['LightSettings'] = None,
         default_light_settings: typing.Optional[bool] = None,
-        visibility: typing.Optional[typing.Literal["private", "public", "secret"]] = None,
+        visibility: typing.Optional[typing.Literal['private', 'public', 'secret']] = None,
         notification_count: typing.Optional[int] = None,
-        proxy: typing.Optional[typing.Literal["allow", "deny"]] = None,
+        proxy: typing.Optional[typing.Literal['allow', 'deny']] = None,
     ) -> None:
         self.title = title
         self.body = body
@@ -288,8 +288,8 @@ def __init__(
         self,
         headers: typing.Optional[typing.Dict[str, str]] = None,
         data: typing.Optional[typing.Dict[str, str]] = None,
-        notification: typing.Optional["WebpushNotification"] = None,
-        fcm_options: typing.Optional["WebpushFCMOptions"] = None,
+        notification: typing.Optional['WebpushNotification'] = None,
+        fcm_options: typing.Optional['WebpushFCMOptions'] = None,
     ) -> None:
         self.headers = headers
         self.data = data
@@ -355,7 +355,7 @@ def __init__(
         actions: typing.Optional[typing.List[WebpushNotificationAction]] = None,
         badge: typing.Optional[str] = None,
         data: typing.Optional[typing.Any] = None,
-        direction: typing.Optional[typing.Literal["auto", "ltr", "rtl"]] = None,
+        direction: typing.Optional[typing.Literal['auto', 'ltr', 'rtl']] = None,
         image: typing.Optional[str] = None,
         language: typing.Optional[str] = None,
         renotify: typing.Optional[bool] = None,
@@ -415,8 +415,8 @@ class APNSConfig:
     def __init__(
         self,
         headers: typing.Optional[typing.Dict[str, str]] = None,
-        payload: typing.Optional["APNSPayload"] = None,
-        fcm_options: typing.Optional["APNSFCMOptions"] = None,
+        payload: typing.Optional['APNSPayload'] = None,
+        fcm_options: typing.Optional['APNSFCMOptions'] = None,
         live_activity_token: typing.Optional[str] = None,
     ) -> None:
         self.headers = headers
@@ -434,7 +434,7 @@ class APNSPayload:
             (optional).
     """
 
-    def __init__(self, aps: "Aps", **kwargs: typing.Any) -> None:
+    def __init__(self, aps: 'Aps', **kwargs: typing.Any) -> None:
         self.aps = aps
         self.custom_data = kwargs
 
@@ -459,9 +459,9 @@ class Aps:
 
     def __init__(
         self,
-        alert: typing.Optional[typing.Union["ApsAlert", str]] = None,
+        alert: typing.Optional[typing.Union['ApsAlert', str]] = None,
         badge: typing.Optional[float] = None,  # should it be int?
-        sound: typing.Optional[typing.Union[str, "CriticalSound"]] = None,
+        sound: typing.Optional[typing.Union[str, 'CriticalSound']] = None,
         content_available: typing.Optional[bool] = None,
         category: typing.Optional[str] = None,
         thread_id: typing.Optional[str] = None,