Merge pull request apache#234 from jlowin/Py3

Python 3 compatibility, Stage 1

airflow/bin/cli.py

@@ -1,4 +1,5 @@
 #!/usr/bin/env python
+from __future__ import print_function
 import argparse
 import dateutil.parser
 from datetime import datetime
@@ -134,9 +135,9 @@ def run(args):
                 session.add(pickle)
                 session.commit()
                 pickle_id = pickle.id
-                print(
+                print((
                     'Pickled dag {dag} '
-                    'as pickle_id:{pickle_id}').format(**locals())
+                    'as pickle_id:{pickle_id}').format(**locals()))
             except Exception as e:
                 print('Could not pickle the DAG')
                 print(e)
@@ -169,7 +170,7 @@ def task_state(args):
     dag = dagbag.dags[args.dag_id]
     task = dag.get_task(task_id=args.task_id)
     ti = TaskInstance(task, args.execution_date)
-    print ti.current_state()
+    print(ti.current_state())
 
 
 def list_dags(args):

airflow/example_dags/example_python_operator.py

@@ -1,3 +1,4 @@
+from __future__ import print_function
 from airflow.operators import PythonOperator
 from airflow.models import DAG
 from datetime import datetime, timedelta

airflow/hooks/druid_hook.py

@@ -1,3 +1,4 @@
+from __future__ import print_function
 import logging
 import json
 import time
@@ -115,8 +116,8 @@ def send_ingest_query(
             metric_spec, intervals)
 	r = requests.post(
             self.ingest_post_url, headers=self.header, data=query)
-        print self.ingest_post_url
-        print query
+        print(self.ingest_post_url)
+        print(query)
         print(r.text)
         d = json.loads(r.text)
         if "task" not in d:

airflow/hooks/hdfs_hook.py

@@ -1,5 +1,9 @@
 from airflow.hooks.base_hook import BaseHook
-from snakebite.client import Client, HAClient, Namenode
+try:
+    snakebite_imported = True
+    from snakebite.client import Client, HAClient, Namenode
+except ImportError:
+    snakebite_imported = False
 
 from airflow.utils import AirflowException
 
@@ -13,6 +17,12 @@ class HDFSHook(BaseHook):
     Interact with HDFS. This class is a wrapper around the snakebite library.
     '''
     def __init__(self, hdfs_conn_id='hdfs_default'):
+        if not snakebite_imported:
+            raise ImportError(
+                'This HDFSHook implementation requires snakebite, but '
+                'snakebite is not compatible with Python 3 '
+                '(as of August 2015). Please use Python 2 if you require '
+                'this hook  -- or help by submitting a PR!')
         self.hdfs_conn_id = hdfs_conn_id
 
     def get_conn(self):

airflow/hooks/hive_hooks.py

@@ -1,3 +1,4 @@
+from __future__ import print_function
 import csv
 import logging
 import subprocess

airflow/macros/__init__.py

@@ -1,7 +1,8 @@
+from __future__ import absolute_import
 from random import random
 from datetime import datetime, timedelta
 import time
-import hive
+from . import hive
 import uuid
 
 

airflow/models.py

@@ -1,3 +1,4 @@
+from __future__ import print_function
 import copy
 from datetime import datetime, timedelta
 import getpass
@@ -842,7 +843,7 @@ def signal_handler(signum, frame):
                     else:
                         task_copy.execute(context=context)
                     task_copy.post_execute(context=context)
-            except (Exception, StandardError, KeyboardInterrupt) as e:
+            except (Exception, KeyboardInterrupt) as e:
                 self.handle_failure(e, test_mode, context)
                 raise
 
@@ -1848,7 +1849,7 @@ def tree_view(self):
         Shows an ascii tree representation of the DAG
         """
         def get_downstream(task, level=0):
-            print (" " * level * 4) + str(task)
+            print((" " * level * 4) + str(task))
             level += 1
             for t in task.upstream_list:
                 get_downstream(t, level)

airflow/operators/sensors.py

@@ -1,3 +1,4 @@
+from __future__ import print_function
 from datetime import datetime
 import logging
 from urlparse import urlparse

airflow/utils.py

@@ -1,3 +1,4 @@
+from __future__ import print_function
 from copy import copy
 from datetime import datetime, timedelta
 from email.mime.text import MIMEText

airflow/www/app.py

@@ -1,3 +1,4 @@
+from __future__ import print_function
 import copy
 from datetime import datetime, timedelta
 import dateutil.parser
@@ -1920,7 +1921,7 @@ def integrate_plugins():
     for v in admin_views:
         admin.add_view(v)
     for bp in flask_blueprints:
-        print bp
+        print(bp)
         app.register_blueprint(bp)
     for ml in menu_links:
         admin.add_link(ml)

requirements.txt

@@ -9,6 +9,7 @@ flask-admin
 flask-cache
 flask-login
 flower
+future
 hive-thrift-py
 ipython[all]
 jinja2
@@ -27,7 +28,7 @@ redis
 requests
 setproctitle
 statsd
-snakebite
+#snakebite -> not compatible with Python 3
 sphinx
 sphinx-argparse
 sphinx_rtd_theme

setup.py

@@ -1,4 +1,5 @@
 from setuptools import setup, find_packages
+import sys
 
 # Kept manually in sync with airflow.__version__
 version = '1.3.0'
@@ -45,6 +46,7 @@
         'flask-cache>=0.13.1',
         'flask-login>=0.2.11',
         'flower>=0.7.3',
+        'future>=0.15.0',
         'jinja2>=2.7.3',
         'markdown>=2.5.2',
         'pandas>=0.15.2',

tests/__init__.py

@@ -1 +1,2 @@
-from core import *
+from __future__ import absolute_import
+from .core import *