Artikel ini menjelaskan cara membangun API penerima hasil lab (HL7 → JSON) menggunakan Python Flask dan menyimpannya ke MySQL. API ini digunakan sebagai endpoint penerima dari middleware HL7 listener.
📁 Struktur Proyek
lis_receiver/
├── app.py
├── config.py
├── db.py
├── models.py
├── .env
├── requirements.txt
├── sample_lab_result.json
📦 1. File requirements.txt
Flask
mysql-connector-python
python-dotenv
Instalasi:
pip install -r requirements.txt
⚙️ 2. File .env
API_KEY=your_secret_api_key_here
DB_USER=root
DB_PASSWORD=password
DB_HOST=localhost
DB_NAME=lis_lab
🔐 3. File config.py
import os
from dotenv import load_dotenv
load_dotenv()
API_KEY = os.getenv('API_KEY')
DB_CONFIG = {
'user': os.getenv('DB_USER'),
'password': os.getenv('DB_PASSWORD'),
'host': os.getenv('DB_HOST'),
'database': os.getenv('DB_NAME')
}
🛠️ 4. File db.py
import mysql.connector
from config import DB_CONFIG
def get_db():
return mysql.connector.connect(**DB_CONFIG)
📋 5. File models.py
from db import get_db
def save_lab_result(data):
conn = get_db()
cursor = conn.cursor()
cursor.execute("""
INSERT INTO patients (patient_id, name, dob, gender)
VALUES (%s, %s, %s, %s)
ON DUPLICATE KEY UPDATE name=%s, dob=%s, gender=%s
""", (
data['patient_id'], data['patient_name'], data['dob'], data['gender'],
data['patient_name'], data['dob'], data['gender']
))
order_id = data.get('order_id') or data['order']['order_id']
cursor.execute("""
INSERT INTO lab_orders (order_id, patient_id)
VALUES (%s, %s)
ON DUPLICATE KEY UPDATE patient_id=%s
""", (order_id, data['patient_id'], data['patient_id']))
for test in data['tests']:
cursor.execute("""
INSERT INTO lab_results (order_id, test_code, test_name, value, unit, reference_range, abnormal_flag)
VALUES (%s, %s, %s, %s, %s, %s, %s)
""", (
order_id,
test.get('test_code'),
test.get('test_name'),
test.get('value'),
test.get('unit'),
test.get('reference_range'),
test.get('abnormal_flag')
))
conn.commit()
cursor.close()
conn.close()
🚀 6. File app.py
from flask import Flask, request, jsonify
from models import save_lab_result
from config import API_KEY
app = Flask(__name__)
@app.route('/api/lab-results/', methods=['POST'])
def lab_results():
client_key = request.headers.get('Authorization')
if not client_key or client_key.replace("Bearer ", "") != API_KEY:
return jsonify({"status": "unauthorized"}), 401
try:
data = request.json
save_lab_result(data)
return jsonify({"status": "success"}), 201
except Exception as e:
return jsonify({"status": "error", "message": str(e)}), 500
if __name__ == '__main__':
app.run(debug=True, port=8000)
🗃️ Struktur Database MySQL
CREATE DATABASE IF NOT EXISTS lis_lab;
USE lis_lab;
CREATE TABLE patients (
patient_id VARCHAR(50) PRIMARY KEY,
name VARCHAR(100),
dob VARCHAR(20),
gender VARCHAR(10)
);
CREATE TABLE lab_orders (
order_id VARCHAR(50) PRIMARY KEY,
patient_id VARCHAR(50),
FOREIGN KEY (patient_id) REFERENCES patients(patient_id)
);
CREATE TABLE lab_results (
id INT AUTO_INCREMENT PRIMARY KEY,
order_id VARCHAR(50),
test_code VARCHAR(50),
test_name VARCHAR(100),
value VARCHAR(50),
unit VARCHAR(20),
reference_range VARCHAR(50),
abnormal_flag VARCHAR(5),
FOREIGN KEY (order_id) REFERENCES lab_orders(order_id)
);
🧪 Contoh Payload JSON (sample_lab_result.json)
{
"patient_id": "12345",
"patient_name": "JOHN DOE",
"dob": "19800101",
"gender": "M",
"order_id": "1001",
"tests": [
{
"test_code": "GLU",
"test_name": "Glucose",
"value": "5.2",
"unit": "mmol/L",
"reference_range": "3.5-6.1",
"abnormal_flag": "N"
}
]
}
🧪 Tes API dengan cURL
curl -X POST http://localhost:8000/api/lab-results/ \
-H "Authorization: Bearer your_secret_api_key_here" \
-H "Content-Type: application/json" \
-d @sample_lab_result.json
Tips: Gunakan Mirth Connect, HL7 sender, atau middleware TCP listener untuk mengirimkan data HL7 yang diubah menjadi JSON ke API ini.