class DbtBaseOperator:

    @apply_defaults
    def __init__(self,
                 macro=None,
                 args=None,
                 *positional_args,
                 **kwargs):
        super(DbtBaseOperator, self).__init__(*positional_args, **kwargs)

        if args is not None and not isinstance(args, dict):
            raise AirflowException('The argument "args" should be a dictionary')

        self.macro = macro
        self.args = args

    def create_hook(self):
        self.hook = DbtHook(
            macro=self.macro,
            args=self.args)

        return self.hook

class DbtRunOperationOperator(DbtBaseOperator):
    @apply_defaults
    def __init__(self, profiles_dir=None, target=None, *args, **kwargs):
        super(DbtRunOperationOperator, self).__init__(profiles_dir=profiles_dir, target=target, *args, **kwargs)

    def execute(self, context):
        self.create_hook().execute('run-operation', self.macro)