ENH: First attempt at tackling biocore#77

labman/db/plate.py

@@ -536,3 +536,60 @@ def get_previously_plated_wells(self):
                         res[well].append(plate)
             res = {well: list(set(plates)) for well, plates in res.items()}
         return res
+
+    def upstream(self):
+        """ Get all of the other plates that share a common composition id.
+
+        Returns
+        -------
+        list of plates
+            Plates that share a common composition id.
+        """
+        plates = []
+        compositions = self._upstream_compositions()
+        for c in compositions:
+            plates += Plate._downstream_plates(c)
+        plates = list(set(plates))
+        return plates
+
+    def _upstream_compositions(self):
+        """ Get all of the compositions that share a common plate id.
+
+        Returns
+        -------
+        list of compositions
+            Compositions that contain the specified plate.
+        """
+        plates = []
+        with sql_connection.TRN as TRN:
+            sql = """SELECT DISTINCT composition_id
+                     FROM qiita.composition
+                          JOIN qiita.well USING (container_id)
+                     WHERE plate_id = %s"""
+            TRN.add(sql, [self.id])
+            compositions = TRN.execute_fetchflatten()
+        return compositions
+
+    @staticmethod
+    def _downstream_plates(composition_id):
+        """ Get all plates under a specified composition id.
+
+        Parameters
+        ----------
+        composition_id : int
+           A specified composition id.
+
+        Returns
+        -------
+        list of processes
+            Plates that are derived from the given composition.
+        """
+        plates = []
+        with sql_connection.TRN as TRN:
+            sql = """SELECT DISTINCT plate_id
+                     FROM qiita.well
+                          JOIN qiita.composition USING (container_id)
+                     WHERE composition_id = %s"""
+            TRN.add(sql, [composition_id])
+            plates = TRN.execute_fetchflatten()
+            return plates

labman/db/tests/test_plate.py

@@ -41,6 +41,7 @@ def test_create(self):
 
 
 class TestPlate(LabmanTestCase):
+
     def test_search(self):
         with self.assertRaises(ValueError):
             Plate.search()
@@ -396,6 +397,12 @@ def test_get_previously_plated_wells(self):
         obs = tester.get_previously_plated_wells()
         self.assertEqual(obs, exp)
 
+    def test_upstream(self):
+        tester = Plate(21)
+        res = tester.upstream()
+        exp = [21]
+        self.assertListEqual(res, exp)
+
 
 if __name__ == '__main__':
     main()