satusehat-integration

Open-source Python SDK for integrating with SATUSEHAT — Indonesia's national health data platform powered by FHIR R4. Pure Python, no framework dependency.

Overview

satusehat-integration is an open-source Python SDK for integrating with SATUSEHAT — Indonesia's national health data platform powered by FHIR R4.

Built on the official SATUSEHAT Platform Guidelines. Ships with:

• 140 PayloadBuilder classes — fluent builders for all FHIR R4 resources (Patient, Practitioner, Organization, Encounter, Observation, Procedure, etc.)
• 50 DataType classes — composable FHIR R4 value objects with to_json() serialization
• TerminologyResolver — castable terminology strings ("ICD10:A00", "LOINC:2951-2", "SNOMED:38341003") directly to CodeableConcept
• 3 SATUSEHAT-specific resources: BillingStatus (NON-FHIR JSON), PurificationDecision (NON-FHIR JSON), Endpoint (FHIR R4)
• Queue + Rate Limiter — in-memory queue with configurable RPM rate limiting
• pytest test suite — all builders have comprehensive unit tests

Zero dependencies beyond standard library. Works standalone or with Django/FastAPI/Flask.

Requirements

• Python 3.10 or later

Quick Install

pip install satusehat-integration
# or
git clone https://github.com/ivanwilliammd/satusehat-integration.git
cd satusehat-integration && pip install -e .

# .env
SATUSEHAT_ENV=DEV          # DEV | STG | PROD
SATUSEHAT_BASE_URL_DEV=https://api-satusehat-dev.dto.kemkes.go.id
CLIENTID_DEV=your_client_id
CLIENTSECRET_DEV=your_client_secret
ORGID_DEV=your_org_id

Architecture

DataType Classes (src/datatype/)

Atomic FHIR R4 value objects. All provide a to_json() method — nested types serialize to clean FHIR JSON automatically.

Example — HumanName:

from src.datatype.datatypes import HumanName

name = HumanName(
    family='Doe',
    given=['John', 'Michael'],
    use='official'
)
# name.to_json() → {'family': 'Doe', 'given': ['John', 'Michael'], 'use': 'official'}

PayloadBuilder Pattern (src/builder/)

Fluent builder for each FHIR resource. Each builder exposes chainable methods and returns the resource payload via to_json().

from src.builder.payload_builders import PatientBuilder, HumanName

patient = PatientBuilder()
patient.add_name(HumanName(family='Doe', given=['John'], use='official').to_json())
patient.set_gender('male')
patient.set_birth_date('1990-01-15')

payload = patient.to_json()

Supported FHIR Resources

140 PayloadBuilder classes covering all FHIR R4 resources used in SATUSEHAT interoperability, plus 3 SATUSEHAT-specific resources.

SATUSEHAT Interoperability Resources (47)

⚡ = NON-FHIR JSON (SATUSEHAT-specific extension)

BillingStatus (NON-FHIR JSON)

from src.builder.billing_status import BillingStatusBuilder

billing = (BillingStatusBuilder()
    .set_id('bs-001')
    .add_identifier('http://sys-ids.kemkes.go.id/billing/org-001', 'BILL-12345')
    .set_status('active')
    .set_insurer('Organization/org-bpjs', 'BPJS Kesehatan')
    .set_subject('100000030009', 'Budi Santoso')
    .set_request('cer-001')
    .build())

Endpoint (FHIR R4)

from src.builder.endpoint import EndpointBuilder

endpoint = (EndpointBuilder()
    .set_id('ep-001')
    .set_status('active')
    .set_connection_type('ihe-xcpd', 'IHE XCPD')
    .set_name('SATUSEHAT FHIR Endpoint')
    .set_managing_organization('Organization/org-ihs')
    .set_address('https://satusehat-api.example.com/fhir/r4')
    .build())

PurificationDecision (NON-FHIR JSON)

from src.builder.purification_decision import PurificationDecisionBuilder

pd = (PurificationDecisionBuilder()
    .set_id('pd-001')
    .add_identifier('http://sys-ids.kemkes.go.id/purification/org-001', 'PD-12345')
    .set_status('approved', 'Approved')
    .set_insurer('Organization/org-bpjs', 'BPJS Kesehatan')
    .set_provider('Organization/hos-001', 'Rumah Sakit Sehat')
    .set_claim_response('cr-001')
    .set_created('2024-01-15T10:35:00+00:00')
    .build())

TerminologyResolver — castable codes

from src.terminology.resolver import resolve, expand_array

# Cast terminology strings directly to CodeableConcept
resolve('ICD10:A00')
# → {'coding': [{'system': 'http://hl7.org/fhir/sid/icd-10', 'code': 'A00', 'display': 'A00'}], 'text': 'A00'}

resolve('LOINC:2951-2')
# → {'coding': [{'system': 'http://loinc.org', 'code': '2951-2', 'display': '2951-2'}], 'text': '2951-2'}

# Batch expand
expand_array(['ICD10:A00', 'ICD10:J18.9'])
# → [resolved_A00, resolved_J18.9]

TerminologyClient — HTTP client SmarTerm (terminology.ivanmd.id)

TerminologyClient adalah HTTP client standalone untuk SmarTerm, TERPISAH dari ekosistem SATUSEHAT — jangan reuse OAuth SATUSEHAT Kemkes. SmarTerm punya auth sendiri, dual-mode:

1. Sanctum PAT (token statis) — header Bearer. Env TERMINOLOGY_API_TOKEN.
2. OIDC client_credentials ke auth.ivanmd.id (Passport) — JWT di-fetch lazy, di-cache sampai expire, auto-refresh sekali saat respon 401. Env TERMINOLOGY_CLIENT_ID, TERMINOLOGY_CLIENT_SECRET, TERMINOLOGY_TOKEN_URL (default https://auth.ivanmd.id/oauth/token).

Jika keduanya diberikan, PAT menang. Env lain: TERMINOLOGY_BASE_URL (default https://terminology.ivanmd.id).

from src.terminology import TerminologyClient

term = TerminologyClient()  # baca env, atau: TerminologyClient(token='pat-xxx')

# Fluent chaining
term.query('pemasangan kateter').on('kptl').limit(10).get()
term.query('demam').category('diagnosis').first()
term.query('paracetamol').on('kfa').all()

# as_code() → POST validate, sistem ditebak dari bentuk kode
term.query('93004944').as_code().get()   # 93xxxx → system 'kfa'
term.query('J18.9').as_code().get()      # ICD-10

# Pencarian & mapping
term.search('demam', category='all', limit=20)          # unified search
term.search_system('icd10', 'demam', limit=20)          # per-sistem
term.kptl_combine('9.10.1', ['9.11.1', '9.11.2'])       # modifier join koma
term.map('icd10', 'snomed', 'J18.9')                    # cross-mapping
term.validate('icd10', 'J18.9')
term.auto_map('icd10', 'J18.9')                         # → SNOMED

# Context-aware FHIR resolution (body otomatis buang field None)
term.resolve_condition('J18.9')
term.resolve_chief_complaint('demam')
term.resolve_observation('2951-2', system_hint='loinc')
term.resolve_medication('93004944')
term.resolve_service_request('LAB001', category='lab')

# Value sets
term.value_sets()
term.value_set('kfa')

HTTP status >= 400 melempar RuntimeError("TerminologyClient HTTP {status}: …"); network error melempar RuntimeError("TerminologyClient network error: …").

Usage Examples

Patient

from src.builder.payload_builders import PatientBuilder

patient = PatientBuilder()
patient.add_name({
    'family': 'Doe',
    'given': ['John'],
    'use': 'official'
})
patient.set_gender('male')
patient.set_birth_date('1990-01-15')
patient.add_telecom({'system': 'phone', 'value': '081234567890', 'use': 'mobile'})

payload = patient.to_json()
print(payload)

Claim (BPJS Klaim)

from src.builder.payload_builders import ClaimBuilder

claim = ClaimBuilder()
claim.set_status('active')
claim.set_use('claim')
claim.set_type('institutional')
claim.set_patient('pat-123', 'enc-456')
claim.add_item(1, 'PROCID001', 150000, 'IDR')
claim.set_total(150000, 'IDR')

payload = claim.to_json()

Documentation

External Resources

• HL7 FHIR R4 Specification
• SATUSEHAT Platform Docs
• Main PHP SDK
• SATUSEHAT Sandbox API

Contributing

Contributions are welcome. Please ensure tests pass and follow existing code conventions.

License

MIT — see LICENSE.