feat: add lifecycle hooks

Author: tjholm

nitric/lifecycle/__init__.py

@@ -0,0 +1,48 @@
+#
+# Copyright (c) 2021 Nitric Technologies Pty Ltd.
+#
+# This file is part of Nitric Python 3 SDK.
+# See https://github.com/nitrictech/python-sdk for further info.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+"""
+Lifecycle utilities for Nitric applications.
+
+This module provides utilities for detecting the current lifecycle stage and
+conditionally executing code based on the current stage.
+"""
+
+from nitric.lifecycle.lifecycle import (
+    Lifecycle,
+    LifecycleStage,
+    get_current_lifecycle,
+    is_collecting,
+    is_in_lifecycle,
+    is_running,
+    when_collecting,
+    when_in_lifecycles,
+    when_running,
+)
+
+__all__ = [
+    "Lifecycle",
+    "LifecycleStage",
+    "get_current_lifecycle",
+    "is_collecting",
+    "is_in_lifecycle",
+    "is_running",
+    "when_collecting",
+    "when_in_lifecycles",
+    "when_running",
+]

nitric/lifecycle/lifecycle.py

@@ -0,0 +1,216 @@
+#
+# Copyright (c) 2021 Nitric Technologies Pty Ltd.
+#
+# This file is part of Nitric Python 3 SDK.
+# See https://github.com/nitrictech/python-sdk for further info.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import os
+from enum import Enum
+from typing import Callable, List, TypeVar, Optional
+
+# The environment variable key that will be used to determine the current Nitric lifecycle/executing environment
+NITRIC_ENVIRONMENT = 'NITRIC_ENVIRONMENT'
+
+
+# Possible nitric execution environments
+class LifecycleStage(Enum):
+    """Represents the different stages of the Nitric application lifecycle."""
+
+    # Local development run (using nitric run/start)
+    LOCAL_RUN = 'run'
+
+    # Local development requirements building/collection (using nitric up)
+    BUILD = 'build'
+
+    # When the code is running in a deployed environment
+    CLOUD = 'cloud'
+
+
+def get_current_lifecycle() -> LifecycleStage:
+    """
+    Get the current lifecycle stage from the environment variable.
+
+    Returns:
+        The current lifecycle stage.
+
+    Raises:
+        ValueError: If the NITRIC_ENVIRONMENT environment variable is not set or is invalid.
+    """
+    lifecycle = os.environ.get(NITRIC_ENVIRONMENT)
+
+    if not lifecycle or lifecycle not in [stage.value for stage in LifecycleStage]:
+        raise ValueError(
+            f"Unable to determine the current Nitric lifecycle, please ensure the {NITRIC_ENVIRONMENT} "
+            "environment variable is set"
+        )
+
+    return LifecycleStage(lifecycle)
+
+
+def is_in_lifecycle(stages: List[LifecycleStage]) -> bool:
+    """
+    Check if the current environment is one of the provided stages.
+
+    Args:
+        stages: The stages to check against.
+
+    Returns:
+        True if the current stage is in the provided stages, False otherwise.
+    """
+    current_stage = get_current_lifecycle()
+    return current_stage in stages
+
+
+T = TypeVar('T')
+
+
+def when_in_lifecycles(stages: List[LifecycleStage], callback: Callable[[], T]) -> Optional[T]:
+    """
+    If the current environment is one of the provided stages, execute the provided callback.
+
+    Args:
+        stages: The stages to check against.
+        callback: The callback to execute if the current stage is in the provided stages.
+
+    Returns:
+        The result of the callback if executed, None otherwise.
+    """
+    if is_in_lifecycle(stages):
+        return callback()
+    return None
+
+
+def when_running(callback: Callable[[], T]) -> Optional[T]:
+    """
+    If the current environment is running (local or cloud), execute the provided callback.
+
+    Args:
+        callback: The callback to execute if the current stage is running.
+
+    Returns:
+        The result of the callback if executed, None otherwise.
+    """
+    return when_in_lifecycles([LifecycleStage.LOCAL_RUN, LifecycleStage.CLOUD], callback)
+
+
+def when_collecting(callback: Callable[[], T]) -> Optional[T]:
+    """
+    If the current environment is collecting requirements, execute the provided callback.
+
+    Args:
+        callback: The callback to execute if the current stage is collecting.
+
+    Returns:
+        The result of the callback if executed, None otherwise.
+    """
+    return when_in_lifecycles([LifecycleStage.BUILD], callback)
+
+
+def is_running() -> bool:
+    """
+    Check if the current lifecycle is running the app.
+
+    Returns:
+        True if the current stage is running, False otherwise.
+    """
+    return is_in_lifecycle([LifecycleStage.LOCAL_RUN, LifecycleStage.CLOUD])
+
+
+def is_collecting() -> bool:
+    """
+    Check if the current lifecycle is collecting application requirements.
+
+    Returns:
+        True if the current stage is collecting, False otherwise.
+    """
+    return is_in_lifecycle([LifecycleStage.BUILD])
+
+
+# Create a class to match the Node SDK's export style
+class Lifecycle:
+    """Lifecycle utilities for Nitric applications."""
+
+    @staticmethod
+    def lifecycle_is(stages: List[LifecycleStage]) -> bool:
+        """
+        Check if the current environment is one of the provided stages.
+
+        Args:
+            stages: The stages to check against.
+
+        Returns:
+            True if the current stage is in the provided stages, False otherwise.
+        """
+        return is_in_lifecycle(stages)
+
+    @staticmethod
+    def is_collecting() -> bool:
+        """
+        Check if the current lifecycle is collecting application requirements.
+
+        Returns:
+            True if the current stage is collecting, False otherwise.
+        """
+        return is_collecting()
+
+    @staticmethod
+    def is_running() -> bool:
+        """
+        Check if the current lifecycle is running the app.
+
+        Returns:
+            True if the current stage is running, False otherwise.
+        """
+        return is_running()
+
+    @staticmethod
+    def when(stages: List[LifecycleStage], callback: Callable[[], T]) -> Optional[T]:
+        """
+        If the current environment is one of the provided stages, execute the provided callback.
+
+        Args:
+            stages: The stages to check against.
+            callback: The callback to execute if the current stage is in the provided stages.
+
+        Returns:
+            The result of the callback if executed, None otherwise.
+        """
+        return when_in_lifecycles(stages, callback)
+
+    @staticmethod
+    def when_collecting(callback: Callable[[], T]) -> Optional[T]:
+        """
+        If the current environment is collecting requirements, execute the provided callback.
+
+        Args:
+            callback: The callback to execute if the current stage is collecting.
+
+        Returns:
+            The result of the callback if executed, None otherwise.
+        """
+        return when_collecting(callback)
+
+    @staticmethod
+    def when_running(callback: Callable[[], T]) -> Optional[T]:
+        """
+        If the current environment is running (local or cloud), execute the provided callback.
+
+        Args:
+            callback: The callback to execute if the current stage is running.
+
+        Returns:
+            The result of the callback if executed, None otherwise.
+        """
+        return when_running(callback)