Coverage for airflow/plugins_manager.py : 51%

# -*- coding: utf-8 -*- 

# 

# 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. 

# 

from __future__ import absolute_import 

from __future__ import division 

from __future__ import print_function 

from __future__ import unicode_literals 

from builtins import object 

import imp 

import inspect 

import logging 

import os 

import re 

import sys 

from itertools import chain 

merge = chain.from_iterable 

from airflow import configuration 

from airflow.utils.timeout import timeout 

class AirflowPluginException(Exception): 

pass 

class AirflowPlugin(object): 

name = None 

operators = [] 

hooks = [] 

executors = [] 

macros = [] 

admin_views = [] 

flask_blueprints = [] 

menu_links = [] 

@classmethod 

def validate(cls): 

if not cls.name: 

raise AirflowPluginException("Your plugin needs a name.") 

plugins_folder = configuration.get('core', 'plugins_folder') 

if not plugins_folder: 

plugins_folder = configuration.get('core', 'airflow_home') + '/plugins' 

plugins_folder = os.path.expanduser(plugins_folder) 

if plugins_folder not in sys.path: 

sys.path.append(plugins_folder) 

plugins = [] 

norm_pattern = re.compile(r'[/|.]') 

plugin_import_timeout = configuration.getint('core', 'plugin_import_timeout') 

# Crawl through the plugins folder to find AirflowPlugin derivatives 

for root, dirs, files in os.walk(plugins_folder, followlinks=True): 

logging.debug('Loading plugin located in {}'.format(root)) 

for f in files: 

print("-="*40) 

print(f) 

try: 

filepath = os.path.join(root, f) 

if not os.path.isfile(filepath): 

continue 

mod_name, file_ext = os.path.splitext( 

os.path.split(filepath)[-1]) 

if file_ext != '.py': 

continue 

# normalize root path as namespace 

namespace = '_'.join([re.sub(norm_pattern, '__', root), mod_name]) 

with timeout(plugin_import_timeout): 

m = imp.load_source(namespace, filepath) 

for obj in list(m.__dict__.values()): 

if ( 

inspect.isclass(obj) and 

issubclass(obj, AirflowPlugin) and 

obj is not AirflowPlugin): 

obj.validate() 

if obj not in plugins: 

plugins.append(obj) 

except Exception as e: 

logging.exception(e) 

logging.error('Failed to import plugin ' + filepath) 

def make_module(name, objects): 

name = name.lower() 

module = imp.new_module(name) 

module._name = name.split('.')[-1] 

module._objects = objects 

for o in objects: 

module.__dict__.update((o.__name__, o)) 

return module 

operators, hooks, executors, macros, admin_views = [], [], [], [], [] 

flask_blueprints, menu_links = [], [] 

for p in plugins: 

operators.append(make_module('airflow.operators.' + p.name, p.operators)) 

hooks.append(make_module('airflow.hooks.' + p.name, p.hooks)) 

executors.append(make_module('airflow.executors.' + p.name, p.executors)) 

macros.append(make_module('airflow.macros.' + p.name, p.macros)) 

admin_views.append( 

make_module('airflow.www.admin_views' + p.name, p.admin_views)) 

flask_blueprints.append( 

make_module( 

'airflow.www.flask_blueprints' + p.name, p.flask_blueprints)) 

menu_links.append( 

make_module('airflow.www.menu_links' + p.name, p.menu_links))