import abc

from airflow.models.baseoperator import BaseOperator
from airflow.operators.python import get_current_context

from fansifter.hooks.lambda_function import FanSifterLambdaHook


class FanSifterBaseOperator(BaseOperator, metaclass=abc.ABCMeta):
    ui_color = "#ffefeb"

    def _get_config(self):
        context = get_current_context()
        return context["dag_run"].conf


class FanSifterLambdaOperator(FanSifterBaseOperator, metaclass=abc.ABCMeta):
    def invoke_lambda(self, function_name, payload):
        lambda_hook = FanSifterLambdaHook(function_name=function_name)

        return lambda_hook.invoke_lambda(payload)