feat: Integrate oehrpy SDK for type-safe EHRBase composition building #27
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,170 +1,146 @@ | ||
| """EHRBase client wrapper using oehrpy SDK. | ||
|
|
||
| Wraps the oehrpy EHRBaseClient to provide a consistent interface | ||
| for Open CIS, with logging and error handling. | ||
| """ | ||
|
|
||
| from typing import Any | ||
|
|
||
| from openehr_sdk.client import EHRBaseClient as OehrpyClient | ||
| from openehr_sdk.client import EHRBaseError | ||
|
|
||
| from src.config import settings | ||
|
|
||
|
|
||
| class EHRBaseClient: | ||
| """Async EHRBase client backed by oehrpy SDK.""" | ||
|
|
||
| def __init__(self) -> None: | ||
| self.base_url = settings.ehrbase_url | ||
| self._client: OehrpyClient | None = None | ||
|
|
||
| async def _ensure_connected(self) -> OehrpyClient: | ||
| if self._client is None: | ||
| self._client = OehrpyClient( | ||
| base_url=settings.ehrbase_url, | ||
| username=settings.ehrbase_user, | ||
| password=settings.ehrbase_password, | ||
| ) | ||
| await self._client.connect() | ||
| return self._client | ||
|
Comment on lines
+22
to
30
There was a problem hiding this comment. 🧩 Analysis chain🏁 Script executed: # Find and examine the ehrbase client file
find . -type f -name "client.py" | grep -E "api|ehrbase"Repository: platzhersh/open-cis Length of output: 113 🏁 Script executed: # Examine the settings to see if timeout/verify_ssl are defined
find . -type f -name "settings.py" | head -10Repository: platzhersh/open-cis Length of output: 45 🏁 Script executed: # Check imports in the ehrbase client file
head -40 api/src/ehrbase/client.py 2>/dev/null || find . -path "*/api/*/ehrbase/client.py" -exec head -40 {} \;Repository: platzhersh/open-cis Length of output: 1405 🌐 Web query:
💡 Result: I couldn’t find any public docs/source for That said, many OpenAPI-generated Python clients expose the same constructor kwargs and use them to build an internal
And If you paste the Sources: [1] [2] 🏁 Script executed: # Find and examine settings/config file
find . -type f -name "settings.py" -o -name "config.py" | head -10Repository: platzhersh/open-cis Length of output: 82 🏁 Script executed: # Search for ehrbase_timeout and ehrbase_verify_ssl in the codebase
rg "ehrbase_timeout|ehrbase_verify_ssl" --type pyRepository: platzhersh/open-cis Length of output: 168 🏁 Script executed: # Check for requirements, pyproject.toml, or setup files for openehr_sdk version
find . -type f \( -name "requirements*.txt" -o -name "pyproject.toml" -o -name "setup.py" -o -name "setup.cfg" -o -name "poetry.lock" \) | head -10Repository: platzhersh/open-cis Length of output: 83 🌐 Web query:
💡 Result: I couldn’t find any So there are no If your goal is to control timeouts / TLS verification with EHRbase:
If you meant a different library actually named
Citations:
🏁 Script executed: # Read the config file to see all settings
cat -n api/src/config.pyRepository: platzhersh/open-cis Length of output: 963 🏁 Script executed: # Check pyproject.toml for openehr_sdk version
cat -n api/pyproject.toml | head -50Repository: platzhersh/open-cis Length of output: 1476 🌐 Web query:
💡 Result:
Sources: 🌐 Web query:
💡 Result: I couldn’t find any PyPI package named What does exist in the EHRbase ecosystem is an “EHRbase Client Library” referenced in the official EHRbase docs—but it’s a Java client/code-generator workflow (generate Java classes from an OPT), not a Python package. [1] If you’re trying to talk to an EHRbase server from Python, the supported/standard approach is to call the official openEHR REST API endpoints that EHRbase implements. EHRbase documents that it implements openEHR REST API v1.0.2, and its API docs point you to the openEHR REST spec for endpoint details. [2] [3] Sources: Citations:
🏁 Script executed: # Search for any references to oehrpy in the codebase and verify import path
rg "oehrpy|openehr_sdk" --type py -B2 -A2Repository: platzhersh/open-cis Length of output: 4196 🏁 Script executed: # Check if there are any type stubs or additional config for oehrpy
find . -name "*.pyi" -o -name "py.typed" | head -20Repository: platzhersh/open-cis Length of output: 45 🏁 Script executed: # Search for any local openehr_sdk or oehrpy package definition
find . -type f -name "__init__.py" | xargs grep -l "EHRBaseClient\|openehr_sdk\|oehrpy" 2>/dev/null | head -20Repository: platzhersh/open-cis Length of output: 92 🏁 Script executed: # Check for git submodules
cat .gitmodules 2>/dev/null || echo "No .gitmodules file"Repository: platzhersh/open-cis Length of output: 82 🏁 Script executed: # Look for any package directories that might define openehr_sdk
find . -type d -name "openehr*" -o -type d -name "*oehr*" 2>/dev/null | head -20Repository: platzhersh/open-cis Length of output: 109 🏁 Script executed: # Check full client.py file to see all methods and how _ensure_connected is used
cat -n api/src/ehrbase/client.pyRepository: platzhersh/open-cis Length of output: 6448 Pass Settings define Also verify whether 🤖 Prompt for AI Agents |
||
|
|
||
| async def create_ehr(self, ehr_id: str | None = None) -> dict[str, Any]: | ||
| """Create a new EHR, optionally with a specific ID.""" | ||
| client = await self._ensure_connected() | ||
| ehr = await client.create_ehr(ehr_id=ehr_id) | ||
| return {"ehr_id": {"value": ehr.ehr_id}, "system_id": ehr.system_id} | ||
|
|
||
| async def get_ehr(self, ehr_id: str) -> dict[str, Any]: | ||
| """Get an EHR by ID.""" | ||
| client = await self._ensure_connected() | ||
| ehr = await client.get_ehr(ehr_id) | ||
| return {"ehr_id": {"value": ehr.ehr_id}, "system_id": ehr.system_id} | ||
|
|
||
| async def get_ehr_by_subject( | ||
| self, subject_id: str, subject_namespace: str = "cis" | ||
| ) -> dict[str, Any] | None: | ||
| """Get an EHR by subject (patient) ID.""" | ||
| client = await self._ensure_connected() | ||
| try: | ||
| ehr = await client.get_ehr_by_subject(subject_id, subject_namespace) | ||
| return {"ehr_id": {"value": ehr.ehr_id}, "system_id": ehr.system_id} | ||
| except EHRBaseError: | ||
| return None | ||
|
|
||
| async def create_composition( | ||
| self, | ||
| ehr_id: str, | ||
| template_id: str, | ||
| composition: dict[str, Any], | ||
| format: str = "FLAT", | ||
| ) -> dict[str, Any]: | ||
| """Create a composition in an EHR.""" | ||
| client = await self._ensure_connected() | ||
| result = await client.create_composition( | ||
| ehr_id=ehr_id, | ||
| composition=composition, | ||
| template_id=template_id, | ||
| format=format, | ||
| ) | ||
| return { | ||
| "uid": {"value": result.uid}, | ||
| "compositionUid": result.uid, | ||
| } | ||
|
|
||
| async def get_composition(self, ehr_id: str, composition_uid: str) -> dict[str, Any]: | ||
| """Get a composition by UID in default format.""" | ||
| client = await self._ensure_connected() | ||
| result = await client.get_composition(ehr_id, composition_uid) | ||
| return result.composition or {} | ||
|
|
||
| async def get_composition_formatted( | ||
| self, ehr_id: str, composition_uid: str, format: str = "FLAT" | ||
| ) -> dict[str, Any]: | ||
| """Get a composition in a specific format (FLAT or STRUCTURED).""" | ||
| client = await self._ensure_connected() | ||
| result = await client.get_composition(ehr_id, composition_uid, format=format) | ||
| return result.composition or {} | ||
|
|
||
| async def delete_composition(self, ehr_id: str, composition_uid: str) -> bool: | ||
| """Delete a composition by UID. Returns True if successful.""" | ||
| client = await self._ensure_connected() | ||
| await client.delete_composition(ehr_id, composition_uid) | ||
| return True | ||
|
|
||
| async def execute_aql( | ||
| self, | ||
| query: str, | ||
| parameters: dict[str, Any] | None = None, | ||
| ) -> dict[str, Any]: | ||
| """Execute an AQL query.""" | ||
| client = await self._ensure_connected() | ||
| result = await client.query(query, query_parameters=parameters) | ||
| return { | ||
| "columns": result.columns, | ||
| "rows": result.rows, | ||
| } | ||
|
|
||
| async def list_templates(self) -> list[dict[str, Any]]: | ||
| """List all available templates.""" | ||
| client = await self._ensure_connected() | ||
| templates = await client.list_templates() | ||
| return [ | ||
| { | ||
| "template_id": t.template_id, | ||
| "concept": t.concept, | ||
| "archetype_id": t.archetype_id, | ||
| } | ||
| for t in templates | ||
| ] | ||
|
|
||
| async def upload_template(self, template_content: str) -> dict[str, Any]: | ||
| """Upload an operational template (OPT).""" | ||
| client = await self._ensure_connected() | ||
| result = await client.upload_template(template_content) | ||
| return {"template_id": result.template_id, "status": "uploaded"} | ||
|
|
||
| async def get_template_example( | ||
| self, template_id: str, format: str = "FLAT" | ||
| ) -> dict[str, Any]: | ||
| """Get template info.""" | ||
| client = await self._ensure_connected() | ||
| result = await client.get_template(template_id) | ||
| return {"template_id": result.template_id, "concept": result.concept} | ||
|
|
||
| async def health_check(self) -> bool: | ||
| """Check if EHRBase is available.""" | ||
| client = await self._ensure_connected() | ||
| return await client.health_check() | ||
|
|
||
| async def close(self) -> None: | ||
| """Close the underlying client.""" | ||
| if self._client: | ||
| await self._client.close() | ||
| self._client = None | ||
|
|
||
|
|
||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
🧩 Analysis chain
🏁 Script executed:
rg -n "ehrbase_timeout|ehrbase_verify_ssl" --type=py -B 2 -A 2Repository: platzhersh/open-cis
Length of output: 353
🏁 Script executed:
rg -n "EHRBaseClient" --type=py -B 3 -A 5Repository: platzhersh/open-cis
Length of output: 4589
🏁 Script executed:
Repository: platzhersh/open-cis
Length of output: 963
🏁 Script executed:
Repository: platzhersh/open-cis
Length of output: 6448
Pass
ehrbase_timeoutandehrbase_verify_sslto the OehrpyClient initialization.The new config settings are defined but never consumed. Update the
_ensure_connected()method inapi/src/ehrbase/client.py(lines 24–28) to passtimeoutandverify_sslparameters when instantiatingOehrpyClient:🤖 Prompt for AI Agents