# -*- coding: utf-8 -*-
"""
This Example demonstrates interaction with the Pathway, Patient, Care Team, and Worklist APIs.
The CLI utilizes the requests library to make HTTP requests to the Carium API.
The CLI is organized into classes for each API, with a main class to run the CLI.
Main functionality demonstrated includes:
.. list-table::
:widths: 35 65
:header-rows: 1
* - Functionality
- Description
* - Login to the Carium API
- Authenticates the user to the Carium API.
* - Get organization ID
- Retrieves and displays the organization ID.
* - Show pathways
- Displays information about available pathways.
* - Enroll patient
- Enrolls a patient in a pathway.
* - Show patients
- Displays information about patients.
* - Show care team
- Displays information about the care team.
* - Show worklist
- Displays information about the worklist.
* - Complete worklist
- Marks the worklist item as completed.
* - Update worklist
- Updates worklist details (due date, assignment to care team, completion).
To run the CLI, execute the following command:
.. code-block:: bash
# run the CLI
python guideexamples/pathway/simple_cli.py
"""
from datetime import datetime
from getpass import getpass
from typing import Callable, Generator, Optional
import requests
from prettytable import PrettyTable
from requests.models import Response
[docs]
class SimpleCli:
"""A simple CLI to interact with the Carium API"""
# Endpoint URLs, used to make requests to the Carium API
API_BASE_URL = "https://api-cicd.carium.com"
LOGIN_URL = f"{API_BASE_URL}/identity/v1/login/"
ORGANIZATION_URL = f"{API_BASE_URL}/identity/v1/organizations/"
PATHWAY_URL = f"{API_BASE_URL}/overlord/v1/pathway-specs/"
PATHWAY_STAGES_URL = f"{API_BASE_URL}/overlord/v1/pathway-spec-stages/"
PATHWAY_INSTANCE_URL = f"{API_BASE_URL}/overlord/v1/pathway-instances/"
PATIENTS_URL = f"{API_BASE_URL}/caredb/v1/aggregation-individuals/"
CARE_TEAM_URL = f"{API_BASE_URL}/identity/v1/individual-groups/"
WORKLIST_URL = f"{API_BASE_URL}/caredb/v1/todo-entries/"
def __init__(self):
self.api = Api(self)
self.organization_id = None
[docs]
@classmethod
def create_table(cls, fields: list[str]) -> PrettyTable:
"""Create a table with the given fields, will append an 'Index' to column field headers
Args:
fields: list[str]: List of column names for the table
Returns:
PrettyTable: An instance of the PrettyTable class.
"""
def split_camel_case(text: str) -> str:
"""
Args:
text: str: text to reformat to capitalize the first letter of each word
Returns:
"""
return " ".join(x[0].capitalize() + x[1:] for x in text.split("-"))
table = PrettyTable()
table.align = "r"
# Extract column names from the keys of the first dictionary in the data list
columns = fields
# Set the field names of the table
table.field_names = ["Index"] + [split_camel_case(x) for x in columns]
return table
[docs]
@classmethod
def print_table(
cls, table: PrettyTable, data: Generator[list[dict], None, None], data_fields: list[str], mapper: Callable
):
"""Print the table with the data, using the mapper to map the data to the fields
Args:
table: PrettyTable: table with the column headers
data: Generator[list[dict], None, None]: List of dict results from the GET API
data_fields: list[str]: response attribute names in the same order as the create_table fields
mapper: Callable: function to map the API response data to the fields
"""
# Add each row to the table
idx = 1
for _data in data:
table.clear_rows()
for row in _data:
_row = mapper(row)
table.add_row([idx] + [_row.get(column, "") for column in data_fields])
idx += 1
print(table)
[docs]
def run(self):
"""Run the main loop of the CLI"""
self.api.login()
self.organization_id = Organization(self).get_organization_id()
while True:
choice = self.menu()
match choice:
case "1":
p = Pathway(self)
p.run()
case "2":
pt = Patient(self)
pt.run()
case "3":
ct = CareTeam(self)
ct.run()
case "4":
w = Worklist(self)
w.run()
case "5":
break
case _:
print("Invalid choice. Please enter a valid number.")
[docs]
class Api:
"""A class to wrap python requests method to include Carium API headers.
It also includes a method to login to the API"""
def __init__(self, cli: SimpleCli):
self.cli = cli
self.token = None
self.headers = {}
[docs]
def login(self):
"""Login to the Carium API
Raises: RuntimeError: If the login fails
"""
username = input("Enter username: ")
password = getpass("Enter password: ")
payload = {"username": username, "password": password}
response = requests.post(self.cli.LOGIN_URL, json=payload)
if response.ok:
print("Login successful.\n")
self.token = response.json()["access-token"]
self.headers = {"Authorization": f"Bearer {self.token}", "accept": "application/vnd.api+json"}
else:
raise RuntimeError("Login failed. Please try again.")
[docs]
def list_with_paging(
self, url: str, params: dict, limit: Optional[int] = None
) -> Generator[list[dict], None, None]:
"""Get data from the API with paging, prompt the user to page more data if available
Args:
url: str: http url for the GET API
params: dict: query parameters for the GET API
limit: Optional[int]: maxium number of records to return in a single request
Returns:
A generator of list of dict results from the GET API
"""
paging = True
offset = 0
while paging:
params.update(
{
"page[offset]": offset,
"page[limit]": limit or 100,
}
)
response = self.get(url, params=params)
if not response.ok:
print(f"Failed to get data: {response.content.decode()}")
break
result = response.json()
_data = result["data"]
total = result["meta"]["page[total]"]
limit = result["meta"]["page[limit]"]
if offset + limit >= total:
paging = False
else:
offset += limit
yield _data
if paging:
choice = input("\nPress Enter to view more or q to exit...")
if choice.lower() == "q":
paging = False
[docs]
def post(self, url: str, json: dict) -> Response:
"""Make a POST request to the API
Args:
url: str: http url for the POST API
json: dict: request body for the POST API
Returns:
The response from the POST request.
"""
return requests.post(url, headers=self.headers, json=json)
[docs]
def get(self, url: str, params: dict) -> Response:
"""Make a GET request to the API
Args:
url: str: http url for the GET API
params: dict: query parameters for the GET API
Returns:
The response from the GET request
"""
return requests.get(url, headers=self.headers, params=params)
[docs]
def patch(self, url: str, json: dict) -> Response:
"""Make a PATCH request to the API
Args:
url: str: http url for the PATCH API
json: dict: request body for the PATCH API
Returns:
The response from the PATCH request
"""
return requests.patch(url, headers=self.headers, json=json)
[docs]
class Organization:
"""A class to interact with the Organization API"""
def __init__(self, cli: SimpleCli):
self.cli = cli
self.api = cli.api
[docs]
def get_organization_id(self) -> str:
"""Get the organization ID from the user, by asking for the organization name (prefix)
Returns:
The organization ID
"""
while True:
org_name = input("Enter organization name: ")
response = self.api.get(self.cli.ORGANIZATION_URL, params={"filter[name][contains]": org_name})
if response.ok:
orgs = response.json()["data"]
if len(orgs) == 1:
return orgs[0]["id"]
if len(orgs) > 1:
print("Multiple organizations found. Please enter a more specific name.")
else:
print("Organization not found. Please try again.")
else:
print("Organization not found. Please try again.")
[docs]
class Pathway:
"""A class to interact with the Pathway API"""
fields = ["id", "name", "description", "version", "version-id"]
def __init__(self, cli: SimpleCli):
self.cli = cli
self.api = cli.api
self.organization_id = cli.organization_id
self.individual_id = None
[docs]
def run(self):
"""Prompt the user for the patient ID and show the pathway menu"""
self.individual_id = input("\nEnter patient ID (optional): ")
while True:
choice = self.menu()
match choice:
case "1":
self.show_pathways()
case "2":
self.enroll_patient()
case "3":
break
case _:
print("Invalid choice. Please enter a valid number.")
[docs]
def show_pathways(self):
"""Show the pathways available in the organization, or optionally filter by patient ID
Displays the pathway ID, name, description, version, and version ID.
"""
def attr_map(d: dict) -> dict:
"""
Args:
d: dict: response API data
Returns:
Mapped data used by the print_table method
"""
return {
"id": d["id"],
"name": d["attributes"]["name"],
"description": d["attributes"]["version"]["description"],
"version": d["attributes"]["version"]["name"],
"version-id": d["attributes"]["version"]["id"],
}
params = {"organization-id": self.cli.organization_id}
if self.individual_id:
params.update({"filter[individual-id]": self.individual_id})
table = self.cli.create_table(Pathway.fields)
data = self.api.list_with_paging(self.cli.PATHWAY_URL, params=params)
self.cli.print_table(
table,
data,
Pathway.fields,
attr_map,
)
def _get_stages(self, pathway_spec_version_id: str) -> list[tuple[str, str]]:
"""Get the stages for a pathway version
Args:
pathway_spec_version_id: str: The pathway version ID
Returns:
A list of tuples containing the stage title and stage ID
"""
response = self.api.get(self.cli.PATHWAY_STAGES_URL, params={"version-id": pathway_spec_version_id})
if not response.ok:
print(f"Failed to get stages: {response.content.decode()}")
return []
return [(x["attributes"]["title"], x["id"]) for x in response.json()["data"]]
[docs]
def enroll_patient(self):
"""Enroll a patient in a pathway
Steps:
1. Prompt the user for the pathway version ID.
2. Get the stages for the pathway version.
3. Prompt the user to choose a stage to enroll the patient.
4. Enroll the patient in the chosen stage
"""
if not (individual_id := self.individual_id):
print("Patient ID is required to enroll.\n")
return
pathway_spec_version_id = input("\nEnter pathway Version ID: ")
stages = self._get_stages(pathway_spec_version_id)
if not stages:
return
stage = input(
"\nChoose a stage to enroll the patient: \n"
+ "\n".join([f"\t{i}. {x[0]}" for i, x in enumerate(stages)])
+ "\n"
)
stage_id = stages[int(stage)][1]
payload = {
"data": {
"attributes": {
"auto-start": True,
"description": "",
"individual-id": individual_id,
"initial-spec-stage-id": stage_id,
"version-id": pathway_spec_version_id,
},
"type": "pathway-instances",
}
}
response = self.api.post(self.cli.PATHWAY_INSTANCE_URL, json=payload)
if not response.ok:
print(f"Failed to enroll patient: {response.content.decode()}")
return
print(f"Patient enrolled successfully: pathway-instance id: {response.json()['data']['id']}.\n")
[docs]
class Patient:
"""A class to interact with the Patient API"""
fields: list[str] = ["id", "first-name", "last-name", "email"]
def __init__(self, cli: SimpleCli):
self.cli = cli
self.api = cli.api
[docs]
def run(self):
"""Show the patient menu"""
while True:
choice = self.menu()
match choice:
case "1":
self.show_patients()
case "2":
break
case _:
print("Invalid choice. Please enter a valid number.")
[docs]
def show_patients(self):
"""Show the patients in the organization
Displays the patient ID, first name, last name, and email.
"""
def attr_map(d: dict) -> dict:
"""
Args:
d: dict: response API data
Returns:
Mapped data used by the print_table method
"""
return {
"id": d["id"],
"first-name": d["attributes"].get("first-name", ""),
"last-name": d["attributes"].get("last-name", ""),
"email": d["attributes"].get("email", ""),
}
table = self.cli.create_table(Patient.fields)
data = self.api.list_with_paging(
self.cli.PATIENTS_URL,
params={
"organization-id": self.cli.organization_id,
"filter[account-type]": "regular",
"filter[registered]": True,
"fields": ["email", "first-name", "last-name"],
},
)
self.cli.print_table(
table,
data,
Patient.fields,
attr_map,
)
[docs]
class CareTeam:
"""A class to interact with the Care Team API"""
fields: list[str] = ["id", "name"]
def __init__(self, cli: SimpleCli):
self.cli = cli
self.api = cli.api
[docs]
def run(self):
"""Show the care team menu"""
while True:
choice = self.menu()
match choice:
case "1":
self.show_care_team()
case "2":
break
case _:
print("Invalid choice. Please enter a valid number.")
[docs]
def show_care_team(self):
"""Show the care team in the organization
Displays the care team ID and name.
"""
def attr_map(d: dict) -> dict:
"""
Args:
d: dict: response API data
Returns:
Mapped data used by the print_table method
"""
return {"id": d["id"], "name": d["attributes"]["name"]}
table = self.cli.create_table(CareTeam.fields)
data = self.api.list_with_paging(
self.cli.CARE_TEAM_URL,
params={"organization-id": self.cli.organization_id, "filter[node-type]": "group", "type": "provider_only"},
)
self.cli.print_table(
table,
data,
CareTeam.fields,
attr_map,
)
[docs]
class Worklist:
"""A class to interact with the Worklist API"""
fields = ["id", "text", "created-at", "done", "care-team", "care-team-id", "assigned-at", "due-time"]
def __init__(self, cli: SimpleCli):
self.cli = cli
self.api = cli.api
[docs]
def run(self):
"""Show the worklist menu"""
while True:
choice = self.menu()
match choice:
case "1":
self.show_worklist()
case "2":
self.complete_worklist()
case "3":
self.update_worklist()
case "4":
break
case _:
print("Invalid choice. Please enter a valid number.")
[docs]
def show_worklist(self):
"""Show the worklist in the organization
Displays the worklist ID, text, created-at, done, care-team, care-team-id, assigned-at, and due-time.
"""
def attr_map(d: dict) -> dict:
"""
Args:
d: dict: response API data
Returns:
Mapped data used by the print_table method
"""
return {
"id": d["id"],
"text": d["attributes"]["text"],
"created-at": d["attributes"]["created-at"],
"done": d["attributes"]["done"],
"care-team": d["attributes"].get("individual-group", {}).get("name", ""),
"care-team-id": d["attributes"].get("individual-group-id", ""),
"assigned-at": d["attributes"]["assigned-at"],
"due-time": d["attributes"].get("due-time", ""),
}
table = self.cli.create_table(Worklist.fields)
# include all worklists in the organization
params = {
"organization-id": self.cli.organization_id,
"filter[include-done]": True,
"filter[provider-only]": True,
}
data = self.api.list_with_paging(self.cli.WORKLIST_URL, params=params)
self.cli.print_table(
table,
data,
Worklist.fields,
attr_map,
)
[docs]
def complete_worklist(self):
"""Complete a worklist entry
Steps:
1. Prompt the user for the worklist ID to complete
2. Mark the worklist as completed
"""
worklist_id = input("Enter worklist ID to complete: ")
payload = {
"data": {
"attributes": [{"op": "replace", "path": "/done", "value": True}],
"id": worklist_id,
"type": "todo-entries",
}
}
response = self.api.patch(self.cli.WORKLIST_URL + f"{worklist_id}/", json=payload)
if response.ok:
print("Worklist entry completed successfully.\n")
else:
print(f"Failed to complete worklist: {response.content.decode()}")
[docs]
def update_worklist(self):
"""Update a worklist entry
Steps:
1. Prompt the user for the worklist ID to update
2. Show the update worklist menu
"""
worklist_id = input("Enter worklist ID to update: ")
while True:
print("\nChoose an option:")
print("1. Update due date")
print("2. Assign to Care Team")
print("3. Exit")
choice = input("Enter the number of your choice: ")
match choice:
case "1":
self.update_due_time(worklist_id)
case "2":
self.assign_to_care_team(worklist_id)
case "3":
break
case _:
print("Invalid choice. Please enter a valid number.")
[docs]
def update_due_time(self, worklist_id: str):
"""Update the due date of a worklist entry
Steps:
1. Prompt the user for the due date
2. Update the due date of the worklist entry
Args:
worklist_id: str: worklist ID to update
"""
while True:
due_date = input("Enter due date (YYYY-MM-DD HH:MM:SS): ")
if self.validate_date_format(due_date):
break
print("Invalid date time format. Please try again.")
payload = {
"data": {
"attributes": [{"op": "replace", "path": "/due-time", "value": due_date}],
"id": worklist_id,
"type": "todo-entries",
}
}
response = self.api.patch(self.cli.WORKLIST_URL + f"{worklist_id}/", json=payload)
if response.ok:
print("Worklist due time updated successfully.\n")
else:
print(f"Failed to update due time worklist: {response.content.decode()}")
[docs]
def assign_to_care_team(self, worklist_id: str):
"""Assign a worklist entry to a care team
Steps:
1. Prompt the user for the care team ID
2. Assign the worklist entry to the care team
Args:
worklist_id: str: worklist ID to update
"""
care_team_id = input("Enter care team ID: ")
payload = {
"data": {
"attributes": [{"op": "replace", "path": "/individual-group-id", "value": care_team_id}],
"id": worklist_id,
"type": "todo-entries",
}
}
response = self.api.patch(self.cli.WORKLIST_URL + f"{worklist_id}/", json=payload)
if response.ok:
print("Worklist assign care team updated successfully.\n")
else:
print(f"Failed to update care team worklist: {response.content.decode()}")
[docs]
def main():
"""Main function to run the Simple CLI"""
print("Welcome to the Carium Pathway Example!\n")
cli = SimpleCli()
cli.run()
print("Goodbye!")
if __name__ == "__main__": # pragma: no cover
main()