Repository: vogt4nick/dequindre

Issue:
Allow user to define dependencies when DAG is initiated
write a test for it too. 

Discussion:

Patch:
diff --git a/dequindre/__init__.py b/dequindre/__init__.py
index 3914cbf..876b76c 100644
--- a/dequindre/__init__.py
+++ b/dequindre/__init__.py
@@ -4,8 +4,6 @@
 This module defines the Task, DAG, and Dequindre classes. Tasks are intended to
 hold task-level data. DAGs are intended to hold relationships between Tasks.
 Dequindre schedules Tasks in accordance with the DAG(s).
-
-Design is intended
 """
 
 from collections import defaultdict
@@ -14,6 +12,7 @@ from hashlib import md5
 import os
 from typing import Dict, Set
 from subprocess import run as subprocess_run
+from subprocess import check_output, CalledProcessError
 from time import sleep
 
 
@@ -21,28 +20,46 @@ class CyclicGraphError(Exception):
     pass
 
 
+class CondaVersionError(Exception):
+    pass
+
+
+def check_conda():
+    """Verify the machine has a version of conda capable of using `run`. 
+
+    `conda run -n base python 'print("test")'` 
+    """
+    host_version = check_output('conda --version', shell=True).decode().strip()
+    cmd = """ conda run -n base python -c 'print("test...")' """
+    try:
+        check_output(cmd, shell=True)
+    except CalledProcessError:
+        msg = (
+            """Your version of conda does not support the 'conda run' """
+            f"""function. Your machine has {host_version} installed. """
+            """You must upgrade to upgrade to conda >=4.6 to use dequindre."""
+        )
+        raise CondaVersionError(msg)
+
+
 class Task:
     """Defines a Task and its relevant attributes. Tasks with the same loc
     are equal.
 
-    Tasks are instantiated by three variables: loc, stage, and env.
+    Tasks are instantiated by three variables: loc, and env.
 
     Args and Attributes:
         loc (str): location of the python script that runs the task.
-        stage (int, >0): Prioritizes tasks. e.g. Tasks with stage 2 run after
-            all stage 1 tasks have completed.
         env (str): Which environment to run.
     """
-    def __init__(self, loc: str, stage: int, env: str):
+    def __init__(self, loc: str, env: str):
+        # check_conda()
         assert isinstance(loc, str), 'loc must be a str'
         assert len(loc) > 0, 'loc cannot be an empty string'
-        assert isinstance(stage, int), 'stage must be an int'
-        assert stage > 0, 'Stage must be > 0'
         assert isinstance(env, str), 'env must be a str'
         assert len(env) > 0, 'env cannot be an empty string'
 
         self.loc = loc
-        self.stage = stage
         self.env = env
 
         return None
@@ -51,9 +68,8 @@ class Task:
     def __hash__(self):
         """md5 is fast, and chances of colision are really low"""
         loc = self.loc
-        stage = str(self.stage)
         env = self.env
-        hash_str = '-'.join((loc, stage, env))
+        hash_str = '-'.join((loc, env))
         big_int = int(md5(hash_str.encode()).hexdigest(), 16)
         
         return big_int
@@ -72,8 +88,7 @@ class Task:
 
     def __repr__(self):
         """Tasks are idenfied by their loc"""
-        filename = os.path.basename(self.loc)
-        return f"{Task.__qualname__}({filename})"
+        return f"{Task.__qualname__}({self.loc})"
 
 
     def __str__(self):
@@ -100,9 +115,19 @@ class DAG:
         for users.
     """
 
-    def __init__(self):
+    def __init__(self, *, tasks: set = None, dependencies: dict = None):
+        # check_conda()
         self.tasks = set()
         self.edges = defaultdict(set)
+
+        if tasks is not None:
+            assert isinstance(tasks, set), '`tasks` must be a set of tasks'
+            self.add_tasks(tasks)
+
+        if dependencies is not None:
+            assert isinstance(dependencies, dict), '`dependencies` must be a dict'
+            self.add_dependencies(dependencies)
+
         return None
     
 
@@ -123,13 +148,13 @@ class DAG:
         return None
 
 
-    def add_tasks(self, tasks: list):
+    def add_tasks(self, tasks: set):
         """Add multiple tasks to the set of tasks
 
         TODO: Handle iterables
         TODO: Handle non-iterables
         """
-        assert isinstance(tasks, list), TypeError('tasks is not a list')
+        assert isinstance(tasks, set), TypeError('tasks is not a set')
 
         for t in tasks:
             self.add_task(t)
@@ -162,7 +187,7 @@ class DAG:
         assert isinstance(start, Task), TypeError('start is not a dequindre Task')
         assert isinstance(end, Task), TypeError('end is not a dequindre Task')
 
-        self.add_tasks([start, end])
+        self.add_tasks({start, end})
         self.edges[start].add(end)
 
         return None
@@ -191,7 +216,7 @@ class DAG:
         assert isinstance(task, Task), TypeError('start is not a dequindre Task')
         assert isinstance(depends_on, Task), TypeError('end is not a dequindre Task')
 
-        self.add_tasks([task, depends_on])
+        self.add_tasks({task, depends_on})
         self.edges[depends_on].add(task)
 
         # cycles can only be introduced here
@@ -303,7 +328,6 @@ class Dequindre:
 
     Arguments:
         dag (DAG): A DAG of tasks to be scheduled.
-        activate_env_cmd (str): e.g. ". activate", "source activate"
 
     Attributes:
         dag (DAG): A copy of the originally supplied DAG. This attribute is
@@ -314,11 +338,8 @@ class Dequindre:
     TODO: The DAG should catch cycles before they get to Dequindre.
     TODO: Define exception for when cycles are detected
     """
-    def __init__(self, dag: DAG, activate_env_cmd: str):
-        assert isinstance(activate_env_cmd, str), 'activate_env_cmd must be a str'
-        assert len(activate_env_cmd) > 0, 'activate_env_cmd must not be an empty str'
-        self.activate_env_cmd = activate_env_cmd
-
+    def __init__(self, dag: DAG):
+        check_conda()
         self.original_dag = dag
         self.refresh_dag()
 
@@ -336,7 +357,7 @@ class Dequindre:
         return None
 
 
-    def get_task_priorities(self, max_stage) -> Dict[Task, int]:
+    def get_task_priorities(self) -> Dict[Task, int]:
         """Define priority level for each task
 
         Example:
@@ -347,16 +368,7 @@ class Dequindre:
                 drink_tea: 3
             }
         """
-        assert isinstance(max_stage, int), 'max_stage must be an int'
-        assert max_stage > 1, 'max_stage must be greater than 1'
-        
         dag = self.dag  # copy to something easier to read
-
-        too_high_stage = set([t for t in dag.tasks if t.stage > max_stage])
-
-        for task in too_high_stage:
-            dag.remove_task(task)
-
         task_priority = defaultdict(int)
         visited = set()
         i = 1
@@ -373,7 +385,7 @@ class Dequindre:
         return task_priority
 
 
-    def get_priorities(self, max_stage) -> Dict[int, Set[Task]]:
+    def get_priorities(self) -> Dict[int, Set[Task]]:
         """Define tasks for each priority level.
 
         Example:
@@ -385,7 +397,7 @@ class Dequindre:
             }
         """
         priorities = defaultdict(set)
-        task_priorities = self.get_task_priorities(max_stage=max_stage)
+        task_priorities = self.get_task_priorities()
         for k, v in task_priorities.items():
             priorities[v].add(k)
 
@@ -401,16 +413,16 @@ class Dequindre:
 
         print(f'\nRunning {repr(task)}\n')
         r = subprocess_run(
-            f'{task.env} {task.loc}',
+            f'conda run -n {task.env} python {task.loc}',
             shell=True, check=True)
 
         return None
 
 
-    def run_tasks(self, max_stage: int = 10000):
-        """Run all tasks on the DAG that are less than max_stage"""
+    def run_tasks(self):
+        """Run all tasks on the DAG"""
         self.refresh_dag()  # refresh just in case
-        priorities = self.get_priorities(max_stage=max_stage)
+        priorities = self.get_priorities()
 
         for k, tasks in priorities.items():
             for task in tasks:
diff --git a/setup.py b/setup.py
index 14f4d96..73c4fac 100644
--- a/setup.py
+++ b/setup.py
@@ -2,7 +2,7 @@ from setuptools import setup, find_packages
 
 setup(
     name='dequindre',
-    version='0.2.0',
+    version='0.3.0',
     description="Dequindre: a lightweight scheduler.",
     long_description=open('readme.md').read(),
     url='https://github.com/vogt4nick/dequindre',


Test patch:
diff --git a/dequindre/tests/test_DAG.py b/dequindre/tests/test_DAG.py
index 9e52298..7efa2d0 100644
--- a/dequindre/tests/test_DAG.py
+++ b/dequindre/tests/test_DAG.py
@@ -10,15 +10,15 @@ from dequindre import Task, DAG, CyclicGraphError
 
 def get_two_tasks():
     return (
-        Task('A.py', stage=1, env='test-env'),
-        Task('B.py', stage=1, env='test-env')
+        Task('A.py', env='test-env'),
+        Task('B.py', env='test-env')
     )
 
 
 def get_cyclic_graph():
-    A = Task('A.py', stage=1, env='test-env')
-    B = Task('B.py', stage=1, env='test-env')
-    C = Task('C.py', stage=1, env='test-env')
+    A = Task('A.py', env='test-env')
+    B = Task('B.py', env='test-env')
+    C = Task('C.py', env='test-env')
 
     dag = DAG()
     dag.add_edges({
@@ -34,11 +34,16 @@ def get_cyclic_graph():
 # ----------------------------------------------------------------------------
 @pytest.mark.run(order=2)
 def test__DAG_init():
-    assert isinstance(DAG(), DAG), 'DAG init failed'
+    DAG()
+
+    # init with dependencies
+    make_tea = Task('make_tea.py', 'test-env')
+    drink_tea = Task('drink_tea.py', 'test-env')
+    DAG(dependencies={drink_tea: make_tea})
 
 
 def test__DAG_repr():
-    make_tea = Task('make_tea.py', 1, 'test-env')
+    make_tea = Task('make_tea.py', 'test-env')
     dag = DAG()
     dag.add_task(make_tea)
     assert repr(dag) == "DAG({Task(make_tea.py)})"
@@ -61,7 +66,7 @@ def test__DAG_add_tasks():
     A, B = get_two_tasks()
 
     dag = DAG()
-    dag.add_tasks([A, B])
+    dag.add_tasks({A, B})
 
     assert dag.tasks == {A,B}, 'Test Tasks were not added to the DAG'
 
@@ -71,7 +76,7 @@ def test__DAG_remove_task():
     A, B = get_two_tasks()
 
     dag = DAG()
-    dag.add_tasks([A, B])
+    dag.add_tasks({A, B})
     dag.remove_task(A)
 
     assert dag.tasks == {B}, 'Test Task was not added to the DAG'
@@ -84,7 +89,7 @@ def test__DAG_add_edge():
     A, B = get_two_tasks()
 
     dag = DAG()
-    dag.add_tasks([A, B])
+    dag.add_tasks({A, B})
     dag.add_edge(A, B)
     assert dag.edges == {A: {B,}}, 'edge was not created'
 
@@ -121,7 +126,7 @@ def test__DAG_add_dependency_detect_cycle():
 
 def test__DAG_add_dependencies():
     A, B = get_two_tasks()
-    C = Task('C.py', stage=1, env='test-env')
+    C = Task('C.py', env='test-env')
     dag = DAG()
     dag.add_dependencies({B: A})
     assert dag.edges[A] == set([B])
@@ -134,7 +139,7 @@ def test__DAG_add_dependencies():
 
 def test__DAG_add_dependency_detect_cycle():
     A, B = get_two_tasks()
-    C = Task('C.py', stage=1, env='test-env')
+    C = Task('C.py', env='test-env')
 
     dag = DAG()
     with pytest.raises(CyclicGraphError):
diff --git a/dequindre/tests/test_Dequindre.py b/dequindre/tests/test_Dequindre.py
index 4906827..a48eb58 100644
--- a/dequindre/tests/test_Dequindre.py
+++ b/dequindre/tests/test_Dequindre.py
@@ -11,24 +11,15 @@ from dequindre import Task, DAG, Dequindre
 def test__Dequindre_init_exceptions():
     """Raise expected exceptions
     """
-    A = Task('A.py', 1, 'test-env')
-    B = Task('B.py', 1, 'test-env')
-    C = Task('C.py', 1, 'test-env')
+    A = Task('A.py', 'test-env')
+    B = Task('B.py', 'test-env')
+    C = Task('C.py', 'test-env')
 
     dag = DAG()
-    dag.add_tasks([A, B, C])
+    dag.add_tasks({A, B, C})
 
     with pytest.raises(TypeError):
         Dequindre()
-    with pytest.raises(TypeError):
-        Dequindre(dag)
-    with pytest.raises(AssertionError, match='activate_env_cmd must be a str'):
-        Dequindre(dag, activate_env_cmd=1)
-    with pytest.raises(AssertionError, match='activate_env_cmd must be a str'):
-        Dequindre(dag, activate_env_cmd=None)
-    with pytest.raises(AssertionError,
-        match='activate_env_cmd must not be an empty str'):
-        Dequindre(dag, activate_env_cmd='')
 
     return None
 
@@ -37,32 +28,32 @@ def test__Dequindre_init_exceptions():
 def test__Dequindre_init():
     """Nothing should break here
     """
-    A = Task('A.py', 1, 'test-env')
-    B = Task('B.py', 1, 'test-env')
-    C = Task('C.py', 1, 'test-env')
+    A = Task('A.py', 'test-env')
+    B = Task('B.py', 'test-env')
+    C = Task('C.py', 'test-env')
 
     dag = DAG()
-    dag.add_tasks([A, B, C])
-    dq = Dequindre(dag, activate_env_cmd='. activate')
+    dag.add_tasks({A, B, C})
+    dq = Dequindre(dag)
 
     return None
 
 
 def test__Dequindre_repr():
-    make_tea = Task('make_tea.py', 1, 'test-env')
+    make_tea = Task('make_tea.py', 'test-env')
     dag = DAG()
     dag.add_task(make_tea)
-    dq = Dequindre(dag, 'activate')
+    dq = Dequindre(dag)
     assert repr(dq) == "Dequindre(DAG({Task(make_tea.py)}))"
 
 
 def test__Dequindre_refresh_dag():
-    A = Task('A.py', 1, 'test-env')
-    B = Task('B.py', 1, 'test-env')
-    C = Task('C.py', 1, 'test-env')
+    A = Task('A.py', 'test-env')
+    B = Task('B.py', 'test-env')
+    C = Task('C.py', 'test-env')
     dag = DAG()
-    dag.add_tasks([A, B, C])
-    dq = Dequindre(dag, activate_env_cmd='. activate')
+    dag.add_tasks({A, B, C})
+    dq = Dequindre(dag)
 
     tasks = sorted(list(dq.dag.tasks))
     for t in tasks:
@@ -78,31 +69,16 @@ def test__Dequindre_refresh_dag():
 
 
 def test__Dequindre_get_task_priorities():
-    A = Task('A.py', 1, 'test-env')
-    dag = DAG()
-    dag.add_task(A)
-    dq = Dequindre(dag, activate_env_cmd='. activate')
-
-    with pytest.raises(AssertionError, match='max_stage must be an int'):
-        dq.get_task_priorities(max_stage='prod')
-    with pytest.raises(AssertionError, 
-        match='max_stage must be greater than 1'):
-        dq.get_task_priorities(max_stage=0)
-
-    return None
-
-
-def test__Dequindre_get_task_priorities():
-    A = Task('A.py', 1, 'test-env')
-    B = Task('B.py', 1, 'test-env')
-    C = Task('C.py', 1, 'test-env')
-    Z = Task('Z.py', 1, 'test-env')
+    A = Task('A.py', 'test-env')
+    B = Task('B.py', 'test-env')
+    C = Task('C.py', 'test-env')
+    Z = Task('Z.py', 'test-env')
     dag = DAG()
-    dag.add_tasks([A, B, C, Z])
+    dag.add_tasks({A, B, C, Z})
     dag.add_edges({A:B, B:C})
-    dq = Dequindre(dag, activate_env_cmd='. activate')
+    dq = Dequindre(dag)
 
-    priorities = dq.get_task_priorities(100)
+    priorities = dq.get_task_priorities()
 
     testable = {hash(k): v for k, v in priorities.items()}
     assert testable == {
@@ -113,32 +89,17 @@ def test__Dequindre_get_task_priorities():
     }
 
 
-def test__Dequindre_get_priorities_exceptions():
-    A = Task('A.py', 1, 'test-env')
-    dag = DAG()
-    dag.add_task(A)
-    dq = Dequindre(dag, activate_env_cmd='. activate')
-
-    with pytest.raises(AssertionError, match='max_stage must be an int'):
-        dq.get_priorities(max_stage='prod')
-    with pytest.raises(AssertionError, 
-        match='max_stage must be greater than 1'):
-        dq.get_priorities(max_stage=0)
-
-    return None
-
-
 def test__Dequindre_get_priorities():
-    A = Task('A.py', 1, 'test-env')
-    B = Task('B.py', 1, 'test-env')
-    C = Task('C.py', 1, 'test-env')
-    Z = Task('Z.py', 1, 'test-env')
+    A = Task('A.py', 'test-env')
+    B = Task('B.py', 'test-env')
+    C = Task('C.py', 'test-env')
+    Z = Task('Z.py', 'test-env')
     dag = DAG()
-    dag.add_tasks([A, B, C, Z])
+    dag.add_tasks({A, B, C, Z})
     dag.add_edges({A:B, B:C})
-    dq = Dequindre(dag, activate_env_cmd='. activate')
+    dq = Dequindre(dag)
 
-    priorities = dq.get_priorities(100)
+    priorities = dq.get_priorities()
     testable = {}
 
     # build a testable result dict
diff --git a/dequindre/tests/test_Task.py b/dequindre/tests/test_Task.py
index b69e45b..c2f77a3 100644
--- a/dequindre/tests/test_Task.py
+++ b/dequindre/tests/test_Task.py
@@ -8,43 +8,33 @@ from dequindre import Task
 @pytest.mark.run(order=1)
 def test__Task_init():
     good_loc = 'path/to/file.py'
-    good_stage = 1
     good_env = 'test-env'
 
     assert isinstance(
-        Task(loc=good_loc, stage=good_stage, env=good_env),
+        Task(loc=good_loc, env=good_env),
         Task
     )
     with pytest.raises(AssertionError):
-        Task(loc=None, stage=good_stage, env=good_env)
+        Task(loc=None, env=good_env)
 
     with pytest.raises(AssertionError):
-        Task(loc='', stage=good_stage, env=good_env)
+        Task(loc='', env=good_env)
 
     with pytest.raises(AssertionError):
-        Task(loc=good_loc, stage=None, env=good_env)
+        Task(loc=good_loc, env=None)
 
     with pytest.raises(AssertionError):
-        Task(loc=good_loc, stage='prod', env=good_env)
-
-    with pytest.raises(AssertionError):
-        Task(loc=good_loc, stage=0, env=good_env)
-
-    with pytest.raises(AssertionError):
-        Task(loc=good_loc, stage=good_stage, env=None)
-
-    with pytest.raises(AssertionError):
-        Task(loc=good_loc, stage=good_stage, env='')
+        Task(loc=good_loc, env='')
 
 
 def test__Task_repr():
-    make_tea = Task('make_tea.py', 1, 'test-env')
+    make_tea = Task('make_tea.py', 'test-env')
     assert repr(make_tea) == "Task(make_tea.py)"
 
 
 def test__Task_hash():
-    A = Task('test.py', 1, 'test-env')
-    B = Task('test.py', 1, 'test-env')
+    A = Task('test.py', 'test-env')
+    B = Task('test.py', 'test-env')
 
     assert hash(A) == hash(B)
 
@@ -54,8 +44,8 @@ def test__Task_hash():
 
 
 def test__Task_eq():
-    A = Task('test.py', 1, 'test-env')
-    B = Task('test.py', 1, 'test-env')
+    A = Task('test.py', 'test-env')
+    B = Task('test.py', 'test-env')
 
     assert A == B