App Functions SDK for Python - App Function Context API

class AppFunctionContext(ABC):
    """
    An abstract base class that defines the interface for an application function context.

    This class provides an interface for cloning the context, getting and setting response data and
    content type, triggering retry for failed data, getting the secret provider, and getting the
    logger.

    Methods:
        clone() -> 'AppFunctionContext': Clones the context.
        correlation_id() -> str: Gets the correlation ID.
        input_content_type() -> str: Gets the input content type.
        set_response_data(data: bytes): Sets the response data.
        response_data() -> bytes: Gets the response data.
        set_response_content_type(content_type: str): Sets the response content type.
        response_content_type() -> str: Gets the response content type.
        set_retry_data(data: bytes): Sets the retry data.
        trigger_retry_failed_data(): Triggers retry for failed data.
        secret_provider() -> SecretProvider: Gets the secret provider.
        logger() -> 'Logger': Gets the logger.
    """

    @abstractmethod
    def clone(self) -> 'AppFunctionContext':
        """
        Clones the context.

        Returns:
            A clone of the context.
        """

    @abstractmethod
    def correlation_id(self) -> str:
        """
        Gets the correlation ID.

        Returns:
            The correlation ID.
        """

    @abstractmethod
    def set_correlation_id(self, correlation_id: str):
        """
        Sets the correlation ID.
        """

    @abstractmethod
    def input_content_type(self) -> str:
        """
        Gets the input content type.

        Returns:
            The input content type.
        """

    @abstractmethod
    def set_input_content_type(self, input_content_type: str):
        """
        Sets the input content type.
        """

    @abstractmethod
    def set_response_data(self, data: bytes):
        """
        Sets the response data.

        Args:
            data: The response data.
        """

    @abstractmethod
    def response_data(self) -> bytes:
        """
        Gets the response data.

        Returns:
            The response data.
        """

    @abstractmethod
    def set_response_content_type(self, content_type: str):
        """
        Sets the response content type.

        Args:
            content_type: The response content type.
        """

    @abstractmethod
    def response_content_type(self) -> str:
        """
        Gets the response content type.

        Returns:
            The response content type.
        """

    @abstractmethod
    def set_retry_data(self, data: bytes):
        """
        Sets the retry data.

        Args:
            data: The retry data.
        """

    @abstractmethod
    def retry_data(self) -> bytes:
        """
        Gets the retry data.
        """

    @abstractmethod
    def trigger_retry_failed_data(self):
        """
        Triggers retry for failed data.
        """

    @abstractmethod
    def secret_provider(self) -> SecretProvider:
        """
        Gets the secret provider.

        Returns:
            The secret provider.
        """

    @abstractmethod
    def logger(self) -> 'Logger':
        """
        Gets the logger.

        Returns:
            The logger.
        """

    @abstractmethod
    def pipeline_id(self) -> str:
        """
        Gets the pipeline ID.

        Returns:
            The pipeline ID.
        """

    @abstractmethod
    def add_value(self, key: str, value: str):
        """
        Adds the key and value to context_data.

        Returns:
            The pipeline ID.
        """

    @abstractmethod
    def remove_value(self, key: str):
        """
        Deletes a value stored in the context at the given key
        """

    @abstractmethod
    def get_value(self, key: str) -> Tuple[str, bool]:
        """
        Attempts to retrieve a value stored in the context at the given key
        """

    @abstractmethod
    def get_values(self) -> dict:
        """
        GetAllValues returns a read-only copy of all data stored in the context
        """

    @abstractmethod
    def apply_values(self, str_format: str) -> str:
        """
        apply_values looks in the provided string for placeholders of the form
        '{any-value-key}' and attempts to replace with the value stored under
        the key in context storage.  An error will be returned if any placeholders
        are not matched to a value in the context.
        """

    @abstractmethod
    def event_client(self) -> EventClientABC:
        """
        event_client returns the event client instance
        """

    @abstractmethod
    def reading_client(self) -> ReadingClientABC:
        """
        reading_client returns the reading client instance
        """

    @abstractmethod
    def command_client(self) -> CommandClientABC:
        """
        command_client returns the command client instance
        """

    @abstractmethod
    def device_service_client(self) -> DeviceServiceClientABC:
        """
        device_service_client returns the device service client instance
        """

class SecretProvider(ABC):
    """
    An abstract base class that defines the interface for a secret provider.

    This class provides an interface for storing and retrieving secrets, checking the last update
    time of secrets, listing secret names, checking the existence of secrets, and registering or
    deregistering secret update callbacks.

    Methods:
        store_secrets(secret_name: str, secrets: Secrets): Stores secrets.
        get_secrets(secret_name: str, *secret_keys: str) -> Secrets: Retrieves secrets.
        secrets_last_updated() -> datetime: Checks the last update time of secrets.
        list_secret_names() -> List[str]: Lists secret names.
        has_secrets(secret_name: str) -> bool: Checks the existence of secrets.
        register_secret_update_callback(secret_name: str, callback: Callable[[str], None]):
        Registers a secret update callback.
        deregister_secret_update_callback(secret_name: str): Deregisters a secret update callback.
    """

    @abstractmethod
    def store_secrets(self, secret_name: str, secrets: Secrets):
        """
        Stores secrets.

        Args:
            secret_name: The name of the secret.
            secrets: The secrets to be stored.
        """

    @abstractmethod
    def get_secrets(self, secret_name: str, *secret_keys: str) -> Secrets:
        """
        Retrieves secrets.

        Args:
            secret_name: The name of the secret.
            secret_keys: The keys of the secrets to be retrieved.

        Returns:
            The retrieved secrets.
        """

    @abstractmethod
    def secrets_last_updated(self) -> datetime:
        """
        Checks the last update time of secrets.

        Returns:
            The last update time of secrets.
        """

    @abstractmethod
    def list_secret_names(self) -> List[str]:
        """
        Lists secret names.

        Returns:
            A list of secret names.
        """

    @abstractmethod
    def has_secret(self, secret_name: str) -> bool:
        """
        Checks the existence of secrets.

        Args:
            secret_name: The name of the secret.

        Returns:
            True if the secret exists, False otherwise.
        """

    @abstractmethod
    def register_secret_update_callback(self, secret_name: str, callback: Callable[[str], None]):
        """
        Registers a secret update callback.

        Args:
            secret_name: The name of the secret.
            callback: The callback to be registered.
        """

    @abstractmethod
    def deregister_secret_update_callback(self, secret_name: str):
        """
        Deregisters a secret update callback.

        Args:
            secret_name: The name of the secret.
        """

from pyformance import meters
from app_functions_sdk_py.interfaces import AppFunctionContext

def register_my_counter_metric(ctx: AppFunctionContext):
    counter_metric_name = "MyCounter"
    my_counter = meters.Counter("")
    my_tags = {"Tag1": "Value1"}
    ctx.metrics_manager().register(counter_metric_name, my_counter, my_tags)