In restriction_and_ligation, add option to sort products based on presence of the first enzyme cut site

src/opencloning/endpoints/assembly.py

@@ -279,6 +279,14 @@ async def restriction_and_ligation(
     source: RestrictionAndLigationSource,
     sequences: Annotated[list[TextFileSequence], Field(min_length=1)],
     circular_only: bool = Query(False, description='Only return circular assemblies.'),
+    sort_by_recognition_sites: bool = Query(
+        False,
+        description='''
+        Sort the products by the number of recognition sites from the first enzyme.
+        This is useful for Golden Gate assembly, where you tipically want the product that lost
+        the Type IIS recognition site.
+        ''',
+    ),
 ):
 
     fragments = [read_dsrecord_from_json(seq) for seq in sequences]
@@ -290,6 +298,10 @@ async def restriction_and_ligation(
     except ValueError as e:
         raise HTTPException(400, *e.args)
 
+    if len(enzymes) > 0 and sort_by_recognition_sites:
+        enzyme = parse_restriction_enzymes([source.restriction_enzymes[0]])
+        products.sort(key=lambda x: len(x.seq.get_cutsites(enzyme)))
+
     return format_products(
         source.id,
         products,

tests/test_endpoints_assembly.py

@@ -903,6 +903,48 @@ def test_too_many_assemblies(self):
         self.assertEqual(response.status_code, 400)
         self.assertTrue('Too many assemblies' in response.json()['detail'])
 
+    def test_sort_by_recognition_sites(self):
+        fragments = [
+            Dseqrecord('AAggtctcaACGTagagtcacacaggactactaCGTAagagaccAA', circular=True),
+            Dseqrecord('AAggtctcaCGTAagagtcacacaggactactaACGTagagaccAA', circular=True),
+        ]
+
+        source = RestrictionAndLigationSource(
+            id=0,
+            restriction_enzymes=['BsaI'],
+        )
+        json_fragments = [format_sequence_genbank(f) for f in fragments]
+        for i, f in enumerate(json_fragments):
+            f.id = i + 1
+
+        for sorting in [True, False]:
+            data = {'source': source.model_dump(), 'sequences': [f.model_dump() for f in json_fragments]}
+            response = client.post(
+                '/restriction_and_ligation', json=data, params={'sort_by_recognition_sites': sorting}
+            )
+            self.assertEqual(response.status_code, 200)
+            payload = response.json()
+            seqs1 = [read_dsrecord_from_json(TextFileSequence.model_validate(s)) for s in payload['sequences']]
+
+            shifted_fragment = fragments[0].shifted(21)
+            json_shifted_fragment = format_sequence_genbank(shifted_fragment)
+            json_shifted_fragment.id = 1
+            data['sequences'][0] = json_shifted_fragment.model_dump()
+            data['sequences'] = data['sequences']
+            response = client.post(
+                '/restriction_and_ligation', json=data, params={'sort_by_recognition_sites': sorting}
+            )
+            self.assertEqual(response.status_code, 200)
+            payload = response.json()
+            seqs2 = [read_dsrecord_from_json(TextFileSequence.model_validate(s)) for s in payload['sequences']]
+
+            if sorting:
+                self.assertEqual(len(seqs1[0]), len(seqs2[0]))
+                self.assertEqual(len(seqs1[1]), len(seqs2[1]))
+            else:
+                self.assertNotEqual(len(seqs1[0]), len(seqs2[0]))
+                self.assertNotEqual(len(seqs1[1]), len(seqs2[1]))
+
 
 class CrisprTest(unittest.TestCase):