Add recombinase endpoint

pyproject.toml

@@ -29,7 +29,7 @@ primer3-py = "^2.3"
 biopython = {git = "https://github.com/manulera/biopython", rev = "2b4e11f0d48ef593f18cba9bf0fe8e99bb6e5bcf"}
 packaging = "^25.0"
 pairwise-alignments-to-msa = "^0.1.1"
-pydna = {git = "https://github.com/pydna-group/pydna", rev = "1ef52e4ba398bf4ea05d5ad4b8c496ef65fe7be0"}
+pydna = {git = "https://github.com/pydna-group/pydna", rev = "b567985784744768579f6c1fad20bd4b70ee29ba"}
 
 [tool.poetry.group.dev.dependencies]
 autopep8 = "^2.0.4"

src/opencloning/endpoints/assembly.py

@@ -1,5 +1,6 @@
 from fastapi import Query, HTTPException
 from typing import Union
+import copy
 from pydna.dseqrecord import Dseqrecord
 from pydna.primer import Primer as PydnaPrimer
 from pydantic import create_model, Field
@@ -25,6 +26,7 @@
     GatewaySource,
     Primer as PrimerModel,
     TextFileSequence,
+    RecombinaseSource,
 )
 
 from pydna.assembly2 import (
@@ -40,10 +42,13 @@
     crispr_integration as _crispr_integration,
     cre_lox_integration as _cre_lox_integration,
     cre_lox_excision as _cre_lox_excision,
+    recombinase_integration as _recombinase_integration,
+    recombinase_excision as _recombinase_excision,
 )
 from pydna.cre_lox import annotate_loxP_sites
 
 from pydna.gateway import annotate_gateway_sites
+from pydna.recombinase import RecombinaseCollection, Recombinase
 from ..get_router import get_router
 
 router = get_router()
@@ -377,3 +382,48 @@ async def cre_lox_recombination(
         source.output_name,
         no_products_error_message='No compatible Cre/Lox recombination was found.',
     )
+
+
+@router.post(
+    '/recombinase',
+    response_model=create_model(
+        'RecombinaseResponse',
+        sources=(list[RecombinaseSource], ...),
+        sequences=(list[TextFileSequence], ...),
+    ),
+)
+async def recombinase(
+    source: RecombinaseSource,
+    sequences: Annotated[list[TextFileSequence], Field(min_length=1)],
+    reverse_recombinase: bool = Query(False, description='Whether to use the reverse reaction of the recombinase.'),
+):
+    fragments = [read_dsrecord_from_json(seq) for seq in sequences]
+    completed_source = source if is_assembly_complete(source) else None
+    try:
+        collection = RecombinaseCollection([Recombinase(**r.model_dump()) for r in source.recombinases])
+    except ValueError as e:
+        raise HTTPException(422, *e.args)
+
+    reverse_collection = copy.deepcopy(collection)
+    reverse_collection.recombinases.extend([r.get_reverse_recombinase() for r in reverse_collection.recombinases])
+    if reverse_recombinase:
+        collection = reverse_collection
+
+    if len(fragments) == 1:
+        products = _recombinase_excision(fragments[0], collection)
+    else:
+        products = []
+        if not fragments[0].circular:
+            products.extend(_recombinase_integration(fragments[0], fragments[1:], collection))
+        if not fragments[1].circular:
+            products.extend(_recombinase_integration(fragments[1], fragments[:1], collection))
+
+    products = [reverse_collection.annotate(p) for p in products]
+
+    return format_products(
+        source.id,
+        products,
+        completed_source,
+        source.output_name,
+        no_products_error_message='No compatible reaction was found with the provided recombinases.',
+    )

tests/test_endpoints_assembly.py

@@ -21,6 +21,8 @@
     CRISPRSource,
     GatewaySource,
     CreLoxRecombinationSource,
+    RecombinaseSource,
+    Recombinase,
 )
 
 
@@ -1271,7 +1273,7 @@ def test_gateway_source(self):
         payload = response.json()
         self.assertIn('Inputs are not compatible for LR reaction', payload['detail'])
         self.assertIn('fragment 1: attB1', payload['detail'])
-        self.assertTrue(payload['detail'].endswith('fragment 2: attB1, attL1, attR1, attP1'))
+        self.assertTrue(payload['detail'].endswith('fragment 2: attB1, attP1, attL1, attR1'))
 
     def test_only_multi_site(self):
         attB1 = self.attB1
@@ -1437,3 +1439,130 @@ def test_no_results(self):
         self.assertEqual(response.status_code, 400)
         payload = response.json()
         self.assertIn('No compatible Cre/Lox', payload['detail'])
+
+
+class RecombinaseTest(unittest.TestCase):
+
+    def test_recombinase(self):
+        site1 = 'ATGCCCTAAaaCT'
+        site2 = 'CAaaTTTTTTTCCCT'
+
+        genome = Dseqrecord(f"cccccc{site1.upper()}aaaaa")
+        insert = Dseqrecord(f"{site2.upper()}bbbbb", circular=True)
+        fragments = [format_sequence_genbank(genome), format_sequence_genbank(insert)]
+        fragments[0].id = 1
+        fragments[1].id = 2
+        rec = Recombinase(
+            name='blah',
+            site1=site1,
+            site2=site2,
+        )
+
+        source = RecombinaseSource(id=0, recombinases=[rec])
+        data = {
+            'source': source.model_dump(),
+            'sequences': [f.model_dump() for f in fragments],
+        }
+        response = client.post('/recombinase', json=data)
+        self.assertEqual(response.status_code, 200)
+        payload = response.json()
+        self.assertEqual(len(payload['sources']), 1)
+
+        # We can do the reverse reaction
+        data = {
+            'source': source.model_dump(),
+            'sequences': payload['sequences'],
+        }
+        response = client.post('/recombinase', json=data, params={'reverse_recombinase': True})
+        self.assertEqual(response.status_code, 200)
+        payload = response.json()
+        self.assertEqual(len(payload['sources']), 2)
+
+    def test_recombinase_integration_two_site_pairs(self):
+        site1 = 'AAaaTTC'
+        site2 = 'CCaaGC'
+        site3 = 'GAccACC'
+        site4 = 'TCccAAC'
+        rec1 = Recombinase(
+            site1=site1,
+            site2=site2,
+            site1_name='s1',
+            site2_name='s2',
+        )
+        rec2 = Recombinase(
+            site1=site3,
+            site2=site4,
+            site1_name='s1',
+            site2_name='s2',
+        )
+        source = RecombinaseSource(id=0, recombinases=[rec1, rec2])
+        seq = Dseqrecord(f"ggg{site1}aaa{site3}ttt")
+        seq2 = Dseqrecord(f"ccc{site2}ttt{site4}aaa")
+        fragments = [format_sequence_genbank(seq), format_sequence_genbank(seq2)]
+        fragments[0].id = 1
+        fragments[1].id = 2
+        data = {
+            'source': source.model_dump(),
+            'sequences': [f.model_dump() for f in fragments],
+        }
+        response = client.post('/recombinase', json=data)
+        self.assertEqual(response.status_code, 200)
+        payload = response.json()
+        resulting_sequences = [
+            read_dsrecord_from_json(TextFileSequence.model_validate(s)) for s in payload['sequences']
+        ]
+        self.assertEqual(len(resulting_sequences), 2)
+        self.assertEqual(str(resulting_sequences[0].seq).upper(), 'GGGAAAAGCTTTTCCCACCTTT')
+        self.assertEqual(str(resulting_sequences[1].seq).upper(), 'CCCCCAATTCAAAGACCAACAAA')
+        # The number of recombinases returned is the same:
+        source_recombinases = [Recombinase.model_validate(s) for s in payload['sources'][0]['recombinases']]
+        self.assertEqual(len(source_recombinases), 2)
+        self.assertEqual(source_recombinases[0], rec1)
+        self.assertEqual(source_recombinases[1], rec2)
+
+        # The same reaction again does not work
+        data = {
+            'source': source.model_dump(),
+            'sequences': payload['sequences'],
+        }
+        response2 = client.post('/recombinase', json=data)
+        self.assertEqual(response2.status_code, 400)
+        payload2 = response2.json()
+        self.assertIn('No compatible reaction was found with the provided recombinases.', payload2['detail'])
+
+        # The reverse does, and regenerates the original sequences
+        data = {
+            'source': source.model_dump(),
+            'sequences': payload['sequences'],
+        }
+        response = client.post('/recombinase', json=data, params={'reverse_recombinase': True})
+        self.assertEqual(response.status_code, 200)
+        payload = response.json()
+        resulting_sequences = [
+            read_dsrecord_from_json(TextFileSequence.model_validate(s)) for s in payload['sequences']
+        ]
+        self.assertEqual(len(resulting_sequences), 2)
+        self.assertEqual(str(resulting_sequences[0].seq).upper(), str(seq.seq).upper())
+        self.assertEqual(str(resulting_sequences[1].seq).upper(), str(seq2.seq).upper())
+
+    def test_recombinase_validation(self):
+        site1 = 'AAaaTTC'
+        site2 = 'CCggGC'
+
+        data = {
+            'source': {'id': 0, 'recombinases': [{'site1': site1, 'site2': site2}]},
+            'sequences': [format_sequence_genbank(Dseqrecord(f"ggg{site1}aaa")).model_dump()],
+        }
+        response = client.post('/recombinase', json=data)
+        self.assertEqual(response.status_code, 422)
+        payload = response.json()
+        self.assertIn('Recombinase recognition sites do not have matching homology cores', payload['detail'])
+
+        site1 = 'AAAAAAA'
+
+        data = {
+            'source': {'id': 0, 'recombinases': [{'site1': site1, 'site2': site2}]},
+            'sequences': [format_sequence_genbank(Dseqrecord(f"ggg{site1}aaa")).model_dump()],
+        }
+        response = client.post('/recombinase', json=data)
+        self.assertEqual(response.status_code, 422)