add type annotations

rebase

Author: ViktorSky

firebase_admin/__init__.py

@@ -13,27 +13,41 @@
 # limitations under the License.
 
 """Firebase Admin SDK for Python."""
-import datetime
+
 import json
 import os
 import threading
+from collections.abc import Callable
+from typing import Any, Optional, TypeVar, Union, overload
+
+import google.auth.credentials
+import google.auth.exceptions
 
-from google.auth.credentials import Credentials as GoogleAuthCredentials
-from google.auth.exceptions import DefaultCredentialsError
 from firebase_admin import credentials
 from firebase_admin.__about__ import __version__
 
+__all__ = (
+    'App',
+    'delete_app',
+    'get_app',
+    'initialize_app',
+)
 
-_apps = {}
+_T = TypeVar('_T')
+
+_apps: dict[str, 'App'] = {}
 _apps_lock = threading.RLock()
-_clock = datetime.datetime.utcnow
 
 _DEFAULT_APP_NAME = '[DEFAULT]'
 _FIREBASE_CONFIG_ENV_VAR = 'FIREBASE_CONFIG'
 _CONFIG_VALID_KEYS = ['databaseAuthVariableOverride', 'databaseURL', 'httpTimeout', 'projectId',
                       'storageBucket']
 
-def initialize_app(credential=None, options=None, name=_DEFAULT_APP_NAME):
+def initialize_app(
+    credential: Optional[Union[credentials.Base, google.auth.credentials.Credentials]] = None,
+    options: Optional[dict[str, Any]] = None,
+    name: str = _DEFAULT_APP_NAME,
+) -> 'App':
     """Initializes and returns a new App instance.
 
     Creates a new App instance using the specified options
@@ -86,7 +100,7 @@ def initialize_app(credential=None, options=None, name=_DEFAULT_APP_NAME):
         'you call initialize_app().')
 
 
-def delete_app(app):
+def delete_app(app: 'App') -> None:
     """Gracefully deletes an App instance.
 
     Args:
@@ -113,7 +127,7 @@ def delete_app(app):
         'second argument.')
 
 
-def get_app(name=_DEFAULT_APP_NAME):
+def get_app(name: str = _DEFAULT_APP_NAME) -> 'App':
     """Retrieves an App instance by name.
 
     Args:
@@ -147,7 +161,7 @@ def get_app(name=_DEFAULT_APP_NAME):
 class _AppOptions:
     """A collection of configuration options for an App."""
 
-    def __init__(self, options):
+    def __init__(self, options: Optional[dict[str, Any]]) -> None:
         if options is None:
             options = self._load_from_environment()
 
@@ -157,11 +171,16 @@ def __init__(self, options):
                 'Options must be a dictionary.')
         self._options = options
 
-    def get(self, key, default=None):
+    @overload
+    def get(self, key: str, default: None = None) -> Optional[Any]: ...
+    # possible issue: needs return Any | _T ?
+    @overload
+    def get(self, key: str, default: _T) -> _T: ...
+    def get(self, key: str, default: Optional[Any] = None) -> Optional[Any]:
         """Returns the option identified by the provided key."""
         return self._options.get(key, default)
 
-    def _load_from_environment(self):
+    def _load_from_environment(self) -> dict[str, Any]:
         """Invoked when no options are passed to __init__, loads options from FIREBASE_CONFIG.
 
         If the value of the FIREBASE_CONFIG environment variable starts with "{" an attempt is made
@@ -194,7 +213,12 @@ class App:
     common to all Firebase APIs.
     """
 
-    def __init__(self, name, credential, options):
+    def __init__(
+        self,
+        name: str,
+        credential: Union[credentials.Base, google.auth.credentials.Credentials],
+        options: Optional[dict[str, Any]],
+    ) -> None:
         """Constructs a new App using the provided name and options.
 
         Args:
@@ -211,7 +235,7 @@ def __init__(self, name, credential, options):
                 'non-empty string.')
         self._name = name
 
-        if isinstance(credential, GoogleAuthCredentials):
+        if isinstance(credential, google.auth.credentials.Credentials):
             self._credential = credentials._ExternalCredentials(credential) # pylint: disable=protected-access
         elif isinstance(credential, credentials.Base):
             self._credential = credential
@@ -220,37 +244,38 @@ def __init__(self, name, credential, options):
                              'with a valid credential instance.')
         self._options = _AppOptions(options)
         self._lock = threading.RLock()
-        self._services = {}
+        self._services: Optional[dict[str, Any]] = {}
 
         App._validate_project_id(self._options.get('projectId'))
-        self._project_id_initialized = False
+        self._project_id_initialized: bool = False
 
     @classmethod
-    def _validate_project_id(cls, project_id):
+    def _validate_project_id(cls, project_id: Optional[Any]) -> Optional[str]:
         if project_id is not None and not isinstance(project_id, str):
             raise ValueError(
                 f'Invalid project ID: "{project_id}". project ID must be a string.')
+        return project_id
 
     @property
-    def name(self):
+    def name(self) -> str:
         return self._name
 
     @property
-    def credential(self):
+    def credential(self) -> credentials.Base:
         return self._credential
 
     @property
-    def options(self):
+    def options(self) -> _AppOptions:
         return self._options
 
     @property
-    def project_id(self):
+    def project_id(self) -> Optional[str]:
         if not self._project_id_initialized:
             self._project_id = self._lookup_project_id()
             self._project_id_initialized = True
         return self._project_id
 
-    def _lookup_project_id(self):
+    def _lookup_project_id(self) -> Optional[str]:
         """Looks up the Firebase project ID associated with an App.
 
         If a ``projectId`` is specified in app options, it is returned. Then tries to
@@ -264,16 +289,16 @@ def _lookup_project_id(self):
         project_id = self._options.get('projectId')
         if not project_id:
             try:
-                project_id = self._credential.project_id
-            except (AttributeError, DefaultCredentialsError):
+                project_id = getattr(self._credential, 'project_id')
+            except (AttributeError, google.auth.exceptions.DefaultCredentialsError):
                 pass
         if not project_id:
             project_id = os.environ.get('GOOGLE_CLOUD_PROJECT',
                                         os.environ.get('GCLOUD_PROJECT'))
         App._validate_project_id(self._options.get('projectId'))
         return project_id
 
-    def _get_service(self, name, initializer):
+    def _get_service(self, name: str, initializer: Callable[['App'], _T]) -> _T:
         """Returns the service instance identified by the given name.
 
         Services are functional entities exposed by the Admin SDK (e.g. auth, database). Each
@@ -303,15 +328,16 @@ def _get_service(self, name, initializer):
                 self._services[name] = initializer(self)
             return self._services[name]
 
-    def _cleanup(self):
+    def _cleanup(self) -> None:
         """Cleans up any services associated with this App.
 
         Checks whether each service contains a close() method, and calls it if available.
         This is to be called when an App is being deleted, thus ensuring graceful termination of
         any services started by the App.
         """
         with self._lock:
-            for service in self._services.values():
-                if hasattr(service, 'close') and hasattr(service.close, '__call__'):
-                    service.close()
-            self._services = None
+            if self._services is not None:
+                for service in self._services.values():
+                    if hasattr(service, 'close') and hasattr(service.close, '__call__'):
+                        service.close()
+                self._services = None