hl7apy

HL7apy supports custom Z elements:

• Z Messages: Created by providing a name where both parts of the trigger event start with 'Z' (e.g., ZBE_Z01).
• Z Segments: Segments with names starting with 'Z' (e.g., ZIN).
• Z Fields: Fields belonging to a Z segment, named using the segment name and position (e.g., ZIN_1).

By default, a Z field's datatype is ST. If a value assigned to a Z field contains components (e.g., abc^def), the datatype is automatically converted to None.

When encoding to ER7, segments and groups are encoded in the order they were created.

from hl7apy.core import Message, Segment, Field, Group

# Create a Z Message
m = Message('ZBE_Z01')

# Add standard segments and Z segments
m.pid = 'PID|1||566-554-3423^^^GHH^MR||EVERYMAN^ADAM^A|||M|||2222 HOME STREET^^ANN ARBOR^MI^^USA||555-555-2004~444-333-222|||M\r'
m.zin = 'ZIN|aa|bb|cc'

# Add a group
m.add(Group('ADT_A01_INSURANCE'))

# Encoding order follows creation order
print(m.to_er7())

# Z Segment and Field usage
s = Segment('ZIN')
s.zin_1 = 'abc'
s.add_field('zin_2')
z_field = Field('ZIN_3', datatype='CX')
s.add(z_field)

HL7apy supports two validation levels:

• STRICT: The message must completely adhere to HL7 structures. It checks child names, cardinality (mandatory vs optional), and value constraints (e.g., date formats or string lengths). In this mode, instantiating unknown elements or violating structure (like adding a second MSH segment) will raise exceptions.
• TOLERANT: The library does not perform strict structural or constraint checks during instantiation. To check compliance in this mode, you must explicitly call the .validate() method on an element.

When a message is created using a message profile, validation is performed against that profile instead of the standard.

from hl7apy.core import Message
from hl7apy.consts import VALIDATION_LEVEL

# STRICT mode example
m = Message("ADT_A01", validation_level=VALIDATION_LEVEL.STRICT)
# This would raise MaxChildLimitReached
# m.add_segment('MSH') 

# TOLERANT mode example
m = Message("ADT_A01")
# Manually trigger validation
# m.validate() 

# Validate and save a report
m.validate(report_file='report')

HL7apy supports base and complex datatypes. Datatypes can be applied to Field, Component, and SubComponent (though SubComponent only supports base datatypes).

• Naming Convention: If a name is specified, it must follow the pattern <complex_datatype>_<position>. If not specified, the name defaults to the datatype name.
• Validation: The library provides base datatype classes (e.g., ST, NM, DTM) that validate values upon assignment. If a value violates the specification (e.g., exceeding max length or invalid format), an exception is raised.
• SubComponents with Datatypes: A SubComponent's .value can be an instance of a base datatype.
• WD Datatype: The WD (Withdrawn) identifier is used to mark fields withdrawn by the specification; presence of these fields causes strict validation to fail.

from hl7apy.v2_4 import ST, NM, DTM

# Validation examples
s = ST('some information')
n = NM(111)
d = DTM('20131010')

# SubComponent with a base datatype value
s_sub = SubComponent(datatype="FT")
s_sub.value = ST('some information')

To handle incoming HL7 messages over MLLP, use the hl7apy.mllp module. You must implement handlers for different message types and register them in a handlers dictionary.

• AbstractHandler: Subclass this to define how to process an incoming message. Implement the reply() method, which should return the response message (e.g., an RSP message) converted via .to_mllp().
• AbstractErrorHandler: Subclass this to handle exceptions (like UnsupportedMessageType) using the .exc attribute.

Handlers can also accept custom arguments passed via the handlers dictionary.

from hl7apy.parser import parse_message
from hl7apy.mllp import AbstractHandler, MLLPServer, AbstractErrorHandler, UnsupportedMessageType
from hl7apy.core import Message

# 1. Define a handler for a specific message type
class PDQHandler(AbstractHandler):
    def __init__(self, msg, database_name):
        super().__init__(msg)
        self.database_name = database_name

    def reply(self):
        msg = parse_message(self.incoming_message)
        # ... process message using self.database_name ...
        res = Message('RSP_K21')
        return res.to_mllp()

# 2. Define an error handler
class ErrorHandler(AbstractErrorHandler):
    def reply(self):
        if isinstance(self.exc, UnsupportedMessageType):
            # Custom response for unsupported types
            pass

# 3. Register handlers and start server
handlers = {
    'QBP^Q22^QBP_Q21': (PDQHandler, 'demographic_db'),
    'ERR': (ErrorHandler,)
}

server = MLLPServer('localhost', 2575, handlers)

You can use custom message profiles instead of standard HL7 structures.

1. Generate the profile file: Use the hl7apy_profile_parser utility script with an XML static definition of your profile.
2. Load the profile: Use load_message_profile to create a profile object.
3. Apply the profile: Pass the profile object to the reference argument when instantiating a Message or to the message_profile argument when calling parse_message.

Important: When using a profile, you must populate the message using traversal or specific Element methods (like add_group, add_segment, or add_field) rather than the generic add() method. This ensures the library uses the profile's specific structure instead of the official HL7 structure.

# Create the profile file from an XML definition
python hl7apy_profile_parser message_profile.xml -o $HOME/message_profile

from hl7apy import load_message_profile, parse_message
from hl7apy.core import Message

# Load the profile
mp = load_message_profile('$HOME/message_profile')

# Use it for a new message
m1 = Message('RSP_K21', reference=mp)

# Use it for parsing
m2 = parse_message(er7_str, message_profile=mp)

# Correct way to populate to respect profile structure:
m.pid.pid_3.cwe_1 = 'aaa'  # via traversal
pid_3 = m.pid.add_field('PID_3')  # via specific method

You can navigate the HL7 hierarchy using several methods:

• Traversal Styles: Access children by name (s.pid_5), long name (s.patient_name), or position (s.pid_5.pid_5_1). Traversal is case-insensitive.
• Handling Repetitions: By default, accessing a child returns the first instance. To access subsequent instances, use index notation: s.pid_13[1].
• Iterating:
  • Use .children to iterate over all direct children of an element.
  • Iterating over a named child (e.g., for spm in m.spm:) will yield all repetitions of that child.
• Deletion: Use the del keyword to remove a child (e.g., del m.MSA or del m.spm[1].spm_1).
• Missing Elements: During traversal, accessing a valid but uncreated child returns an empty list []. Accessing a child that does not exist in the specification raises an exception.

# Accessing by name, long name, or position
s.pid_5
s.patient_name
s.pid_5.pid_5_1

# Accessing repetitions
s.pid_13[1]

# Iterating
for child in m.children:
    pass

for spm in m.spm:
    pass

# Deleting
del m.MSA