aidbox-python-sdk

1. Create a python 3.9+ environment pyenv
2. Set env variables and activate virtual environment source activate_settings.sh
3. Install the required packages with pipenv install --dev
4. Make sure the app's settings are configured correctly (see activate_settings.sh and aidbox_python_sdk/settings.py). You can also use environment variables to define sensitive settings, eg. DB connection variables (see example .env-ptl)
5. You can then run example with python example.py.

Getting started

Minimal application

main.py

fromaidbox_python_sdk.mainimportcreate_appas_create_appfromaidbox_python_sdk.settingsimportSettingsfromaidbox_python_sdk.sdkimportSDKsettings=Settings(**{})
sdk=SDK(settings, resources={}, seeds={})
defcreate_app():
app=await_create_app(SDK)
returnappasyncdefcreate_gunicorn_app() ->web.Application:
returncreate_app()

Register handler for operation

importloggingfromaiohttpimportwebfromaidbox_python_sdk.typesimportSDKOperation, SDKOperationRequestfromyourappfolderimportsdk@sdk.operation(methods=["POST", "PATCH"],path=["signup", "register", {"name": "date"}, {"name": "test"}],timeout=60000## Optional parameter to set a custom timeout for operation in milliseconds)defsignup_register_op(_operation: SDKOperation, request: SDKOperationRequest):
""" POST /signup/register/21.02.19/testvalue PATCH /signup/register/22.02.19/patchtestvalue """logging.debug("`signup_register_op` operation handler")
logging.debug("Operation data: %s", operation)
logging.debug("Request: %s", request)
returnweb.json_response({"success": "Ok", "request": request["route-params"]})

Usage of AppKeys

To access Aidbox Client, SDK, settings, DB Proxy the app (web.Application) is extended by default with the following app keys that are defined in aidbox_python_sdk.app_keys module:

fromaidbox_python_sdkimportapp_keysasakfromaidbox_python_sdk.typesimportSDKOperation, SDKOperationRequest@sdk.operation(["POST"], ["example"])asyncdefupdate_organization_op(_operation: SDKOperation, request: SDKOperationRequest):
app=request.appclient=app[ak.client] # AsyncAidboxClientsdk=app[ak.sdk] # SDKsettings=app[ak.settings] # Settingsdb=app[ak.db] # DBProxyreturnweb.json_response()

Usage of FHIR Client

FHIR Client is not plugged in by default, however, to use it you can extend the app by adding new AppKey

app/app_keys.py

fromfhirpyimportAsyncFHIRClientfhir_client: web.AppKey[AsyncFHIRClient] =web.AppKey("fhir_client", AsyncFHIRClient)

main.py

fromcollections.abcimportAsyncGeneratorfromaidbox_python_sdk.mainimportcreate_appas_create_appfromaidbox_python_sdk.settingsimportSettingsfromaidbox_python_sdk.sdkimportSDKfromaiohttpimportBasicAuth, webfromfhirpyimportAsyncFHIRClientfromappimportapp_keysasaksettings=Settings(**{})
sdk=SDK(settings, resources={}, seeds={)
defcreate_app():
app=await_create_app(SDK)
app.cleanup_ctx.append(fhir_clients_ctx)
returnappasyncdefcreate_gunicorn_app() ->web.Application:
returncreate_app()
asyncdeffhir_clients_ctx(app: web.Application) ->AsyncGenerator[None, None]:
app[ak.fhir_client] =awaitinit_fhir_client(app[ak.settings], "/fhir")
yieldasyncdefinit_fhir_client(settings: Settings, prefix: str="") ->AsyncFHIRClient:
basic_auth=BasicAuth(
login=settings.APP_INIT_CLIENT_ID,
password=settings.APP_INIT_CLIENT_SECRET,
)
returnAsyncFHIRClient(
f"{settings.APP_INIT_URL}{prefix}",
authorization=basic_auth.encode(),
dump_resource=lambdax: x.model_dump(),
)

After that, you can use app[ak.fhir_client] that has the type AsyncFHIRClient everywhere where the app is available.

Usage of request_schema

fromaidbox_python_sdk.typesimportSDKOperation, SDKOperationRequestschema= {
"required": ["params", "resource"],
"properties": {
"params": {
"type": "object",
"required": ["abc", "location"],
"properties": {"abc": {"type": "string"}, "location": {"type": "string"}},
"additionalProperties": False,
},
"resource": {
"type": "object",
"required": ["organizationType", "employeesCount"],
"properties": {
"organizationType": {"type": "string", "enum": ["profit", "non-profit"]},
"employeesCount": {"type": "number"},
},
"additionalProperties": False,
},
},
}
@sdk.operation(["POST"], ["Organization", {"name": "id"}, "$update"], request_schema=schema)asyncdefupdate_organization_op(_operation: SDKOperation, request: SDKOperationRequest):
location=request["params"]["location"]
returnweb.json_response({"location": location})

Valid request example

POST /Organization/org-1/$update?abc=xyz&location=us
organizationType: non-profit
employeesCount: 10

Testing with the pytest plugin

The SDK provides a pytest plugin that starts your app, exposes fixtures for the SDK and Aidbox client, and helps isolate tests that create resources.

Activating the plugin

In your project’s conftest.py (e.g. tests/conftest.py), register the plugin:

pytest_plugins= ["aidbox_python_sdk.pytest_plugin"]

Configuring the app factory

The plugin needs your app factory (the callable that returns the web.Application). You can set it in pytest ini:

pyproject.toml

[tool.pytest.ini_options]
aidbox_create_app = "main:create_app"

Use the dotted path to your callable: either module:name (e.g. main:create_app) or module.submodule.name (e.g. mypackage.main.create_app). The default is main:create_app.

To use a different factory without changing ini, override the fixture in your conftest.py:

@pytest.fixture(scope="session")defcreate_app():
frommyapp.entryimportcreate_appreturncreate_app

Fixtures provided

Fixture	Description
app	The running web.Application (server in a background thread on port 8081).
client	HTTP client for the app + client.server.app for the application instance.
sdk	The SDK instance: app[ak.sdk].
aidbox_client	AsyncAidboxClient for calling Aidbox (operations, /$psql, etc.).
aidbox_db	DB proxy: app[ak.db].
safe_db	Isolated DB for the test; see below.

Using safe_db for tests that create resources

Use the safe_db fixture in tests that create or change data. It records the current transaction id, runs your test, then rolls back everything created in that test so the DB stays clean.

NOTE: Without safe_db, all subscriptions are implicitly ignored for that test.

Example:

@pytest.mark.asyncioasyncdeftest_create_patient(aidbox_client, safe_db):
patient=awaitaidbox_client.resource("Patient", name=[{"family": "Test"}]).save()
patients=awaitaidbox_client.resources("Patient").fetch_all()
assertlen(patients) >=1# after the test, safe_db rolls back and the patient is not persisted

Subscription trigger helper (was_subscription_triggered)

Use sdk.was_subscription_triggered(entity) (or was_subscription_triggered_n_times(entity, n)) only together with the safe_db fixture. Without safe_db, subscription handling is skipped for the test and the returned future is never completed, so the test will hang until the timeout.

Example:

@pytest.mark.asyncioasyncdeftest_patient_subscription(aidbox_client, safe_db, sdk):
was_patient_sub_triggered=sdk.was_subscription_triggered("Patient")
patient=awaitaidbox_client.resource("Patient", name=[{"family": "Test"}]).save()
awaitwas_patient_sub_triggered# assertions...