Source code for q2_sdk.tools.decorators

import asyncio
import logging
import warnings
from functools import wraps

from q2_sdk.core import configuration
from q2_sdk.core.exceptions import DangerousCodeError, DevOnlyError

DANGEROUS_MSG = (
    "!!!{func_name}: {reason}!!!\n\n "
    "{func_name} not in DANGEROUS_WHITELIST. "
    "Some functions are particularly powerful, "
    "so just to ensure this is actually what you meant "
    "to do you need to opt-in to their use by adding them "
    "to the DANGEROUS_WHITELIST list in your settings file. "
)


def dangerous(reason):
    def decorator(function):
        async def async_wrapper(*args, **kwargs):
            settings = configuration.get_settings()
            func_name = function.__name__
            if func_name not in settings.DANGEROUS_WHITELIST:
                error_msg = DANGEROUS_MSG.format(func_name=func_name, reason=reason)
                raise DangerousCodeError(error_msg)
            return await function(*args, **kwargs)

        return async_wrapper

    return decorator


def dev_only(function):
    @wraps(function)
    async def async_wrapper(*args, **kwargs):
        env = configuration.settings.DEPLOY_ENV
        if env != "DEV":
            raise DevOnlyError("This is a dev only method")
        return await function(*args, **kwargs)

    return async_wrapper


def deprecated(function):
    def wrapping_logic():
        logger = logging.getLogger("q2_sdk.general")
        function_name = function.__name__
        warning_message = f'The function "{function_name}" is deprecated and may be removed in a future release'
        logger.warning(warning_message)

        warnings.warn(warning_message, DeprecationWarning, stacklevel=2)

    @wraps(function)
    def wrapper(*args, **kwargs):
        if not asyncio.iscoroutinefunction(function):
            wrapping_logic()
            return function(*args, **kwargs)
        else:

            async def tmp():
                wrapping_logic()
                return await function(*args, **kwargs)

            return tmp()

    return wrapper