Add Open DNA Collections (#358)

* add Open DNA Collections, closes #357

* fix model version in test file

Author: manulera

poetry.lock

@@ -26,7 +26,7 @@ pydantic = "^2.7.1"
 pandas = "^2.2.3"
 openpyxl = "^3.1.5"
 pyyaml = "^6.0.2"
-opencloning-linkml = "0.4.3"
+opencloning-linkml = "0.4.4"
 primer3-py = "2.2.0"
 biopython = "^1.85"
 packaging = "^25.0"

pyproject.toml

@@ -44,6 +44,7 @@ def parse_bool(value: str) -> bool:
     'http://www.euroscarf.de/',
     'https://wekwikgene.wllsb.edu.cn',
     'http://bahlerweb.cs.ucl.ac.uk',
+    'https://assets.opencloning.org/open-dna-collections',
 ]
 
 if os.environ.get('ALLOWED_EXTERNAL_URLS') is not None:

src/opencloning/app_settings.py

@@ -24,6 +24,7 @@
     SequenceFileFormat,
     SEVASource,
     SequenceLocationStr,
+    OpenDNACollectionsSource,
 )
 from ..dna_functions import (
     format_sequence_genbank,
@@ -234,7 +235,8 @@ def repository_id_http_error_handler(exception: HTTPError, source: RepositoryIdS
             | list[BenchlingUrlSource]
             | list[EuroscarfSource]
             | list[WekWikGeneIdSource]
-            | list[SEVASource],
+            | list[SEVASource]
+            | list[OpenDNACollectionsSource],
             ...,
         ),
         sequences=(list[TextFileSequence], ...),
@@ -249,6 +251,7 @@ async def get_from_repository_id(
         | EuroscarfSource
         | WekWikGeneIdSource
         | SEVASource
+        | OpenDNACollectionsSource
     ),
 ):
     return RedirectResponse(f'/repository_id/{source.repository_name}', status_code=307)
@@ -384,6 +387,22 @@ async def get_from_repository_id_igem(source: IGEMSource):
         repository_id_http_error_handler(exception, source)
 
 
+@router.post(
+    '/repository_id/open_dna_collections',
+    response_model=create_model(
+        'OpenDNACollectionsResponse',
+        sources=(list[OpenDNACollectionsSource], ...),
+        sequences=(list[TextFileSequence], ...),
+    ),
+)
+async def get_from_repository_id_open_dna_collections(source: OpenDNACollectionsSource):
+    try:
+        dseq = (await get_sequences_from_file_url(source.sequence_file_url))[0]
+        return {'sequences': [format_sequence_genbank(dseq, source.output_name)], 'sources': [source]}
+    except HTTPError as exception:
+        repository_id_http_error_handler(exception, source)
+
+
 @router.post(
     '/genome_coordinates',
     response_model=create_model(

src/opencloning/endpoints/external_import.py

@@ -22,7 +22,7 @@ class AllowedExternalUrlsTransport(AsyncHTTPTransport):
     async def handle_async_request(self, request: Request) -> Response:
         if any(str(request.url).startswith(url) for url in allowed_external_urls):
             return await super().handle_async_request(request)
-
+        print('>>>', request.url)
         raise HTTPError(request.url, 403, f'Request to {request.url} is not allowed', None, None)
 
 

src/opencloning/http_client.py

@@ -50,6 +50,7 @@
     CreLoxRecombinationSource as _CreLoxRecombinationSource,
     InVivoAssemblySource as _InVivoAssemblySource,
     SourceInput as _SourceInput,
+    OpenDNACollectionsSource as _OpenDNACollectionsSource,
 )
 from pydna.assembly2 import (
     edge_representation2subfragment_representation,
@@ -183,6 +184,10 @@ def validate_repository_id(self):
         return self
 
 
+class OpenDNACollectionsSource(SourceCommonClass, _OpenDNACollectionsSource):
+    pass
+
+
 class SEVASource(SourceCommonClass, _SEVASource):
     pass
 

src/opencloning/pydantic_models.py

@@ -23,6 +23,7 @@
     IGEMSource,
     WekWikGeneIdSource,
     SEVASource,
+    OpenDNACollectionsSource,
 )
 from opencloning import app_settings, http_client
 
@@ -1044,6 +1045,36 @@ def test_circularize(self):
             self.assertEqual(seq.circular, True)
 
 
+class OpenDNACollectionsSourceTest(unittest.TestCase):
+    def test_valid_url(self):
+        source = OpenDNACollectionsSource(
+            id=0,
+            repository_id='Ecoli Nanobody Toolkit/BC_RJ_SD8',
+            repository_name='open_dna_collections',
+            sequence_file_url='https://assets.opencloning.org/open-dna-collections/Ecoli%20Nanobody%20Toolkit/genbank_seq/BC_RJ_SD8.gb',
+        )
+        response = client.post('/repository_id/open_dna_collections', json=source.model_dump())
+        self.assertEqual(response.status_code, 200)
+        payload = response.json()
+        self.assertEqual(len(payload['sequences']), 1)
+        self.assertEqual(len(payload['sources']), 1)
+        out_source = payload['sources'][0]
+        self.assertEqual(out_source, source.model_dump())
+        seq = read_dsrecord_from_json(TextFileSequence.model_validate(payload['sequences'][0]))
+        self.assertEqual(seq.name, 'BC_RJ_SD8')
+
+    def test_errors(self):
+
+        source = OpenDNACollectionsSource(
+            id=0,
+            repository_id='Ecoli Nanobody Toolkit/BC_RJ_SD8',
+            repository_name='open_dna_collections',
+            sequence_file_url='https://assets.opencloning.org/open-dna-collections/Ecoli%20Nanobody%20Toolkit/genbank_seq/hello.txt',
+        )
+        response = client.post('/repository_id/open_dna_collections', json=source.model_dump())
+        self.assertEqual(response.status_code, 404)
+
+
 class NotAllowedExternalUrlTest(unittest.TestCase):
     def tearDown(self):
         pytest.MonkeyPatch().delenv('ALLOWED_EXTERNAL_URLS', raising=False)

tests/test_endpoints_external_import.py

@@ -139,7 +139,7 @@
   ],
   "description": "Deletion of the ORF of ase1 by homologous recombination in S.pombe",
   "files": null,
-  "schema_version": "0.4.3",
+  "schema_version": "0.4.4",
   "backend_version": null,
   "frontend_version": null
 }