- remove redundant overriding methods

- fix some pylint issues

Author: ViktorSky

firebase_admin/_auth_utils.py

@@ -24,13 +24,11 @@
     Literal,
     Optional,
     Protocol,
-    Union,
     cast,
     overload,
 )
 from urllib import parse
 
-import httpx
 import requests
 from typing_extensions import Self, TypeVar
 
@@ -476,28 +474,12 @@ class UidAlreadyExistsError(exceptions.AlreadyExistsError):
 
     default_message = 'The user with the provided uid already exists'
 
-    def __init__(
-        self,
-        message: str,
-        cause: Optional[Exception],
-        http_response: Optional[Union[httpx.Response, requests.Response]],
-    ) -> None:
-        super().__init__(message, cause, http_response)
-
 
 class EmailAlreadyExistsError(exceptions.AlreadyExistsError):
     """The user with the provided email already exists."""
 
     default_message = 'The user with the provided email already exists'
 
-    def __init__(
-        self,
-        message: str,
-        cause: Optional[Exception],
-        http_response: Optional[Union[httpx.Response, requests.Response]],
-    ) -> None:
-        super().__init__(message, cause, http_response)
-
 
 class InsufficientPermissionError(exceptions.PermissionDeniedError):
     """The credential used to initialize the SDK lacks required permissions."""
@@ -507,169 +489,70 @@ class InsufficientPermissionError(exceptions.PermissionDeniedError):
                        'https://firebase.google.com/docs/admin/setup for details '
                        'on how to initialize the Admin SDK with appropriate permissions')
 
-    def __init__(
-        self,
-        message: str,
-        cause: Optional[Exception],
-        http_response: Optional[Union[httpx.Response, requests.Response]],
-    ) -> None:
-        super().__init__(message, cause, http_response)
-
 
 class InvalidDynamicLinkDomainError(exceptions.InvalidArgumentError):
     """Dynamic link domain in ActionCodeSettings is not authorized."""
 
     default_message = 'Dynamic link domain specified in ActionCodeSettings is not authorized'
 
-    def __init__(
-        self,
-        message: str,
-        cause: Optional[Exception],
-        http_response: Optional[Union[httpx.Response, requests.Response]],
-    ) -> None:
-        super().__init__(message, cause, http_response)
-
 
 class InvalidIdTokenError(exceptions.InvalidArgumentError):
     """The provided ID token is not a valid Firebase ID token."""
 
     default_message = 'The provided ID token is invalid'
 
-    def __init__(
-        self,
-        message: str,
-        cause: Optional[Exception] = None,
-        http_response: Optional[Union[httpx.Response, requests.Response]] = None,
-    ) -> None:
-        super().__init__(message, cause, http_response)
-
 
 class PhoneNumberAlreadyExistsError(exceptions.AlreadyExistsError):
     """The user with the provided phone number already exists."""
 
     default_message = 'The user with the provided phone number already exists'
 
-    def __init__(
-        self,
-        message: str,
-        cause: Optional[Exception],
-        http_response: Optional[Union[httpx.Response, requests.Response]],
-    ) -> None:
-        super().__init__(message, cause, http_response)
-
 
 class UnexpectedResponseError(exceptions.UnknownError):
     """Backend service responded with an unexpected or malformed response."""
 
-    def __init__(
-        self,
-        message: str,
-        cause: Optional[Exception] = None,
-        http_response: Optional[Union[httpx.Response, requests.Response]] = None,
-    ) -> None:
-        super().__init__(message, cause, http_response)
-
 
 class UserNotFoundError(exceptions.NotFoundError):
     """No user record found for the specified identifier."""
 
     default_message = 'No user record found for the given identifier'
 
-    def __init__(
-        self,
-        message: str,
-        cause: Optional[Exception] = None,
-        http_response: Optional[Union[httpx.Response, requests.Response]] = None,
-    ) -> None:
-        super().__init__(message, cause, http_response)
-
 
 class EmailNotFoundError(exceptions.NotFoundError):
     """No user record found for the specified email."""
 
     default_message = 'No user record found for the given email'
 
-    def __init__(
-        self,
-        message: str,
-        cause: Optional[Exception] = None,
-        http_response: Optional[Union[httpx.Response, requests.Response]] = None,
-    ) -> None:
-        super().__init__(message, cause, http_response)
-
 
 class TenantNotFoundError(exceptions.NotFoundError):
     """No tenant found for the specified identifier."""
 
     default_message = 'No tenant found for the given identifier'
 
-    def __init__(
-        self,
-        message: str,
-        cause: Optional[Exception] = None,
-        http_response: Optional[Union[httpx.Response, requests.Response]] = None,
-    ) -> None:
-        super().__init__(message, cause, http_response)
-
 
 class TenantIdMismatchError(exceptions.InvalidArgumentError):
     """Missing or invalid tenant ID field in the given JWT."""
 
-    def __init__(self, message: str) -> None:
-        super().__init__(message)
-
 
 class ConfigurationNotFoundError(exceptions.NotFoundError):
     """No auth provider found for the specified identifier."""
 
     default_message = 'No auth provider found for the given identifier'
 
-    def __init__(
-        self,
-        message: str,
-        cause: Optional[Exception] = None,
-        http_response: Optional[Union[httpx.Response, requests.Response]] = None,
-    ) -> None:
-        super().__init__(message, cause, http_response)
-
 
 class UserDisabledError(exceptions.InvalidArgumentError):
     """An operation failed due to a user record being disabled."""
 
     default_message = 'The user record is disabled'
 
-    def __init__(
-        self,
-        message: str,
-        cause: Optional[Exception] = None,
-        http_response: Optional[Union[httpx.Response, requests.Response]] = None,
-    ) -> None:
-        super().__init__(message, cause, http_response)
-
 
 class TooManyAttemptsTryLaterError(exceptions.ResourceExhaustedError):
     """Rate limited because of too many attempts."""
 
-    def __init__(
-        self,
-        message: str,
-        cause: Optional[Exception] = None,
-        http_response: Optional[Union[httpx.Response, requests.Response]] = None,
-    ) -> None:
-        super().__init__(message, cause, http_response)
-
 
 class ResetPasswordExceedLimitError(exceptions.ResourceExhaustedError):
     """Reset password emails exceeded their limits."""
 
-    def __init__(
-        self,
-        message: str,
-        cause: Optional[Exception] = None,
-        http_response: Optional[Union[httpx.Response, requests.Response]] = None,
-    ) -> None:
-        super().__init__(message, cause, http_response)
-
 
 _CODE_TO_EXC_TYPE = {
     'CONFIGURATION_NOT_FOUND': ConfigurationNotFoundError,

firebase_admin/_sseclient.py

@@ -163,7 +163,7 @@ def __next__(self) -> Optional['Event']:
         return event
 
     def next(self) -> Optional['Event']:
-        return self.__next__()
+        return next(self)
 
 
 class Event:

firebase_admin/db.py

@@ -36,11 +36,12 @@
     cast,
     overload,
 )
-from typing_extensions import Self, TypeVar
 from urllib import parse
 
 import google.auth.credentials
 import requests
+from typing_extensions import Self, TypeVar
+
 
 import firebase_admin
 from firebase_admin import exceptions
@@ -849,7 +850,10 @@ def __gt__(self, other: '_SortEntry') -> bool:
     def __ge__(self, other: '_SortEntry') -> bool:
         return self._compare(other) >= 0
 
-    def __eq__(self, other: '_SortEntry') -> bool:  # pyright: ignore[reportIncompatibleMethodOverride]
+    def __eq__(  # pyright: ignore[reportIncompatibleMethodOverride]
+        self,
+        other: '_SortEntry',
+    ) -> bool:
         return self._compare(other) == 0
 
 

firebase_admin/ml.py

@@ -290,7 +290,8 @@ def _convert_to_millis(date_string: Optional[str]) -> Optional[int]:
             return None
         format_str = '%Y-%m-%dT%H:%M:%S.%fZ'
         epoch = datetime.datetime.fromtimestamp(0, datetime.timezone.utc)
-        datetime_object = datetime.datetime.strptime(date_string, format_str).replace(tzinfo=datetime.timezone.utc)
+        datetime_object = datetime.datetime.strptime(
+            date_string, format_str).replace(tzinfo=datetime.timezone.utc)
         millis = int((datetime_object - epoch).total_seconds() * 1000)
         return millis
 
@@ -881,7 +882,11 @@ def get_operation(self, op_name: str) -> dict[str, Any]:
         except requests.exceptions.RequestException as error:
             raise _utils.handle_platform_error_from_requests(error)
 
-    def _exponential_backoff(self, current_attempt: int, stop_time: Optional[datetime.datetime]) -> None:
+    def _exponential_backoff(
+        self,
+        current_attempt: int,
+        stop_time: Optional[datetime.datetime],
+    ) -> None:
         """Sleeps for the appropriate amount of time. Or throws deadline exceeded."""
         delay_factor = pow(_MLService.POLL_EXPONENTIAL_BACKOFF_FACTOR, current_attempt)
         wait_time_seconds = delay_factor * _MLService.POLL_BASE_WAIT_TIME_SECONDS

firebase_admin/remote_config.py

@@ -302,7 +302,10 @@ def _get_url(self) -> str:
         return f"/v1/projects/{self._project_id}/namespaces/firebase-server/serverRemoteConfig"
 
     @classmethod
-    def _handle_remote_config_error(cls, error: requests.RequestException) -> exceptions.FirebaseError:
+    def _handle_remote_config_error(
+        cls,
+        error: requests.RequestException,
+    ) -> exceptions.FirebaseError:
         """Handles errors received from the Cloud Functions API."""
         return _utils.handle_platform_error_from_requests(error)
 

firebase_admin/tenant_mgt.py

@@ -340,7 +340,7 @@ def delete_tenant(self, tenant_id: str) -> None:
         """Deletes the tenant corresponding to the given ``tenant_id``."""
         if not isinstance(tenant_id, str) or not tenant_id:
             raise ValueError(
-                'Invalid tenant ID: {0}. Tenant ID must be a non-empty string.'.format(tenant_id))
+                f'Invalid tenant ID: {tenant_id}. Tenant ID must be a non-empty string.')
 
         try:
             self.client.request('delete', f'/tenants/{tenant_id}')