From 509d10fddb8e147be770720460181990b47e456d Mon Sep 17 00:00:00 2001 From: aditiadhikari05 Date: Wed, 26 Aug 2026 16:04:37 -0500 Subject: [PATCH] Add ariadne/ariadnePy tutorial page --- src/ariadnepy/graph/_weave.py | 9 ++++++++- tests/test_core/test_graph/test_weave.py | 25 ++++++++++++++++++++++++ 2 files changed, 33 insertions(+), 1 deletion(-) diff --git a/src/ariadnepy/graph/_weave.py b/src/ariadnepy/graph/_weave.py index 6a3fa04..fb780c3 100644 --- a/src/ariadnepy/graph/_weave.py +++ b/src/ariadnepy/graph/_weave.py @@ -303,6 +303,12 @@ def _strip_iri(ids: Sequence[str], name: str) -> list[str]: _KEGG_REST = "https://rest.kegg.jp" _KEGG_EXT_DBS = {"chebi", "geneid", "proteinid", "pubchem", "uniprotkb"} +# KEGG's /link endpoint rejects "ec" as a target db (HTTP 400) when the +# source is a whole database rather than specific entries, e.g. +# /link/ec/ko fails but /link/enzyme/ko succeeds. "enzyme" is accepted in +# both positions and returned rows are still labelled "ec:...", so this +# swap is URL-only and doesn't affect downstream parsing. +_KEGG_LINK_TARGET_ALIASES = {"ec": "enzyme"} def _fetch_kegg_edge( @@ -333,7 +339,8 @@ def _fetch_kegg_edge( for chunk_start in range(0, max(1, len(query_targets)), 100): chunk = query_targets[chunk_start : chunk_start + 100] query = "+".join(chunk) if len(chunk) > 1 else chunk[0] - url = f"{_KEGG_REST}/{endpoint}/{to}/{query}" + url_target = _KEGG_LINK_TARGET_ALIASES.get(to, to) + url = f"{_KEGG_REST}/{endpoint}/{url_target}/{query}" resp = _requests.get(url, timeout=30) resp.raise_for_status() for line in resp.text.strip().splitlines(): diff --git a/tests/test_core/test_graph/test_weave.py b/tests/test_core/test_graph/test_weave.py index 24fcb95..fdf74a5 100644 --- a/tests/test_core/test_graph/test_weave.py +++ b/tests/test_core/test_graph/test_weave.py @@ -11,6 +11,7 @@ from ariadnepy.exceptions import AriadneError from ariadnepy.graph._weave import ( _draw_path, + _fetch_kegg_edge, _get_sorted_edge_key, _graph_from_path_df, _map_complex_modules, @@ -587,3 +588,27 @@ def test_map_complex_missing_table_raises(): """Missing required linkmap tables raises before scipy is touched.""" with pytest.raises(AriadneError, match="Missing required"): _map_complex_modules({"origin2feature": pd.DataFrame()}, "origin", "feature") + + +# ── _fetch_kegg_edge — target db aliasing ───────────────────────────────────── + + +def test_fetch_kegg_edge_aliases_ec_target_to_enzyme(): + """KEGG's /link endpoint 400s on 'ec' as a target db for whole-db linking + (e.g. /link/ec/ko) but accepts 'enzyme' — same rows, both labelled 'ec:'. + The request URL must use the 'enzyme' alias whenever the target is 'ec'. + """ + step = pd.Series({ + "specFrom": "ko", "specTo": "ec", + "initFrom": "ko", "initTo": "ec", "source": "KEGG", + }) + mock_resp = type("R", (), { + "text": "ko:K00001\tec:1.1.1.1", + "raise_for_status": lambda self: None, + })() + with patch("ariadnepy.graph._weave._requests.get", return_value=mock_resp) as mock_get: + df = _fetch_kegg_edge(step, None) + called_url = mock_get.call_args[0][0] + assert called_url.endswith("/link/enzyme/ko") + assert "/link/ec/" not in called_url + assert list(df.columns) == ["ko", "ec"]