Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 12 additions & 6 deletions src/omophub/resources/concepts.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
from __future__ import annotations

from typing import TYPE_CHECKING, Any, TypedDict
from urllib.parse import quote

if TYPE_CHECKING:
from .._request import AsyncRequest, Request
Expand Down Expand Up @@ -56,6 +57,11 @@ class RelationshipsParams(TypedDict, total=False):
page_size: int


def _path_segment(value: str) -> str:
"""Percent-encode an API path segment without preserving slashes."""
return quote(value, safe="")
Comment thread
konstjar marked this conversation as resolved.


class Concepts:
"""Synchronous concepts resource."""

Expand Down Expand Up @@ -108,8 +114,8 @@ def get_by_code(
"""Get a concept by vocabulary and code.

Args:
vocabulary_id: The vocabulary ID (e.g., "SNOMED", "ICD10CM")
concept_code: The concept code within the vocabulary
vocabulary_id: The raw, URL-unencoded vocabulary ID (e.g., "SNOMED", "ICD10CM")
concept_code: The raw, URL-unencoded concept code within the vocabulary
include_relationships: Include related concepts (parents/children)
include_synonyms: Include concept synonyms
include_hierarchy: Include hierarchy information
Expand All @@ -129,7 +135,7 @@ def get_by_code(
params["vocab_release"] = vocab_release

return self._request.get(
f"/concepts/by-code/{vocabulary_id}/{concept_code}",
f"/concepts/by-code/{_path_segment(vocabulary_id)}/{_path_segment(concept_code)}",
params=params or None,
)

Expand Down Expand Up @@ -383,8 +389,8 @@ async def get_by_code(
"""Get a concept by vocabulary and code.

Args:
vocabulary_id: The vocabulary ID (e.g., "SNOMED", "ICD10CM")
concept_code: The concept code within the vocabulary
vocabulary_id: The raw, URL-unencoded vocabulary ID (e.g., "SNOMED", "ICD10CM")
concept_code: The raw, URL-unencoded concept code within the vocabulary
include_relationships: Include related concepts (parents/children)
include_synonyms: Include concept synonyms
include_hierarchy: Include hierarchy information
Expand All @@ -404,7 +410,7 @@ async def get_by_code(
params["vocab_release"] = vocab_release

return await self._request.get(
f"/concepts/by-code/{vocabulary_id}/{concept_code}",
f"/concepts/by-code/{_path_segment(vocabulary_id)}/{_path_segment(concept_code)}",
params=params or None,
)

Expand Down
33 changes: 33 additions & 0 deletions tests/unit/resources/test_concepts.py
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,22 @@ def test_get_concept_by_code(
concept = sync_client.concepts.get_by_code("SNOMED", "44054006")
assert concept["concept_id"] == 201826

@respx.mock
def test_get_concept_by_code_encodes_path_segments(
self, sync_client: OMOPHub, mock_api_response: dict, base_url: str
) -> None:
"""Concept codes containing slashes remain one URL path segment."""
route = respx.get(f"{base_url}/concepts/by-code/ICDO3/8032%2F3").mock(
return_value=Response(200, json=mock_api_response)
)

sync_client.concepts.get_by_code("ICDO3", "8032/3")

assert route.called
assert str(route.calls[0].request.url).endswith(
"/concepts/by-code/ICDO3/8032%2F3"
)

@respx.mock
def test_batch_concepts(
self, sync_client: OMOPHub, base_url: str, mock_concept: dict
Expand Down Expand Up @@ -440,6 +456,23 @@ async def test_async_get_by_code(
concept = await async_client.concepts.get_by_code("SNOMED", "44054006")
assert concept["concept_id"] == 201826

@pytest.mark.asyncio
@respx.mock
async def test_async_get_by_code_encodes_path_segments(
self, async_client: omophub.AsyncOMOPHub, mock_api_response: dict, base_url: str
) -> None:
"""Async concept codes containing slashes remain one URL path segment."""
route = respx.get(f"{base_url}/concepts/by-code/ICDO3/8032%2F3").mock(
return_value=Response(200, json=mock_api_response)
)

await async_client.concepts.get_by_code("ICDO3", "8032/3")

assert route.called
assert str(route.calls[0].request.url).endswith(
"/concepts/by-code/ICDO3/8032%2F3"
)

@pytest.mark.asyncio
@respx.mock
async def test_async_batch_concepts(
Expand Down
Loading