Add support for fhir_code, fhir_url and fhir_resource

CHANGELOG.md

@@ -1,3 +1,7 @@
+## 0.1.15
+
+- Add compliance_params (fhirCore, fhirUrl, fhirResource) as operation parameter
+
 ## 0.1.14
 
 - Fix db innitialization crash for fhirschema

Pipfile

@@ -20,8 +20,7 @@ autohooks-plugin-ruff = "~=23.11.0"
 autohooks-plugin-black = "~=23.7.0"
 coloredlogs = "*"
 ruff = "*"
-
-# dependencies for python 3.8
 exceptiongroup = "~=1.2.1"
 tomli = "~=2.0.1"
 async-timeout = "~=4.0.3"
+fhirpathpy = "==2.0.0"

aidbox_python_sdk/__init__.py

@@ -1,5 +1,5 @@
 __title__ = "aidbox-python-sdk"
-__version__ = "0.1.14"
+__version__ = "0.1.15"
 __author__ = "beda.software"
 __license__ = "None"
 __copyright__ = "Copyright 2024 beda.software"

aidbox_python_sdk/sdk.py

@@ -1,11 +1,13 @@
 import asyncio
 import logging
+from typing import Optional
 
 import jsonschema
 from fhirpy.base.exceptions import OperationOutcome
 
 from .aidboxpy import AsyncAidboxClient
 from .db_migrations import sdk_migrations
+from .types import Compliance
 
 logger = logging.getLogger("aidbox_sdk")
 
@@ -151,6 +153,7 @@ def operation(  # noqa: PLR0913
         access_policy=None,
         request_schema=None,
         timeout=None,
+        compliance: Optional[Compliance] = None,
     ):
         if public and access_policy is not None:
             raise ValueError("Operation might be public or have access policy, not both")
@@ -184,6 +187,7 @@ def wrapped_func(operation, request):
                     "method": method,
                     "path": path,
                     **({"timeout": timeout} if timeout else {}),
+                    **(compliance if compliance else {}),
                 }
                 self._operation_handlers[operation_id] = wrapped_func
                 if public is True:

aidbox_python_sdk/types.py

@@ -1,8 +1,13 @@
-from typing import Any
+from typing import Any, List
 
 from aiohttp import web
 from typing_extensions import TypedDict
 
+class Compliance(TypedDict, total=True):
+    fhirUrl: str
+    fhirCode: str
+    fhirResource: List[str]
+
 SDKOperationRequest = TypedDict(
     "SDKOperationRequest",
     {"app": web.Application, "params": dict, "route-params": dict, "headers": dict, "resource": Any},

docker-compose.yaml

@@ -1,59 +1,42 @@
 version: '3.1'
 services:
-  devbox-db:
-    image: "healthsamurai/aidboxdb:14.5"
+  aidbox-db:
+    image: "healthsamurai/aidboxdb:16.1"
+    env_file:
+      - ./envs/db
     ports:
-      - "5434:5432"
-    environment:
-      POSTGRES_USER:     postgres
-      POSTGRES_PASSWORD: postgres
-      POSTGRES_DB:       devbox
-  devbox:
-    image: "healthsamurai/aidboxone:latest"
+      - 5432:5432
+    healthcheck:
+      test: ["CMD", "pg_isready", "-U", "postgres"]
+      interval: 5s
+      timeout: 5s
+      retries: 10
+  aidbox:
+    image: ${AIDBOX_PROJECT_IMAGE}
+    build: example/aidbox-project/
     depends_on:
-      - devbox-db
-    links:
-      - "devbox-db:database"
-    ports:
-      - "8080:8080"
+      aidbox-db:
+        condition: service_healthy
     env_file:
-    - env_tests
+      - ./envs/aidbox
     environment:
-      PGHOST: database
-      PGDATABASE: devbox
-      PGPORT: 5432
-      PGUSER: postgres
-      PGPASSWORD: postgres
-      AIDBOX_CONFIG: /var/config/config.edn
       AIDBOX_LICENSE: ${AIDBOX_LICENSE}
-    volumes:
-      - ./config:/var/config
-  devbox-healthcheck:
-    image: curlimages/curl
-    entrypoint: /bin/sleep 10000
-    links:
-      - devbox
-    depends_on:
-      - devbox
     healthcheck:
-      test: curl --fail http://devbox:8080/__healthcheck || exit 1
-      interval: 1s
-      timeout: 20s
-      retries: 100
+      test: curl --fail http://localhost:8080/health || exit 1
+      interval: 5s
+      timeout: 30s
+      retries: 50
   app:
     build:
       context: .
       args:
         PYTHON_VERSION: ${PYTHON:-3.11}
     command: ["pipenv", "run", "pytest"]
     depends_on:
-      devbox-healthcheck:
-        condition:
-          service_healthy
-    links:
-     - devbox
+      aidbox:
+        condition: service_healthy
     env_file:
-     - env_tests
+      - ./envs/backend
     ports:
      - "8081:8081"
     volumes:

envs/aidbox

@@ -0,0 +1,19 @@
+AIDBOX_STDOUT_PRETTY=all
+AIDBOX_CLIENT_ID=root
+AIDBOX_CLIENT_SECRET=secret
+AIDBOX_BASE_URL=http://aidbox:8080
+AIDBOX_PORT=8080
+AIDBOX_FHIR_VERSION=4.0.0
+
+PGHOST=aidbox-db
+PGPORT=5432
+PGDATABASE=aidbox-tests
+PGUSER=postgres
+PGPASSWORD=postgres
+
+BOX_PROJECT_GIT_TARGET__PATH=/aidbox-project
+AIDBOX_ZEN_PATHS=path:package-dir:/aidbox-project
+AIDBOX_ZEN_ENTRYPOINT=main/box
+AIDBOX_DEV_MODE=false
+AIDBOX_ZEN_DEV_MODE=false
+AIDBOX_COMPLIANCE=enabled

envs/backend

@@ -0,0 +1,10 @@
+APP_INIT_CLIENT_ID=root
+APP_INIT_CLIENT_SECRET=secret
+APP_INIT_URL=http://aidbox:8080
+
+APP_ID=backend-test
+APP_SECRET=secret
+APP_URL=http://backend:8081
+APP_PORT=8081
+AIO_PORT=8081
+AIO_HOST=0.0.0.0

envs/db

@@ -0,0 +1,3 @@
+POSTGRES_USER=postgres
+POSTGRES_PASSWORD=postgres
+POSTGRES_DB=aidbox-tests

main.py

@@ -160,3 +160,15 @@ async def db_tests(request):
         result = await db.alchemy(statement)
         logging.debug("Result:\n%s", result)
     return web.json_response({})
+
+
+@sdk.operation(
+    ["POST"],
+    ["Observation", "observation-custom-op"],
+    compliance={
+        "fhirCode": "observation-custom-op",
+        "fhirUrl": "http://test.com",
+        "fhirResource": ["Observation"],
+    })
+async def observation_custom_op(operation, request):
+    return {"message": "Observation custom operation response"}

run_test.sh

@@ -1,3 +1,16 @@
-#!/bin/bash
+#!/bin/sh
 
+if [ -f "envs/aidbox" ]; then
+    export `cat envs/aidbox`
+fi
+
+if [ -z "${AIDBOX_LICENSE}" ]; then
+    echo "AIDBOX_LICENSE is required to run tests"
+    exit 1
+fi
+
+
+docker compose -f docker-compose.yaml pull --quiet
 docker compose -f docker-compose.yaml up --exit-code-from app app
+
+exit $?

tests/test_sdk.py

@@ -3,10 +3,38 @@
 from unittest import mock
 
 import pytest
+from fhirpathpy import evaluate
 
 import main
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("expression", "expected"),
+    [
+        (
+            "CapabilityStatement.rest.operation.where(definition='http://test.com').count()",
+            1,
+        ),
+        (
+            "CapabilityStatement.rest.operation.where(definition='http://test.com').first().name",
+            "observation-custom-op",
+        ),
+        (
+            "CapabilityStatement.rest.resource.where(type='Observation').operation.where(definition='http://test.com').count()",
+            1,
+        ),
+        (
+            "CapabilityStatement.rest.resource.where(type='Observation').operation.where(definition='http://test.com').first().name",
+            "observation-custom-op",
+        ),
+    ],
+)
+async def test_operation_with_compliance_params(aidbox_client, expression, expected):
+    response = await aidbox_client.execute("fhir/metadata", method="GET")
+    assert evaluate(response, expression, {})[0] == expected
+
+
 @pytest.mark.asyncio()
 async def test_health_check(client):
     resp = await client.get("/health")