A technical comparison between TCP and UDP protocols implemented in Python: examining performance metrics, security considerations, and practical applications within healthcare systems using FHIR standards for effective data exchange between medical platforms.

Introduction: The Significance of TCP vs UDP

When browsing websites, streaming videos, or making video calls, our data travels across networks using protocols that ensure reliable and efficient delivery. Two transport protocols dominate this landscape: TCP (Transmission Control Protocol) and UDP (User Datagram Protocol).

What fundamental differences exist between these protocols, and how do these differences impact performance in real-world applications?

Table of Contents

In this post, we’ll cover:

• The theoretical foundations of TCP and UDP
• Their practical differences demonstrated with Python
• A hands-on TCP and UDP communication benchmark
• Visual analysis of transmission times
• An asynchronous implementation using asyncio and aiohttp
• Secure Data Transmission in Healthcare IT
• Python for Medical Data Transfer

All the code for this project is available on GitHub

TCP vs UDP: Core Concepts Compared

TCP (Transmission Control Protocol)

• Connection-oriented: establishes a reliable connection with a three-way handshake, ensuring both parties are ready to communicate before any data transfer begins.
• Reliable: guarantees delivery and reorders packets if needed, with mechanisms for acknowledging received data and retransmitting lost packets automatically.
• Flow and congestion control: adapts to network conditions by monitoring bandwidth availability and adjusting transmission rates to prevent network congestion and packet loss.
• Used for: HTTPS, email, file transfers, SSH, web browsing, database connections, and any application where data integrity is critical.

UDP (User Datagram Protocol)

• Connectionless: sends data without setting up a connection, eliminating the overhead associated with connection establishment and termination processes.
• Unreliable: no guarantees for delivery or ordering, which means packets may arrive out of sequence, be duplicated, or not arrive at all without automatic notification.
• Minimal overhead: faster and lighter due to the absence of connection management, acknowledgments, and retransmission mechanisms found in TCP.
• Used for: DNS, video/audio streaming, online gaming, VoIP, live broadcasts, IoT devices, and time-sensitive applications where speed is prioritized over perfect reliability.

Understanding the socket Module in Python

Python’s socket module provides a low-level networking interface based on the BSD socket API. It supports both TCP (SOCK_STREAM) and UDP (SOCK_DGRAM) protocols, enabling developers to send and receive data across networks.

Key Functions and Concepts

• socket.socket(family, type): creates a new socket object for network communication. For our networking purposes, we typically use AF_INET (for IPv4 addressing) and either SOCK_STREAM for TCP connections or SOCK_DGRAM for UDP datagrams, depending on our reliability and performance requirements.
• bind((host, port)): assigns a specific network address (combination of IP address and port number) to the socket, effectively reserving that address for the application. This function is primarily used on the server side to establish a known endpoint where clients can connect.
• listen(): configures a TCP socket to passively wait for and queue incoming connection requests, transforming it into a listening socket. This method is exclusive to TCP sockets since UDP doesn’t maintain connection state.
• accept(): blocks execution and waits for an incoming TCP connection request. When a client connects, it returns a new socket object specifically for that client connection along with the client’s address information.
• connect((host, port)): actively initiates a TCP connection from a client socket to a server at the specified address. This triggers the three-way handshake process that establishes a reliable TCP connection.
• sendall(data) / sendto(data, addr): transmits the specified data to the connected peer. sendall() is used with TCP connections and ensures all data is sent, while sendto() is used with UDP and requires specifying the destination address with each call.
• recv(bufsize) / recvfrom(bufsize): receives incoming data from the peer, with bufsize indicating the maximum amount of data to be received at once. recv() works with established TCP connections, while recvfrom() is used with UDP and additionally returns the sender’s address.
• close(): terminates the socket connection and releases the resources associated with it. For TCP sockets, this initiates the connection termination process, while for UDP sockets, it simply frees the socket descriptor.

The socket module operates in a blocking mode by default, which means function calls like recv() or accept() will pause execution until they complete their operation. In our benchmark, we implement threading to enable the server to listen for incoming data without halting the client’s execution flow.

Benchmarking TCP and UDP in Python

Goal

We’ll benchmark the transmission times for 100 simple messages sent between client and server over both TCP and UDP protocols in a local environment.

Setup

• Server and client implementations for each protocol
• Localhost communication (127.0.0.1)
• threading for concurrent server operation
• time.time() for precise timing measurements
• matplotlib for visualizing performance results

#--------------------
# tcp vs udp
# di Michele Danilo Pierri
# 08/08/2025
#--------------------


"""
What this measures:
  - UDP: one datagram (request) -> echo (response) per transaction.
  - TCP: connect -> send -> recv -> close per transaction.
"""

import argparse
import socket
import threading
import time
from time import perf_counter
import statistics as stats
import matplotlib.pyplot as plt

# ---------------------------
# Defaults (tuneable via CLI)
# ---------------------------
DEFAULT_HOST = "127.0.0.1"
TCP_PORT = 57211
UDP_PORT = 57212

# Small payload accentuates handshake cost for TCP
DEFAULT_PAYLOAD = 32       # bytes
REPEAT = 400               # transactions per protocol
PACE = 0.001               # seconds between transactions to avoid bursts
TIMEOUT = 2.0              # seconds socket timeout

# ---------------------------
# Servers
# ---------------------------

def tcp_transaction_server(host: str, port: int):
    """
    Accepts connections in a loop.
    For each connection:
      - read exactly one payload (client sends once)
      - echo it back
      - close
    No artificial sleep; this stays 'real'.
    """
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        s.bind((host, port))
        s.listen(128)
        while True:
            conn, _ = s.accept()
            try:
                with conn:
                    # Read exactly one message; size unknown to server,
                    # so read once up to some reasonable amount
                    data = conn.recv(65536)
                    if data:
                        conn.sendall(data)
            except ConnectionError:
                continue


def udp_echo_server(host: str, port: int):
    """
    Stateless echo: for each datagram, send it back to sender.
    """
    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        s.bind((host, port))
        while True:
            data, addr = s.recvfrom(65536)
            if data:
                s.sendto(data, addr)

# ---------------------------
# Clients / Measurements
# ---------------------------

def measure_udp_transactions(host: str, port: int, payload: bytes, n: int):
    """
    For each transaction:
      - send one datagram
      - wait for echo
      - record transaction time (application-level RTT)
    """
    durations = []
    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as c:
        c.settimeout(TIMEOUT)
        for _ in range(n):
            t0 = perf_counter()
            c.sendto(payload, (host, port))
            data, _ = c.recvfrom(65536)
            dt = perf_counter() - t0
            durations.append(dt)
            time.sleep(PACE)
    return durations


def measure_tcp_transactions(host: str, port: int, payload: bytes, n: int):
    """
    For each transaction:
      - connect()
      - send payload once
      - recv echo once
      - close
      - record full transaction time (includes handshake)
    """
    durations = []
    for _ in range(n):
        t0 = perf_counter()
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as c:
            c.settimeout(TIMEOUT)
            # Optionally disable Nagle to avoid tiny writes coalescing
            c.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
            c.connect((host, port))
            c.sendall(payload)
            # Expect a single echo; read once is typically enough on localhost/LAN
            data = c.recv(65536)
            # Close via context manager
        dt = perf_counter() - t0
        durations.append(dt)
        time.sleep(PACE)
    return durations

# ---------------------------
# Plot helpers
# ---------------------------

def summarize(name, arr):
    mean = stats.mean(arr)
    med = stats.median(arr)
    stdev = stats.pstdev(arr)
    return f"{name}: mean={mean:.6e}s, median={med:.6e}s, std={stdev:.6e}s, n={len(arr)}"

def plot_results(tcp, udp, payload_size):
    # 1) Boxplot for robust comparison
    plt.figure(figsize=(9,5))
    plt.boxplot([tcp, udp], labels=["TCP per-tx (handshake)", "UDP per-tx"])
    plt.title(f"Per-Transaction RTT (echo), payload={payload_size} bytes")
    plt.ylabel("Seconds")
    plt.tight_layout()

    # 2) Bar plot mean ± std
    plt.figure(figsize=(9,5))
    means = [stats.mean(tcp), stats.mean(udp)]
    stds  = [stats.pstdev(tcp), stats.pstdev(udp)]
    plt.bar(["TCP per-tx", "UDP per-tx"], means, yerr=stds)
    plt.title("Per-Transaction Mean ± Std")
    plt.ylabel("Seconds")
    plt.tight_layout()
    plt.show()

# ---------------------------
# Main
# ---------------------------

def main():
    ap = argparse.ArgumentParser(description="Real UDP vs TCP per-transaction benchmark")
    ap.add_argument("--host", default=DEFAULT_HOST, help="Server bind/target host (use LAN IP for cross-machine test)")
    ap.add_argument("--payload", type=int, default=DEFAULT_PAYLOAD, help="Payload size in bytes (default: 32)")
    ap.add_argument("--repeat", type=int, default=REPEAT, help="Transactions per protocol (default: 400)")
    args = ap.parse_args()

    host = args.host
    payload = b"A" * args.payload
    repeat = args.repeat

    # Start servers as daemons
    t_tcp = threading.Thread(target=tcp_transaction_server, args=(host, TCP_PORT), daemon=True)
    t_udp = threading.Thread(target=udp_echo_server,         args=(host, UDP_PORT), daemon=True)
    t_tcp.start()
    t_udp.start()
    time.sleep(0.3)  # give servers time to bind

    # Measure
    tcp_times = measure_tcp_transactions(host, TCP_PORT, payload, repeat)
    udp_times = measure_udp_transactions(host, UDP_PORT, payload, repeat)

    # Print summaries
    print(summarize("TCP per-transaction", tcp_times))
    print(summarize("UDP per-transaction", udp_times))

    # Plot
    plot_results(tcp_times, udp_times, len(payload))

if __name__ == "__main__":
    main()

#--------------------
# tcp vs udp
# di Michele Danilo Pierri
# 08/08/2025
#--------------------


"""
What this measures:
  - UDP: one datagram (request) -> echo (response) per transaction.
  - TCP: connect -> send -> recv -> close per transaction.
"""

import argparse
import socket
import threading
import time
from time import perf_counter
import statistics as stats
import matplotlib.pyplot as plt

# ---------------------------
# Defaults (tuneable via CLI)
# ---------------------------
DEFAULT_HOST = "127.0.0.1"
TCP_PORT = 57211
UDP_PORT = 57212

# Small payload accentuates handshake cost for TCP
DEFAULT_PAYLOAD = 32       # bytes
REPEAT = 400               # transactions per protocol
PACE = 0.001               # seconds between transactions to avoid bursts
TIMEOUT = 2.0              # seconds socket timeout

# ---------------------------
# Servers
# ---------------------------

def tcp_transaction_server(host: str, port: int):
    """
    Accepts connections in a loop.
    For each connection:
      - read exactly one payload (client sends once)
      - echo it back
      - close
    No artificial sleep; this stays 'real'.
    """
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        s.bind((host, port))
        s.listen(128)
        while True:
            conn, _ = s.accept()
            try:
                with conn:
                    # Read exactly one message; size unknown to server,
                    # so read once up to some reasonable amount
                    data = conn.recv(65536)
                    if data:
                        conn.sendall(data)
            except ConnectionError:
                continue


def udp_echo_server(host: str, port: int):
    """
    Stateless echo: for each datagram, send it back to sender.
    """
    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        s.bind((host, port))
        while True:
            data, addr = s.recvfrom(65536)
            if data:
                s.sendto(data, addr)

# ---------------------------
# Clients / Measurements
# ---------------------------

def measure_udp_transactions(host: str, port: int, payload: bytes, n: int):
    """
    For each transaction:
      - send one datagram
      - wait for echo
      - record transaction time (application-level RTT)
    """
    durations = []
    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as c:
        c.settimeout(TIMEOUT)
        for _ in range(n):
            t0 = perf_counter()
            c.sendto(payload, (host, port))
            data, _ = c.recvfrom(65536)
            dt = perf_counter() - t0
            durations.append(dt)
            time.sleep(PACE)
    return durations


def measure_tcp_transactions(host: str, port: int, payload: bytes, n: int):
    """
    For each transaction:
      - connect()
      - send payload once
      - recv echo once
      - close
      - record full transaction time (includes handshake)
    """
    durations = []
    for _ in range(n):
        t0 = perf_counter()
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as c:
            c.settimeout(TIMEOUT)
            # Optionally disable Nagle to avoid tiny writes coalescing
            c.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
            c.connect((host, port))
            c.sendall(payload)
            # Expect a single echo; read once is typically enough on localhost/LAN
            data = c.recv(65536)
            # Close via context manager
        dt = perf_counter() - t0
        durations.append(dt)
        time.sleep(PACE)
    return durations

# ---------------------------
# Plot helpers
# ---------------------------

def summarize(name, arr):
    mean = stats.mean(arr)
    med = stats.median(arr)
    stdev = stats.pstdev(arr)
    return f"{name}: mean={mean:.6e}s, median={med:.6e}s, std={stdev:.6e}s, n={len(arr)}"

def plot_results(tcp, udp, payload_size):
    # 1) Boxplot for robust comparison
    plt.figure(figsize=(9,5))
    plt.boxplot([tcp, udp], labels=["TCP per-tx (handshake)", "UDP per-tx"])
    plt.title(f"Per-Transaction RTT (echo), payload={payload_size} bytes")
    plt.ylabel("Seconds")
    plt.tight_layout()

    # 2) Bar plot mean ± std
    plt.figure(figsize=(9,5))
    means = [stats.mean(tcp), stats.mean(udp)]
    stds  = [stats.pstdev(tcp), stats.pstdev(udp)]
    plt.bar(["TCP per-tx", "UDP per-tx"], means, yerr=stds)
    plt.title("Per-Transaction Mean ± Std")
    plt.ylabel("Seconds")
    plt.tight_layout()
    plt.show()

# ---------------------------
# Main
# ---------------------------

def main():
    ap = argparse.ArgumentParser(description="Real UDP vs TCP per-transaction benchmark")
    ap.add_argument("--host", default=DEFAULT_HOST, help="Server bind/target host (use LAN IP for cross-machine test)")
    ap.add_argument("--payload", type=int, default=DEFAULT_PAYLOAD, help="Payload size in bytes (default: 32)")
    ap.add_argument("--repeat", type=int, default=REPEAT, help="Transactions per protocol (default: 400)")
    args = ap.parse_args()

    host = args.host
    payload = b"A" * args.payload
    repeat = args.repeat

    # Start servers as daemons
    t_tcp = threading.Thread(target=tcp_transaction_server, args=(host, TCP_PORT), daemon=True)
    t_udp = threading.Thread(target=udp_echo_server,         args=(host, UDP_PORT), daemon=True)
    t_tcp.start()
    t_udp.start()
    time.sleep(0.3)  # give servers time to bind

    # Measure
    tcp_times = measure_tcp_transactions(host, TCP_PORT, payload, repeat)
    udp_times = measure_udp_transactions(host, UDP_PORT, payload, repeat)

    # Print summaries
    print(summarize("TCP per-transaction", tcp_times))
    print(summarize("UDP per-transaction", udp_times))

    # Plot
    plot_results(tcp_times, udp_times, len(payload))

if __name__ == "__main__":
    main()

Asynchronous Implementation

For use cases with high concurrency or where blocking I/O operations create bottlenecks, an asynchronous approach offers superior performance. By leveraging non-blocking I/O patterns, asynchronous code can efficiently handle numerous connections simultaneously without the overhead of traditional threading models. The example below implements this efficient approach using Python’s asyncio library and asyncio.DatagramProtocol class, which provide a robust framework for managing asynchronous network operations with clean, maintainable code structures.

This implementation focuses only on UDP, as it’s particularly well-suited for asynchronous processing due to its connectionless nature and efficiency with non-blocking high-speed datagrams. While TCP could also benefit from async implementations, UDP’s inherently stateless design makes it an ideal candidate for demonstrating the performance advantages of event-driven I/O operations, especially in scenarios requiring high throughput with minimal latency overhead.

#--------------------
# async udp messages
# di Michele Danilo Pierri
# 08/08/2025
#--------------------

import asyncio
import time
import matplotlib.pyplot as plt

REPEAT = 1000
HOST = '127.0.0.1'
PORT = 6000
MESSAGE = b"Async UDP message"
async_durations = []

class EchoServerProtocol(asyncio.DatagramProtocol):
    def datagram_received(self, data, addr):
        pass  # No response needed

async def run_async_server():
    loop = asyncio.get_running_loop()
    transport, _ = await loop.create_datagram_endpoint(
        lambda: EchoServerProtocol(), local_addr=(HOST, PORT))
    await asyncio.sleep(2)  # Wait for messages
    transport.close()

async def run_async_client():
    loop = asyncio.get_running_loop()
    transport, _ = await loop.create_datagram_endpoint(
        lambda: asyncio.DatagramProtocol(), remote_addr=(HOST, PORT))
    for _ in range(REPEAT):
        start = time.time()
        transport.sendto(MESSAGE)
        async_durations.append(time.time() - start)
        await asyncio.sleep(0.01)
    transport.close()

async def main_async():
    server = asyncio.create_task(run_async_server())
    await asyncio.sleep(0.5)
    await run_async_client()
    await server

asyncio.run(main_async())

plt.plot(async_durations, label="Async UDP")
plt.title("Async UDP Transmission Times")
plt.xlabel("Message Index")
plt.ylabel("Duration (s)")
plt.grid(True)
plt.legend()
plt.show()

#--------------------
# async udp messages
# di Michele Danilo Pierri
# 08/08/2025
#--------------------

import asyncio
import time
import matplotlib.pyplot as plt

REPEAT = 1000
HOST = '127.0.0.1'
PORT = 6000
MESSAGE = b"Async UDP message"
async_durations = []

class EchoServerProtocol(asyncio.DatagramProtocol):
    def datagram_received(self, data, addr):
        pass  # No response needed

async def run_async_server():
    loop = asyncio.get_running_loop()
    transport, _ = await loop.create_datagram_endpoint(
        lambda: EchoServerProtocol(), local_addr=(HOST, PORT))
    await asyncio.sleep(2)  # Wait for messages
    transport.close()

async def run_async_client():
    loop = asyncio.get_running_loop()
    transport, _ = await loop.create_datagram_endpoint(
        lambda: asyncio.DatagramProtocol(), remote_addr=(HOST, PORT))
    for _ in range(REPEAT):
        start = time.time()
        transport.sendto(MESSAGE)
        async_durations.append(time.time() - start)
        await asyncio.sleep(0.01)
    transport.close()

async def main_async():
    server = asyncio.create_task(run_async_server())
    await asyncio.sleep(0.5)
    await run_async_client()
    await server

asyncio.run(main_async())

plt.plot(async_durations, label="Async UDP")
plt.title("Async UDP Transmission Times")
plt.xlabel("Message Index")
plt.ylabel("Duration (s)")
plt.grid(True)
plt.legend()
plt.show()

Results & Discussion

• UDP consistently shows shorter durations due to its non-blocking, connectionless nature.
• TCP introduces overhead from connection setup and acknowledgment processes.
• In the async variant, latency is minimal with stable performance.

Limitations of the benchmark:

• Tests run on localhost, eliminating real network congestion and packet loss
• Real-world performance would differ significantly from these controlled conditions
• For comprehensive UDP analysis, use tools like tc or netem on Linux to simulate jitter and packet loss

Secure Data Transmission in Healthcare IT

Medical data transmission (including electronic health records, lab results, imaging data, and wearable sensor streams) must meet strict requirements for confidentiality, integrity, availability, and traceability.

While TCP and UDP serve as foundational transport protocols, security and compliance in healthcare are implemented at higher layers through specialized protocols, encryption methods, and standardized frameworks designed specifically for medical contexts.

Key concepts

• Transport-level security: Protocols like TLS (Transport Layer Security) establish encrypted communication channels over TCP connections, ensuring confidential and tamper-proof data transmission between endpoints. This security layer is commonly implemented in healthcare systems through protocols such as HTTPS for web-based applications and FTPS for secure file transfers, providing essential protection for sensitive patient information during network transit.
• Application-level security: Healthcare standards such as HL7 v2, FHIR, and DICOM implement comprehensive security frameworks that utilize encrypted communication channels and enforce robust security measures including strict authentication protocols, granular role-based access control systems, and comprehensive audit logging mechanisms that track all data access and modifications for compliance and security purposes. These standards are designed to maintain data integrity while enabling secure information exchange between different healthcare systems and providers across organizational boundaries.
• VPN/IPSec tunnels: These establish secure, encrypted communication pathways between healthcare facilities, including hospitals, outpatient clinics, laboratories, and remote patient monitoring devices. By creating protected virtual corridors across public networks, VPN/IPSec implementations ensure that sensitive medical data remains confidential and protected from unauthorized access during transmission, while maintaining compliance with healthcare privacy regulations and security standards.
• Payload encryption: Medical data is often encrypted directly at the application level (using advanced symmetric encryption algorithms like AES-256 or asymmetric cryptographic methods such as RSA-2048) before transmission across any network. This additional security layer ensures that even if transport-level protections are compromised, the medical information itself remains encrypted and inaccessible to unauthorized parties, providing defense-in-depth for sensitive patient data regardless of the underlying transport protocol being used.

Protocols commonly used in medical systems

• HL7 (Health Level 7): classic messaging for lab results, admissions, etc., often over TCP with MLLP framing, or over HTTPS (FHIR).
• FHIR (Fast Healthcare Interoperability Resources): RESTful API standard using HTTP/HTTPS + JSON/XML + OAuth2 for secure access.
• DICOM (Digital Imaging and Communications in Medicine): for imaging data (CT, MRI, ultrasound), built over TCP and optionally secured via TLS.

Standards, however, are a necessary but not sufficient condition. Two systems can both be FHIR-compliant and still fail to exchange anything clinically meaningful, because interoperability breaks down at the semantic and organisational level long before it breaks down at the protocol level.

Concrete technologies used in hospitals or medical software

Practical example: secure transmission of an ECG

1. ECG device captures patient data.
2. Data is formatted as XML or DICOM files.
3. The device creates a secure HTTPS/TLS connection with the central server.
4. The system verifies identity through OAuth2 authentication.
5. Encrypted data travels to either a FHIR API endpoint or an HL7 integration engine.
6. The server records the transaction and stores the data in an encrypted database.
7. Authorized physicians can view the data through a secure internal web portal (with authentication and comprehensive access logging).

Python for Medical Data Transfer

Let’s simulate the transmission of data with healthcare-grade security using HL7/FHIR protocols in Python. For this demonstration, we use the public HAPI FHIR Test Server, a free testing endpoint provided by the HAPI FHIR open-source project and maintained by Smile Digital Health. This server is designed exclusively for development and interoperability testing, with all uploaded resources being periodically purged. Never submit real patient data — use only synthetic or anonymized test data.

#--------------------
# fhir transfer
# di Michele Danilo Pierri
# 08/08/2025
#--------------------

import requests
import json
import uuid
import datetime

# ------------------------
# CONFIGURATION
# ------------------------

# Target FHIR server URL — for example, a test HAPI FHIR server
FHIR_SERVER_URL = "https://hapi.fhir.org/baseR4/Patient"

# Fake bearer token to simulate OAuth2 
ACCESS_TOKEN = "Bearer fake-token-for-demo-use-only"

# ------------------------
# FHIR RESOURCE GENERATION
# ------------------------

# Build a sample Patient resource according to the HL7 FHIR R4 standard
# This object will be serialized as JSON and sent to the FHIR server
def generate_fake_patient():
    patient_id = str(uuid.uuid4())  # generate a random patient ID
    today = datetime.date.today().isoformat()

    patient_resource = {
        "resourceType": "Patient",
        "id": patient_id,
        "active": True,
        "name": [
            {
                "use": "official",
                "family": "Doe",
                "given": ["John"]
            }
        ],
        "gender": "male",
        "birthDate": "1985-05-15",
        "deceasedBoolean": False,
        "address": [
            {
                "use": "home",
                "line": ["1234 Main Street"],
                "city": "Springfield",
                "state": "IL",
                "postalCode": "62704",
                "country": "USA"
            }
        ],
        "identifier": [
            {
                "use": "usual",
                "type": {
                    "coding": [
                        {
                            "system": "http://terminology.hl7.org/CodeSystem/v2-0203",
                            "code": "MR"
                        }
                    ]
                },
                "system": "http://hospital.smarthealth.org/mrn",
                "value": f"MRN-{patient_id[:8]}"
            }
        ],
        "meta": {
            "lastUpdated": today
        }
    }

    return patient_resource

# ------------------------
# SENDING FUNCTION
# ------------------------

def send_patient_to_fhir_server(patient_data):
    """
    Sends the given FHIR Patient resource to the configured FHIR server using HTTPS POST.
    Includes authentication headers and content negotiation headers.
    """
    headers = {
        "Authorization": ACCESS_TOKEN,
        "Content-Type": "application/fhir+json",
        "Accept": "application/fhir+json"
    }

    try:
        print("Sending patient data to FHIR server...")
        response = requests.post(FHIR_SERVER_URL, headers=headers, data=json.dumps(patient_data))

        if response.status_code in [200, 201]:
            print("Patient resource successfully sent.")
            print(f"Server response location: {response.headers.get('Location', 'N/A')}")
        else:
            print(f"Failed to send patient resource. Status code: {response.status_code}")
            print(f"Response body: {response.text}")

    except requests.exceptions.RequestException as e:
        print(f"Network error: {e}")

# ------------------------
# MAIN
# ------------------------

if __name__ == "__main__":
    print("Generating fake FHIR Patient resource...")
    patient = generate_fake_patient()
    print("Payload preview:")
    print(json.dumps(patient, indent=2))
    
    send_patient_to_fhir_server(patient)

#--------------------
# fhir transfer
# di Michele Danilo Pierri
# 08/08/2025
#--------------------

import requests
import json
import uuid
import datetime

# ------------------------
# CONFIGURATION
# ------------------------

# Target FHIR server URL — for example, a test HAPI FHIR server
FHIR_SERVER_URL = "https://hapi.fhir.org/baseR4/Patient"

# Fake bearer token to simulate OAuth2 
ACCESS_TOKEN = "Bearer fake-token-for-demo-use-only"

# ------------------------
# FHIR RESOURCE GENERATION
# ------------------------

# Build a sample Patient resource according to the HL7 FHIR R4 standard
# This object will be serialized as JSON and sent to the FHIR server
def generate_fake_patient():
    patient_id = str(uuid.uuid4())  # generate a random patient ID
    today = datetime.date.today().isoformat()

    patient_resource = {
        "resourceType": "Patient",
        "id": patient_id,
        "active": True,
        "name": [
            {
                "use": "official",
                "family": "Doe",
                "given": ["John"]
            }
        ],
        "gender": "male",
        "birthDate": "1985-05-15",
        "deceasedBoolean": False,
        "address": [
            {
                "use": "home",
                "line": ["1234 Main Street"],
                "city": "Springfield",
                "state": "IL",
                "postalCode": "62704",
                "country": "USA"
            }
        ],
        "identifier": [
            {
                "use": "usual",
                "type": {
                    "coding": [
                        {
                            "system": "http://terminology.hl7.org/CodeSystem/v2-0203",
                            "code": "MR"
                        }
                    ]
                },
                "system": "http://hospital.smarthealth.org/mrn",
                "value": f"MRN-{patient_id[:8]}"
            }
        ],
        "meta": {
            "lastUpdated": today
        }
    }

    return patient_resource

# ------------------------
# SENDING FUNCTION
# ------------------------

def send_patient_to_fhir_server(patient_data):
    """
    Sends the given FHIR Patient resource to the configured FHIR server using HTTPS POST.
    Includes authentication headers and content negotiation headers.
    """
    headers = {
        "Authorization": ACCESS_TOKEN,
        "Content-Type": "application/fhir+json",
        "Accept": "application/fhir+json"
    }

    try:
        print("Sending patient data to FHIR server...")
        response = requests.post(FHIR_SERVER_URL, headers=headers, data=json.dumps(patient_data))

        if response.status_code in [200, 201]:
            print("Patient resource successfully sent.")
            print(f"Server response location: {response.headers.get('Location', 'N/A')}")
        else:
            print(f"Failed to send patient resource. Status code: {response.status_code}")
            print(f"Response body: {response.text}")

    except requests.exceptions.RequestException as e:
        print(f"Network error: {e}")

# ------------------------
# MAIN
# ------------------------

if __name__ == "__main__":
    print("Generating fake FHIR Patient resource...")
    patient = generate_fake_patient()
    print("Payload preview:")
    print(json.dumps(patient, indent=2))
    
    send_patient_to_fhir_server(patient)

Technical notes

• The HAPI server used accepts POST tests, but the data is public and visible to everyone.
• In a real environment:
  • servers must use HTTPS with valid certificates;
  • authentication occurs through OAuth2 or JWT;
  • data must be encrypted at rest (not only in transit).

Legal and compliance framework

• GDPR (EU): mandates encryption, access control, and data minimization.
• HIPAA (US): requires secure transmission and auditability of health data.
• ISO 27799 / ISO 27001: information security management in healthcare.

Practical Guidelines:

• Use TCP when data integrity, delivery confirmation, and packet ordering are critical. This includes applications such as:
  • Clinical databases
  • Electronic health records (EHR)
  • DICOM imaging transfer between systems
• Use UDP when real-time performance is more important than occasional packet loss, such as:
  • Telemedicine video streams
  • IoT patient monitoring devices
  • PACS viewers that preload images
• Use asynchronous approaches (e.g. asyncio, aiohttp) when dealing with:
  • Multiple concurrent data streams (e.g. multi-patient monitoring)
  • Non-blocking UI-driven systems (e.g. healthcare dashboards)
  • Efficient use of network resources and low-latency systems
• Secure all communication at the transport or application level:
  • Prefer HTTPS/TLS channels, even internally
  • Authenticate and authorize using OAuth2 or API keys
  • Log and audit every transaction involving personal data
• Adopt medical standards such as FHIR and HL7 to ensure interoperability across systems, vendors, and national health infrastructures.

Conclusions

In this article, we examine the differences between TCP and UDP in terms of structure, behavior, and performance. Through practical benchmarking in Python, we demonstrated how these protocols behave under controlled conditions. We also extended our investigation to include asynchronous programming and its benefits in high-concurrency environments.

However, beyond theory and speed comparisons, we delved into the specific needs of healthcare IT, where the transmission of data is not just about speed or reliability, but about security, traceability, and compliance with international regulations.

References & Resources

• RFC 793 – TCP
• RFC 768 – UDP
• Python socket
• Python asyncio
• Matplotlib
• Linux tc for traffic control
• General Data Protection Regulation (GDPR), EU 2016/679
• Health Insurance Portability and Accountability Act (HIPAA)
• ISO/IEC 27001 – Information Security Management
• ISO 27799:2016 – Health informatics — Information security management in health

Introduction: Why Build a Medical Database with Docker?

Creating a robust medical database system requires careful consideration of security, scalability, and maintainability. Furthermore, Docker containerization offers an ideal solution for healthcare applications by providing isolated environments that ensure consistent deployment across different systems.

In this comprehensive tutorial, we’ll explore how to create a medical database with Docker and perform operations on it using various tools. Additionally, we’ll use a practical example: a database designed to store patient demographic and anthropometric data (age, sex, height, weight, etc.).

While the structure we present is relatively simple, it can be scaled to accommodate more complex architectures. Moreover, this foundation provides the flexibility needed for future healthcare system expansions.

Table of Contents

Introduction

• Overview of the tutorial
• Purpose and scope

Tools We’ll Use

• Docker
  • Overview and containerization
  • Benefits of isolation
  • MySQL container setup
• SQLAlchemy
  • Database interaction capabilities
  • ORM functionality
• Flask
  • Web framework basics
  • Database interface creation

Step-by-Step Guide

• Step 1: Download and Configure MySQL Container
  • Docker installation
  • Container configuration
  • Basic Docker commands
• Step 2: Creating Tables with SQLAlchemy
  • Database structure setup
  • Table relationships
  • Data modeling
• Step 3: Data Operations with SQLAlchemy
  • Session management
  • Data insertion
  • Query operations
• Step 4: Web Interface with Flask
  • Application setup
  • Route definitions
  • Template organization
• Step 5: Security Consideration
• Step 6: Summary

All the code for this project is available on GitHub

Essential Tools for Medical Database Development

Why Docker Transforms Healthcare Database Management

Building a medical database with Docker provides several advantages including isolation, portability, and ease of setup. First of all, Docker is an open-source tool for developing, distributing, and running software.

Its key feature is containerization—applications run in isolated environments that contain everything needed for the program to work. However, these containers share the host computer’s kernel while remaining isolated from its operating system. Think of them as lightweight virtual machines that are more efficient because they leverage the host’s kernel.

Thanks to isolation from the “host” environment, containers prevent conflicts from different dependencies and configurations. Consequently, they operate independently from the system while maintaining data persistence through mounted volumes.

SQLAlchemy: Simplifying Database Interactions

Next, SQLAlchemy is one of the most popular Python libraries for working with relational databases. It enables Python code to interact directly with various SQL databases through specific drivers, including MySQL, PostgreSQL, Oracle, and SQLite.

A key feature of SQLAlchemy is its Object-Relational Mapper (ORM), which maps database tables to Python classes. As a result, database interactions become straightforward and intuitive, reducing development time significantly.

Flask: Creating User-Friendly Web Interfaces

Flask is a Python framework for creating web applications. With Flask, we can build SQLAlchemy applications that access databases through an HTML interface. Therefore, users can interact with the medical database without requiring technical database knowledge.

Prerequisites

To start the project, you need Docker (available at Docker: Accelerated Container Application Development) and Python (Download Python | Python.org) installed on your computer.

Step 1: Download and Configure MySQL Container

Setting Up Your Docker Environment

With Docker running on your computer, you can download the MySQL database image from the terminal using this command:

docker pull mysql:latest

docker pull mysql:latest

This command downloads the latest MySQL image from the Docker Hub repository. Subsequently, once the image download is complete, we can create a container from it and configure it to meet our requirements.

Container Configuration and Setup

Navigate to the directory where you want to store your database (using standard commands like cd and mkdir). Then, run this script in the terminal:

docker run --name my-mysql-container \\
	-v my_directory/data:/var/lib/mysql \\ 
  -e MYSQL_ROOT_PASSWORD=my-secret-pw \\
  -e MYSQL_DATABASE=mydatabase \\
  -e MYSQL_USER=myuser \\
  -e MYSQL_PASSWORD=mypassword \\
  -p 3306:3306 \\
  -d mysql:latest

docker run --name my-mysql-container \\
	-v my_directory/data:/var/lib/mysql \\ 
  -e MYSQL_ROOT_PASSWORD=my-secret-pw \\
  -e MYSQL_DATABASE=mydatabase \\
  -e MYSQL_USER=myuser \\
  -e MYSQL_PASSWORD=mypassword \\
  -p 3306:3306 \\
  -d mysql:latest

Here’s what each parameter means:

• –name: Sets the container’s name for easy identification
• -v: Specifies the volume where data is stored, ensuring persistence
• -e: Defines environment variables, including database credentials
• -p: Specifies communication ports for external access
• -d: Runs the container in detached mode

Managing Container Operations

These commands are only needed when initializing the container for the first time. After that, the specified parameters are saved and automatically applied whenever you run the container.

To verify the program has started successfully, use:

docker ps

docker ps

Once you’ve created and configured the container, you won’t need to use docker run again. Instead, you’ll use different commands to stop and restart the program.

Furthermore, to manage your container operations use:

To stop the container:

docker stop my-mysql-container

docker stop my-mysql-container

To restart it, use:

docker start my-mysql-container

docker start my-mysql-container

To completely delete the container, use this command:

docker rm -f my-mysql-container

docker rm -f my-mysql-container

Note: An alternative method called docker compose lets you manage container configurations through a docker-compose.yml file. This approach is typically used for applications with multiple containerized programs, but we won’t cover it in this tutorial.

Step 2: Creating Database Tables with SQLAlchemy

Environment Setup and Library Installation

It is recommended to use an IDE (like Visual Studio Code) and create a virtual environment to complete this step; you also need to verify that the container is active or activate it with the command:

docker start my-mysql-container

docker start my-mysql-container

From Visual Studio Code’s terminal, install the required Python libraries:

pip install sqlalchemy pymysql

pip install sqlalchemy pymysql

Establishing Database Connection

From Python, let’s import the required libraries:

from sqlalchemy import create_engine, Column, Integer, String, Date, Float, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship

from sqlalchemy import create_engine, Column, Integer, String, Date, Float, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship

Define the database connection string (Replace ‘myuser’, ‘mypassword’, ‘localhost’, and ‘mydatabase’ with your actual MySQL credentials)

DATABASE_URL = "mysql+pymysql://myuser:mypassword@localhost:3306/mydatabase"

DATABASE_URL = "mysql+pymysql://myuser:mypassword@localhost:3306/mydatabase"

Create an engine to connect to the database. The engine manages the connection pool and database access. Setting echo=True enables SQL statement logging for debugging.

engine = create_engine(DATABASE_URL, echo=True) 

engine = create_engine(DATABASE_URL, echo=True) 

SQLAlchemy uses a foundational “base” class that helps create database tables in Python. This base class acts as a template – when you create new table classes, they inherit from this base class, making it simple to define and work with database tables in your Python code.

To define it, we use the command:

Base = declarative_base()

Base = declarative_base()

Designing Patient Records Table

Now let’s create the first table, the patient’s PatientRecords using SQLAlchemy’s Base class as a template:

class PatientRecords(Base):
    __tablename__ = 'patient_records'  # Table name in the database

    # Columns
    Id_patient = Column(Integer, primary_key=True, autoincrement=True)  # Primary key
    first_name = Column(String(50), nullable=False)  # Patient's first name
    last_name = Column(String(50), nullable=False)  # Patient's last name
    date_of_birth = Column(Date, nullable=False)  # Patient's date of birth

    # Relationship with the "AnthropometricData" table
    anthropometric_data = relationship("AnthropometricData", back_populates="patient")

    def __repr__(self):
        return f"<PatientRecords(Id_patient={self.Id_patient}, first_name={self.first_name}, last_name={self.last_name})>"

class PatientRecords(Base):
    __tablename__ = 'patient_records'  # Table name in the database

    # Columns
    Id_patient = Column(Integer, primary_key=True, autoincrement=True)  # Primary key
    first_name = Column(String(50), nullable=False)  # Patient's first name
    last_name = Column(String(50), nullable=False)  # Patient's last name
    date_of_birth = Column(Date, nullable=False)  # Patient's date of birth

    # Relationship with the "AnthropometricData" table
    anthropometric_data = relationship("AnthropometricData", back_populates="patient")

    def __repr__(self):
        return f"<PatientRecords(Id_patient={self.Id_patient}, first_name={self.first_name}, last_name={self.last_name})>"

In this script, we define both the table name (patients_record) and its fields, while also establishing a relationship with the Anthropometric_data table (relationship = AnthropometricData). This relationship is bidirectional (back_populates = “patient”). When we create the Anthropometric_data table, we’ll set up a corresponding PatientRecord relationship with a bidirectional link (back_populates = AnthropometricData) to the Patient_record table.

This creates a “logical” link between the two tables, complementing the structural connection already established through Foreign Keys at the database level.

The repr(self) method defines how an object should be represented when it is printed or displayed, converting it into a more readable string format.

Creating Anthropometric Data Table

Let’s create the second table (antropometric_data) using the same approach we used for the PatientRecords table.

class AnthropometricData(Base):
    __tablename__ = 'anthropometric_data'  # Table name in the database

    # Columns
    Id_data = Column(Integer, primary_key=True, autoincrement=True)  # Primary key
    Id_patient = Column(Integer, ForeignKey('patient_records.Id_patient'), nullable=False)  # Foreign key to "patient_records"
    height = Column(Float, nullable=False)  # Height in cm
    weight = Column(Float, nullable=False)  # Weight in kg
    BMI = Column(Float, nullable=False)  # Body Mass Index (calculated as weight / (height/100)^2)

    # Relationship with the "PatientRecords" table
    patient = relationship("PatientRecords", back_populates="anthropometric_data")

    def __repr__(self):
        return f"<AnthropometricData(Id_data={self.Id_data}, Id_patient={self.Id_patient}, BMI={self.BMI})>"

class AnthropometricData(Base):
    __tablename__ = 'anthropometric_data'  # Table name in the database

    # Columns
    Id_data = Column(Integer, primary_key=True, autoincrement=True)  # Primary key
    Id_patient = Column(Integer, ForeignKey('patient_records.Id_patient'), nullable=False)  # Foreign key to "patient_records"
    height = Column(Float, nullable=False)  # Height in cm
    weight = Column(Float, nullable=False)  # Weight in kg
    BMI = Column(Float, nullable=False)  # Body Mass Index (calculated as weight / (height/100)^2)

    # Relationship with the "PatientRecords" table
    patient = relationship("PatientRecords", back_populates="anthropometric_data")

    def __repr__(self):
        return f"<AnthropometricData(Id_data={self.Id_data}, Id_patient={self.Id_patient}, BMI={self.BMI})>"

It’s important to note that, at this point, the tables exist only as logical definitions and haven’t been created in the actual database. The following command will transform them into real tables in our archive by converting our logical structure into SQL commands. SQLAlchemy handles this conversion automatically, saving us significant effort.

Base.metadata.create_all(engine)

Base.metadata.create_all(engine)

To verify that the tables were successfully created, you can interact directly with the MySQL database through the terminal with these commands:

docker exec -it my-mysql-container mysql -u myuser -p
USE mydatabase;
SHOW TABLES;
DESCRIBE patient_records;
DESCRIBE anthropometric_data;

docker exec -it my-mysql-container mysql -u myuser -p
USE mydatabase;
SHOW TABLES;
DESCRIBE patient_records;
DESCRIBE anthropometric_data;

These commands will display the following:

Step 3: Data Operations and Management

Session Management and Database Operations

After setting up the database and tables, we can proceed to populate them with content.

If you create a new program to perform database operations, you’ll need to include the table class definitions (PatientRecords and AnthropometricData) again. While you can copy these definitions manually, there are more efficient ways to avoid this duplication, though we’ll keep things simple and won’t cover those techniques here.

In order to perform database operations (such as queries and data insertion) with SQLAlchemy, we first need to use sessionmaker.

Session = sessionmaker(bind=engine)
session = Session()

Session = sessionmaker(bind=engine)
session = Session()

When creating a session using sessionmaker, these operations happen automatically:

• Connection to the database through the engine
• Tracking of all pending database operations
• Execution of all operations in a single block when committed

A session follows this lifecycle:

• Creation (session = Session())
• Database interactions (queries, reads, insertions)
• Saving changes (session.commit())
• Rolling back changes if errors occur (session.rollback())
• Closing the session (session.close())

Adding Patient Records

Now that we have created a session, we can add a new patient to the patient_records table:

new_patient = PatientRecords(
    first_name="Mario",
    last_name="Rossi",
    date_of_birth="1990-05-15"  # Date format: YYYY-MM-DD
)
session.add(new_patient)
session.commit()

new_patient = PatientRecords(
    first_name="Mario",
    last_name="Rossi",
    date_of_birth="1990-05-15"  # Date format: YYYY-MM-DD
)
session.add(new_patient)
session.commit()

Note that we insert the new patient using the Python table class (PatientRecords) rather than the actual table name (patient_records). SQLAlchemy provides this layer of abstraction, letting us focus on logical operations instead of directly referencing table names. Behind the scenes, SQLAlchemy converts our code into SQL instructions to interact with the database.

Managing Anthropometric Data

Next, let’s add data to the anthropometric_data table:

anthropometric_data = AnthropometricData(
    Id_patient=new_patient.Id_patient,
    height=175.0,  # Height in cm
    weight=70.0,   # Weight in kg
    BMI=70.0 / ((175.0 / 100) ** 2)  # Calculate BMI
)
session.add(anthropometric_data)
session.commit()

anthropometric_data = AnthropometricData(
    Id_patient=new_patient.Id_patient,
    height=175.0,  # Height in cm
    weight=70.0,   # Weight in kg
    BMI=70.0 / ((175.0 / 100) ** 2)  # Calculate BMI
)
session.add(anthropometric_data)
session.commit()

Querying and Verification

Let’s query the tables to verify that our data was successfully inserted:

patients = session.query(PatientRecords).all()
print("\\nPatients in the database:")
for patient in patients:
    print(patient)

anthropometric_records = session.query(AnthropometricData).all()
print("\\nAnthropometric Data in the database:")
for record in anthropometric_records:
    print(record)

patients = session.query(PatientRecords).all()
print("\\nPatients in the database:")
for patient in patients:
    print(patient)

anthropometric_records = session.query(AnthropometricData).all()
print("\\nAnthropometric Data in the database:")
for record in anthropometric_records:
    print(record)

Finally, let’s close the session:

session.close()

session.close()

Alternatively, you can query the database directly through Docker using SQL commands in the terminal:

docker exec -it my-mysql-container mysql -u myuser -p
USE mydatabase;
SELECT * FROM patient_records;
SELECT * FROM anthropometric_data;

docker exec -it my-mysql-container mysql -u myuser -p
USE mydatabase;
SELECT * FROM patient_records;
SELECT * FROM anthropometric_data;

The terminal will display the following results:

Step 4: Building a Web Interface with Flask

Flask Installation and Setup

Flask is a lightweight web framework that lets you create browser-accessible applications to interact with MySQL containers.

First, install Flask in Python by running this command in the terminal:

pip install Flask

pip install Flask

Next, we need to create a file called models.py that reuses our previous ORM class definitions for the database tables:

from sqlalchemy import Column, Integer, String, Date, Float, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship

Base = declarative_base()

class PatientRecords(Base):
    __tablename__ = 'patient_records'

    Id_patient = Column(Integer, primary_key=True, autoincrement=True)
    first_name = Column(String(50), nullable=False)
    last_name = Column(String(50), nullable=False)
    date_of_birth = Column(Date, nullable=False)

    anthropometric_data = relationship("AnthropometricData", back_populates="patient")

    def __repr__(self):
        return f"<PatientRecords(Id={self.Id_patient}, Name={self.first_name} {self.last_name})>"

class AnthropometricData(Base):
    __tablename__ = 'anthropometric_data'

    Id_data = Column(Integer, primary_key=True, autoincrement=True)
    Id_patient = Column(Integer, ForeignKey('patient_records.Id_patient'), nullable=False)
    height = Column(Float, nullable=False)
    weight = Column(Float, nullable=False)
    BMI = Column(Float, nullable=False)

    patient = relationship("PatientRecords", back_populates="anthropometric_data")

    def __repr__(self):
        return f"<AnthropometricData(Id={self.Id_data}, PatientId={self.Id_patient}, BMI={self.BMI})>"

from sqlalchemy import Column, Integer, String, Date, Float, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship

Base = declarative_base()

class PatientRecords(Base):
    __tablename__ = 'patient_records'

    Id_patient = Column(Integer, primary_key=True, autoincrement=True)
    first_name = Column(String(50), nullable=False)
    last_name = Column(String(50), nullable=False)
    date_of_birth = Column(Date, nullable=False)

    anthropometric_data = relationship("AnthropometricData", back_populates="patient")

    def __repr__(self):
        return f"<PatientRecords(Id={self.Id_patient}, Name={self.first_name} {self.last_name})>"

class AnthropometricData(Base):
    __tablename__ = 'anthropometric_data'

    Id_data = Column(Integer, primary_key=True, autoincrement=True)
    Id_patient = Column(Integer, ForeignKey('patient_records.Id_patient'), nullable=False)
    height = Column(Float, nullable=False)
    weight = Column(Float, nullable=False)
    BMI = Column(Float, nullable=False)

    patient = relationship("PatientRecords", back_populates="anthropometric_data")

    def __repr__(self):
        return f"<AnthropometricData(Id={self.Id_data}, PatientId={self.Id_patient}, BMI={self.BMI})>"

Finally, we can create an app.py using Flask.

Our medical database project will use three HTML templates as the foundation for interacting with the container:

• index.html – the main page
• add_patient.html – for adding new patients
• edit_patient.html – for modifying patient records

Directory Structure and Organization

The directory organization will be:

flask_app/
│
├── app.py               # Main Flask app file
├── models.py            # ORM class definitions (PatientRecords, AnthropometricData)
├── templates/           # HTML templates folder
│   ├── index.html       # Main page
│   ├── add_patient.html # Form to add a patient
└   └── edit_patient.html# Form to modify a patient

flask_app/
│
├── app.py               # Main Flask app file
├── models.py            # ORM class definitions (PatientRecords, AnthropometricData)
├── templates/           # HTML templates folder
│   ├── index.html       # Main page
│   ├── add_patient.html # Form to add a patient
└   └── edit_patient.html# Form to modify a patient

We’ll create three HTML templates. The first (index.html) serves as the entry page, displaying the database content and allowing users to select various operations.

The second page (add_patient.html) provides a form for adding patients to the patient_records dataset, while the third page (edit_patient.html) enables modification of existing patient data.

At this stage, we’ve prioritized system functionality over aesthetics, though the visual aspects can be easily improved later.

The script for index.html:

<!DOCTYPE html>
<html>
<head>
    <title>Patient List</title>
</head>
<body>
    <h1>Patient List</h1>
    <!-- Link to add a new patient -->
    <a href="{{ url_for('add_patient') }}">Add New Patient</a>
    <ul>
        <!-- Loop through all patients and display their details -->
        {% for patient in patients %}
            <li>
                {{ patient.first_name }} {{ patient.last_name }}
                <!-- Links to edit or delete the patient -->
                (<a href="{{ url_for('edit_patient', patient_id=patient.Id_patient) }}">Edit</a> |
                <a href="{{ url_for('delete_patient', patient_id=patient.Id_patient) }}">Delete</a>)
            </li>
        {% endfor %}
    </ul>

<!DOCTYPE html>
<html>
<head>
    <title>Patient List</title>
</head>
<body>
    <h1>Patient List</h1>
    <!-- Link to add a new patient -->
    <a href="{{ url_for('add_patient') }}">Add New Patient</a>
    <ul>
        <!-- Loop through all patients and display their details -->
        {% for patient in patients %}
            <li>
                {{ patient.first_name }} {{ patient.last_name }}
                <!-- Links to edit or delete the patient -->
                (<a href="{{ url_for('edit_patient', patient_id=patient.Id_patient) }}">Edit</a> |
                <a href="{{ url_for('delete_patient', patient_id=patient.Id_patient) }}">Delete</a>)
            </li>
        {% endfor %}
    </ul>

The script for add_patient.html:

<!DOCTYPE html>
<html>
<head>
    <title>Add Patient</title>
</head>
<body>
    <h1>Add a New Patient</h1>
    <!-- Form to submit new patient details -->
    <form method="POST">
        First Name: <input type="text" name="first_name"><br>
        Last Name: <input type="text" name="last_name"><br>
        Date of Birth: <input type="date" name="date_of_birth"><br>
        <button type="submit">Add Patient</button>
    </form>
    <!-- Link to return to the patient list -->
    <a href="{{ url_for('index') }}">Back to Patient List</a>
</body>
</html>

<!DOCTYPE html>
<html>
<head>
    <title>Add Patient</title>
</head>
<body>
    <h1>Add a New Patient</h1>
    <!-- Form to submit new patient details -->
    <form method="POST">
        First Name: <input type="text" name="first_name"><br>
        Last Name: <input type="text" name="last_name"><br>
        Date of Birth: <input type="date" name="date_of_birth"><br>
        <button type="submit">Add Patient</button>
    </form>
    <!-- Link to return to the patient list -->
    <a href="{{ url_for('index') }}">Back to Patient List</a>
</body>
</html>

The script for edit_patient.html:

<!DOCTYPE html>
<html>
<head>
    <title>Edit Patient</title>
</head>
<body>
    <h1>Edit Patient Details</h1>
    <!-- Form to update patient details -->
    <form method="POST">
        First Name: <input type="text" name="first_name" value="{{ patient.first_name }}"><br>
        Last Name: <input type="text" name="last_name" value="{{ patient.last_name }}"><br>
        Date of Birth: <input type="date" name="date_of_birth" value="{{ patient.date_of_birth }}"><br>
        <button type="submit">Save Changes</button>
    </form>
    <!-- Link to return to the patient list -->
    <a href="{{ url_for('index') }}">Back to Patient List</a>
</body>
</html>

<!DOCTYPE html>
<html>
<head>
    <title>Edit Patient</title>
</head>
<body>
    <h1>Edit Patient Details</h1>
    <!-- Form to update patient details -->
    <form method="POST">
        First Name: <input type="text" name="first_name" value="{{ patient.first_name }}"><br>
        Last Name: <input type="text" name="last_name" value="{{ patient.last_name }}"><br>
        Date of Birth: <input type="date" name="date_of_birth" value="{{ patient.date_of_birth }}"><br>
        <button type="submit">Save Changes</button>
    </form>
    <!-- Link to return to the patient list -->
    <a href="{{ url_for('index') }}">Back to Patient List</a>
</body>
</html>

Flask Application Development

The app.py program follows below. The program uses Flask decorators (marked by “@app.route()”) to connect web pages with Python code, managing database requests and responses.

from flask import Flask, render_template, request, redirect, url_for
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from models import PatientRecords, AnthropometricData

# Database configuration
DATABASE_URL = "mysql+pymysql://myuser:mypassword@localhost:3306/mydatabase"
engine = create_engine(DATABASE_URL)  # Create a connection to the database
Session = sessionmaker(bind=engine)  # Create a session factory
session = Session()  # Initialize a session to interact with the database

# Initialize Flask app
app = Flask(__name__)

# Home Page: Display all patients
@app.route("/")
def index():
    patients = session.query(PatientRecords).all()  # Query all patients from the database
    return render_template("index.html", patients=patients)  # Render the template with patient data

# Add Patient Page: Handle form submission to add a new patient
@app.route("/add", methods=["GET", "POST"])
def add_patient():
    if request.method == "POST":
        # Retrieve form data
        first_name = request.form["first_name"]
        last_name = request.form["last_name"]
        date_of_birth = request.form["date_of_birth"]

        # Create a new patient object
        new_patient = PatientRecords(
            first_name=first_name,
            last_name=last_name,
            date_of_birth=date_of_birth
        )
        session.add(new_patient)  # Add the new patient to the session
        session.commit()  # Commit the transaction to save the data
        return redirect(url_for("index"))  # Redirect to the home page
    return render_template("add_patient.html")  # Render the form for GET requests

# Edit Patient Page: Handle form submission to update an existing patient
@app.route("/edit/<int:patient_id>", methods=["GET", "POST"])
def edit_patient(patient_id):
    patient = session.query(PatientRecords).get(patient_id)  # Retrieve the patient by ID
    if request.method == "POST":
        # Update patient details with form data
        patient.first_name = request.form["first_name"]
        patient.last_name = request.form["last_name"]
        patient.date_of_birth = request.form["date_of_birth"]
        session.commit()  # Commit the changes to the database
        return redirect(url_for("index"))  # Redirect to the home page
    return render_template("edit_patient.html", patient=patient)  # Render the edit form

# Delete Patient Page: Delete a patient by ID
@app.route("/delete/<int:patient_id>")
def delete_patient(patient_id):
    patient = session.query(PatientRecords).get(patient_id)  # Retrieve the patient by ID
    session.delete(patient)  # Delete the patient from the session
    session.commit()  # Commit the transaction to apply the deletion
    return redirect(url_for("index"))  # Redirect to the home page

# Run the Flask app
if __name__ == "__main__":
    app.run(debug=True)  # Start the app in debug mode for development

from flask import Flask, render_template, request, redirect, url_for
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from models import PatientRecords, AnthropometricData

# Database configuration
DATABASE_URL = "mysql+pymysql://myuser:mypassword@localhost:3306/mydatabase"
engine = create_engine(DATABASE_URL)  # Create a connection to the database
Session = sessionmaker(bind=engine)  # Create a session factory
session = Session()  # Initialize a session to interact with the database

# Initialize Flask app
app = Flask(__name__)

# Home Page: Display all patients
@app.route("/")
def index():
    patients = session.query(PatientRecords).all()  # Query all patients from the database
    return render_template("index.html", patients=patients)  # Render the template with patient data

# Add Patient Page: Handle form submission to add a new patient
@app.route("/add", methods=["GET", "POST"])
def add_patient():
    if request.method == "POST":
        # Retrieve form data
        first_name = request.form["first_name"]
        last_name = request.form["last_name"]
        date_of_birth = request.form["date_of_birth"]

        # Create a new patient object
        new_patient = PatientRecords(
            first_name=first_name,
            last_name=last_name,
            date_of_birth=date_of_birth
        )
        session.add(new_patient)  # Add the new patient to the session
        session.commit()  # Commit the transaction to save the data
        return redirect(url_for("index"))  # Redirect to the home page
    return render_template("add_patient.html")  # Render the form for GET requests

# Edit Patient Page: Handle form submission to update an existing patient
@app.route("/edit/<int:patient_id>", methods=["GET", "POST"])
def edit_patient(patient_id):
    patient = session.query(PatientRecords).get(patient_id)  # Retrieve the patient by ID
    if request.method == "POST":
        # Update patient details with form data
        patient.first_name = request.form["first_name"]
        patient.last_name = request.form["last_name"]
        patient.date_of_birth = request.form["date_of_birth"]
        session.commit()  # Commit the changes to the database
        return redirect(url_for("index"))  # Redirect to the home page
    return render_template("edit_patient.html", patient=patient)  # Render the edit form

# Delete Patient Page: Delete a patient by ID
@app.route("/delete/<int:patient_id>")
def delete_patient(patient_id):
    patient = session.query(PatientRecords).get(patient_id)  # Retrieve the patient by ID
    session.delete(patient)  # Delete the patient from the session
    session.commit()  # Commit the transaction to apply the deletion
    return redirect(url_for("index"))  # Redirect to the home page

# Run the Flask app
if __name__ == "__main__":
    app.run(debug=True)  # Start the app in debug mode for development

The initial page displays the database records and all available actions that can be performed on them:

The additional pages enable users to add or modify patient records:

Step 5: Critical Security Considerations

Understanding Healthcare Data Protection

We are dealing with a medical database and therefore sensitive data whose protection is regulated by legislation. Moreover, GDPR (General Data Protection Regulation) in Europe and HIPAA (Health Insurance Portability and Accountability Act) in the United States impose strict requirements.

Identifying Security Vulnerabilities

Even with a superficial analysis, we can identify numerous critical issues in the structure we have built:

• Exposed passwords: Access credentials to the dataset are embedded in the code and therefore easily stolen.
• Unauthenticated access: The Flask application lacks authentication mechanisms for HTML pages.
• Unencrypted data: Data transmission between the Flask server and HTML pages is not encrypted.
• SQL injection vulnerability: Input data is not validated, exposing the system to attacks through harmful SQL commands.
• Cross-Site scripting vulnerability: Malicious users could exploit the web interface to inject harmful scripts.
• Database exposure: The database is accessible on port 3306: if this port is public, it could be targeted for direct attacks.

Implementing Security Measures

Solutions to these issues include:

• Environment variable management for credentials
• Authentication middleware implementation
• HTTPS encryption for data transmission
• Input validation and parameterized queries
• Content Security Policy headers
• Network segmentation and firewall rules

Step 6: Summary and Next Steps

Key Concepts Review

Let’s summarize the key concepts covered in this tutorial:

First, we used Docker to create a MySQL container, providing an isolated and configurable medical database environment. Subsequently, we implemented SQLAlchemy as an ORM to map database tables to Python classes. Finally, we built a Flask web application that enables browser-based database interactions.

Future Development Possibilities

While this structure is straightforward, it serves as a robust foundation. Furthermore, it can be expanded into more complex architectures including:

• Multi-container orchestration with Docker Compose
• Advanced authentication and authorization systems
• Real-time data synchronization capabilities
• Comprehensive audit logging mechanisms
• Integration with Electronic Health Record (EHR) systems

Scaling Considerations

As your medical database grows, consider implementing:

• Database indexing strategies for improved performance
• Caching mechanisms for frequently accessed data
• Load balancing for high-availability deployments
• Backup and disaster recovery procedures
• Compliance monitoring and reporting tools

Conclusion: Building Secure Healthcare Systems

This tutorial has provided a comprehensive foundation for building medical databases with Docker. However, remember that production healthcare systems require additional security measures and compliance considerations.

Therefore, always consult with security professionals and legal experts when handling sensitive medical data. Additionally, stay updated with the latest security best practices and regulatory requirements in your jurisdiction.