jonas/alpinebits_python
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 4 Wiki Activity

trying to figure out best project structure

Browse Source
This commit is contained in:
Jonas Linter
2025-09-25 09:36:07 +02:00
parent d41084cd1b
commit 9ea09ffa3f
12 changed files with 495 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
src/alpine_bits_python/__init__.py Normal file
View File

178
src/alpine_bits_python/alpinebits_guestrequests.py Normal file
View File

@@ -0,0 +1,178 @@
import xml.etree.ElementTree as ET
from datetime import datetime, timezone
from typing import List, Optional
# TimeSpan class according to XSD: <TimeSpan Start="..." End="..." Duration="..." StartWindow="..." EndWindow="..."/>
class TimeSpan:
def __init__(
self,
start: str,
end: str = None,
duration: str = None,
start_window: str = None,
end_window: str = None,
):
self.start = start
self.end = end
self.duration = duration
self.start_window = start_window
self.end_window = end_window
def to_xml(self):
attrib = {"Start": self.start}
if self.end:
attrib["End"] = self.end
if self.duration:
attrib["Duration"] = self.duration
if self.start_window:
attrib["StartWindow"] = self.start_window
if self.end_window:
attrib["EndWindow"] = self.end_window
return ET.Element(_ns("TimeSpan"), attrib)
NAMESPACE = "http://www.opentravel.org/OTA/2003/05"
ET.register_namespace("", NAMESPACE)
def _ns(tag):
return f"{{{NAMESPACE}}}{tag}"
class ResGuest:
def __init__(
self,
given_name: str,
surname: str,
gender: Optional[str] = None,
birth_date: Optional[str] = None,
language: Optional[str] = None,
name_prefix: Optional[str] = None,
name_title: Optional[str] = None,
email: Optional[str] = None,
address: Optional[dict] = None,
telephones: Optional[list] = None,
):
self.given_name = given_name
self.surname = surname
self.gender = gender
self.birth_date = birth_date
self.language = language
self.name_prefix = name_prefix
self.name_title = name_title
self.email = email
self.address = address or {}
self.telephones = telephones or []
def to_xml(self):
resguest_elem = ET.Element(_ns("ResGuest"))
profiles_elem = ET.SubElement(resguest_elem, _ns("Profiles"))
profileinfo_elem = ET.SubElement(profiles_elem, _ns("ProfileInfo"))
profile_elem = ET.SubElement(profileinfo_elem, _ns("Profile"))
customer_elem = ET.SubElement(profile_elem, _ns("Customer"))
if self.gender:
customer_elem.set("Gender", self.gender)
if self.birth_date:
customer_elem.set("BirthDate", self.birth_date)
if self.language:
customer_elem.set("Language", self.language)
personname_elem = ET.SubElement(customer_elem, _ns("PersonName"))
if self.name_prefix:
ET.SubElement(personname_elem, _ns("NamePrefix")).text = self.name_prefix
ET.SubElement(personname_elem, _ns("GivenName")).text = self.given_name
ET.SubElement(personname_elem, _ns("Surname")).text = self.surname
if self.name_title:
ET.SubElement(personname_elem, _ns("NameTitle")).text = self.name_title
for tel in self.telephones:
tel_elem = ET.SubElement(customer_elem, _ns("Telephone"))
for k, v in tel.items():
tel_elem.set(k, v)
if self.email:
ET.SubElement(customer_elem, _ns("Email")).text = self.email
if self.address:
address_elem = ET.SubElement(customer_elem, _ns("Address"))
for k, v in self.address.items():
if k == "CountryName":
country_elem = ET.SubElement(address_elem, _ns("CountryName"))
if isinstance(v, dict):
for ck, cv in v.items():
country_elem.set(ck, cv)
else:
country_elem.text = v
else:
ET.SubElement(address_elem, _ns(k)).text = v
return resguest_elem
def __str__(self):
from lxml import etree
elem = self.to_xml()
xml_bytes = ET.tostring(elem, encoding="utf-8")
parser = etree.XMLParser(remove_blank_text=True)
lxml_elem = etree.fromstring(xml_bytes, parser)
return etree.tostring(lxml_elem, pretty_print=True, encoding="unicode")
class RoomStay:
def __init__(self, room_type: str, timespan: TimeSpan, guests: List[ResGuest]):
self.room_type = room_type
self.timespan = timespan
self.guests = guests
def to_xml(self):
roomstay_elem = ET.Element(_ns("RoomStay"))
ET.SubElement(roomstay_elem, _ns("RoomType")).set(
"RoomTypeCode", self.room_type
)
roomstay_elem.append(self.timespan.to_xml())
guests_elem = ET.SubElement(roomstay_elem, _ns("Guests"))
for guest in self.guests:
guests_elem.append(guest.to_xml())
return roomstay_elem
def __str__(self):
from lxml import etree
elem = self.to_xml()
xml_bytes = ET.tostring(elem, encoding="utf-8")
parser = etree.XMLParser(remove_blank_text=True)
lxml_elem = etree.fromstring(xml_bytes, parser)
return etree.tostring(lxml_elem, pretty_print=True, encoding="unicode")
class Reservation:
def __init__(
self,
reservation_id: str,
hotel_code: str,
roomstays: List[RoomStay],
create_time: Optional[str] = None,
):
self.reservation_id = reservation_id
self.hotel_code = hotel_code
self.roomstays = roomstays
self.create_time = create_time or datetime.now(timezone.utc).isoformat()
def to_xml(self):
res_elem = ET.Element(_ns("HotelReservation"))
uniqueid_elem = ET.SubElement(res_elem, _ns("UniqueID"))
uniqueid_elem.set("Type", "14")
uniqueid_elem.set("ID", self.reservation_id)
hotel_elem = ET.SubElement(res_elem, _ns("Hotel"))
hotel_elem.set("HotelCode", self.hotel_code)
roomstays_elem = ET.SubElement(res_elem, _ns("RoomStays"))
for rs in self.roomstays:
roomstays_elem.append(rs.to_xml())
res_elem.set("CreateDateTime", self.create_time)
return res_elem
def to_xml_string(self):
root = ET.Element(
_ns("OTA_ResRetrieveRS"),
{"Version": "2024-10", "TimeStamp": datetime.now(timezone.utc).isoformat()},
)
success_elem = ET.SubElement(root, _ns("Success"))
reservations_list = ET.SubElement(root, _ns("ReservationsList"))
reservations_list.append(self.to_xml())
return ET.tostring(root, encoding="utf-8", xml_declaration=True).decode("utf-8")

401
src/alpine_bits_python/alpinebits_server.py Normal file
View File

@@ -0,0 +1,401 @@
"""
AlpineBits Server for handling hotel data exchange.
This module provides an asynchronous AlpineBits server that can handle various
OTA (OpenTravel Alliance) actions for hotel data exchange. Currently implements
handshaking functionality with configurable supported actions and capabilities.
"""
import asyncio
import json
from typing import Dict, List, Optional, Any
from xml.etree import ElementTree as ET
from generated.alpinebits import OtaPingRq, OtaPingRs, WarningStatus
from xsdata_pydantic.bindings import XmlSerializer
from xsdata.formats.dataclass.serializers.config import SerializerConfig
class ServerCapabilities:
pass
class AlpineBitsServer:
"""
Asynchronous AlpineBits server for handling hotel data exchange requests.
This server handles various OTA actions and implements the AlpineBits protocol
for hotel data exchange. It maintains a registry of supported actions and
their capabilities, and can respond to handshake requests with its capabilities.
"""
def __init__(self):
"""Initialize the AlpineBits server with default supported actions."""
# Define supported versions and their capabilities
self.supported_versions = {
"2024-10": {
"actions": [
{
"action": "action_OTA_Read"
},
{
"action": "action_OTA_HotelResNotif_GuestRequests"
},
{
"action": "action_OTA_HotelResNotif_GuestRequests_StatusUpdate"
}
]
},
"2022-10": {
"actions": [
{
"action": "action_OTA_Read"
},
{
"action": "action_OTA_HotelResNotif_GuestRequests"
},
{
"action": "action_OTA_HotelResNotif_GuestRequests_StatusUpdate"
}
]
}
}
# XML serializer configuration
self.serializer_config = SerializerConfig(
pretty_print=True,
xml_declaration=True,
encoding="UTF-8"
)
self.xml_serializer = XmlSerializer(config=self.serializer_config)
# Namespace map for XML serialization
self.ns_map = {
None: "http://www.opentravel.org/OTA/2003/05",
"xsi": "http://www.w3.org/2001/XMLSchema-instance"
}
def add_supported_action(self, version: str, action: str, supports: Optional[List[str]] = None):
"""
Add a supported action to a specific version.
Args:
version: AlpineBits version (e.g., "2024-10")
action: Action name (e.g., "action_OTA_HotelInvCountNotif")
supports: List of supported features for this action (optional)
"""
if version not in self.supported_versions:
self.supported_versions[version] = {"actions": []}
action_dict = {"action": action}
if supports:
action_dict["supports"] = supports
self.supported_versions[version]["actions"].append(action_dict)
def remove_supported_action(self, version: str, action: str):
"""
Remove a supported action from a specific version.
Args:
version: AlpineBits version
action: Action name to remove
"""
if version in self.supported_versions:
actions = self.supported_versions[version]["actions"]
self.supported_versions[version]["actions"] = [
a for a in actions if a["action"] != action
]
def is_action_supported(self, action: str, version: str = None) -> bool:
"""
Check if an action is supported by the server.
Args:
action: Action name to check
version: Specific version to check (if None, checks all versions)
Returns:
True if action is supported, False otherwise
"""
if version:
if version not in self.supported_versions:
return False
actions = self.supported_versions[version]["actions"]
return any(a["action"] == action for a in actions)
# Check all versions
for ver_data in self.supported_versions.values():
if any(a["action"] == action for a in ver_data["actions"]):
return True
return False
def get_capabilities_json(self) -> str:
"""
Get the server capabilities as JSON string for handshake responses.
Returns:
JSON string containing supported versions and actions
"""
capabilities = {
"versions": [
{
"version": version,
"actions": version_data["actions"]
}
for version, version_data in self.supported_versions.items()
]
}
return json.dumps(capabilities, indent=4)
async def handle_request(self, action: str, request_xml: str) -> str:
"""
Handle an incoming AlpineBits request.
Args:
action: The action to perform (e.g., "action_OTA_Ping")
request_xml: The XML request body
Returns:
XML response string
Raises:
ValueError: If the action is not supported
"""
# Parse the action to determine the handler method
if action == "action_OTA_Ping":
return await self.handle_ping(request_xml)
elif action == "action_OTA_Read":
return await self.handle_read(request_xml)
elif action == "action_OTA_HotelResNotif_GuestRequests":
return await self.handle_guest_requests(request_xml)
elif action == "action_OTA_HotelResNotif_GuestRequests_StatusUpdate":
return await self.handle_guest_requests_status_update(request_xml)
else:
# Return error response for unsupported actions
return await self.handle_unsupported_action(action, request_xml)
async def handle_ping(self, request_xml: str) -> str:
"""
Handle OTA_Ping requests (handshake).
Args:
request_xml: The OTA_PingRQ XML request
Returns:
OTA_PingRS XML response with server capabilities
"""
try:
# Parse the incoming request to extract any version information
root = ET.fromstring(request_xml)
version = root.get("Version", "8.000")
# Extract echo data if present
echo_data_elem = root.find(".//{http://www.opentravel.org/OTA/2003/05}EchoData")
echo_data = None
if echo_data_elem is not None and echo_data_elem.text:
echo_data = echo_data_elem.text.strip()
# Get capabilities JSON
capabilities_json = self.get_capabilities_json()
# Create warning with capabilities
warning = OtaPingRs.Warnings.Warning(
type_value="11",
status=WarningStatus.ALPINEBITS_HANDSHAKE,
content=[capabilities_json]
)
warnings = OtaPingRs.Warnings(warning=[warning])
# Create successful ping response
ping_response = OtaPingRs(
version=version,
success=None, # Empty success element
warnings=warnings,
echo_data=echo_data or capabilities_json
)
# Serialize to XML
return self.xml_serializer.render(ping_response, ns_map=self.ns_map)
except Exception as e:
# Return error response if something goes wrong
return await self.create_error_response(
"OTA_PingRS",
"8.000",
f"Failed to process ping request: {str(e)}"
)
async def handle_read(self, request_xml: str) -> str:
"""
Handle OTA_Read requests.
Args:
request_xml: The OTA_ReadRQ XML request
Returns:
XML response (placeholder implementation)
"""
# Placeholder implementation - return unsupported for now
return await self.create_error_response(
"OTA_ReadRS",
"8.000",
"OTA_Read action not yet implemented"
)
async def handle_guest_requests(self, request_xml: str) -> str:
"""
Handle guest request notifications.
Args:
request_xml: The guest request XML
Returns:
XML response (placeholder implementation)
"""
# Placeholder implementation - return unsupported for now
return await self.create_error_response(
"OTA_HotelResNotifRS",
"8.000",
"Guest requests action not yet implemented"
)
async def handle_guest_requests_status_update(self, request_xml: str) -> str:
"""
Handle guest request status updates.
Args:
request_xml: The status update XML
Returns:
XML response (placeholder implementation)
"""
# Placeholder implementation - return unsupported for now
return await self.create_error_response(
"OTA_HotelResNotifRS",
"8.000",
"Guest request status updates not yet implemented"
)
async def handle_unsupported_action(self, action: str, request_xml: str) -> str:
"""
Handle unsupported actions by returning an appropriate error response.
Args:
action: The unsupported action name
request_xml: The request XML
Returns:
XML error response
"""
return await self.create_error_response(
"OTA_PingRS", # Use generic ping response for unknown actions
"8.000",
f"Action '{action}' is not supported by this server"
)
async def create_error_response(self, response_type: str, version: str, error_message: str) -> str:
"""
Create a generic error response.
Args:
response_type: The response type (e.g., "OTA_PingRS")
version: Version number
error_message: Error message to include
Returns:
XML error response string
"""
# Create a basic error response structure
error_xml = f'''<?xml version="1.0" encoding="UTF-8"?>
<{response_type} xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns="http://www.opentravel.org/OTA/2003/05"
xsi:schemaLocation="http://www.opentravel.org/OTA/2003/05 {response_type}.xsd"
Version="{version}">
<Errors>
<Error Type="13" Code="450">{error_message}</Error>
</Errors>
</{response_type}>'''
return error_xml
def get_supported_actions(self, version: str = None) -> List[str]:
"""
Get list of supported action names.
Args:
version: Specific version to get actions for (if None, returns all)
Returns:
List of supported action names
"""
actions = set()
if version:
if version in self.supported_versions:
for action_data in self.supported_versions[version]["actions"]:
actions.add(action_data["action"])
else:
for version_data in self.supported_versions.values():
for action_data in version_data["actions"]:
actions.add(action_data["action"])
return sorted(list(actions))
def get_supported_versions(self) -> List[str]:
"""
Get list of supported AlpineBits versions.
Returns:
List of supported version strings
"""
return list(self.supported_versions.keys())
# Example usage and testing
async def main():
"""Example usage of the AlpineBits server."""
server = AlpineBitsServer()
# Add additional supported actions
server.add_supported_action(
"2024-10",
"action_OTA_HotelInvCountNotif",
[
"OTA_HotelInvCountNotif_accept_rooms",
"OTA_HotelInvCountNotif_accept_categories",
"OTA_HotelInvCountNotif_accept_deltas"
]
)
# Example ping request
ping_request = '''<?xml version="1.0" encoding="UTF-8"?>
<OTA_PingRQ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns="http://www.opentravel.org/OTA/2003/05"
xsi:schemaLocation="http://www.opentravel.org/OTA/2003/05 OTA_PingRQ.xsd"
Version="8.000">
<EchoData>{"test": "handshake request"}</EchoData>
</OTA_PingRQ>'''
# Handle the request
try:
response = await server.handle_request("action_OTA_Ping", ping_request)
print("Response:")
print(response)
# Test unsupported action
print("\n--- Testing unsupported action ---")
unsupported_response = await server.handle_request("action_NOT_SUPPORTED", ping_request)
print("Unsupported action response:")
print(unsupported_response)
except Exception as e:
print(f"Error: {e}")
if __name__ == "__main__":
asyncio.run(main())

161
src/alpine_bits_python/generated/__init__.py Normal file
View File

@@ -0,0 +1,161 @@
from generated.alpinebits import (
BaseByGuestAmtType,
BookingRuleCodeContext,
CommentName1,
CommentName2,
ContactInfoLocation,
DefSendComplete,
DescriptionName,
DescriptionTextFormat1,
DescriptionTextFormat2,
DiscountPercent,
ErrorStatus,
ErrorType,
EventIdType,
GuestFirstQualifyingPosition,
HotelReservationResStatus,
ImageItemCategory,
InvCountCountType,
LengthOfStayMinMaxMessageType1,
LengthOfStayMinMaxMessageType2,
LengthOfStayTimeUnit,
MealsIncludedMealPlanCodes,
MealsIncludedMealPlanIndicator,
MultimediaDescriptionInfoCode1,
MultimediaDescriptionInfoCode2,
OccupancyAgeQualifyingCode,
OtaHotelDescriptiveContentNotifRq,
OtaHotelDescriptiveContentNotifRs,
OtaHotelDescriptiveInfoRq,
OtaHotelDescriptiveInfoRs,
OtaHotelInvCountNotifRq,
OtaHotelInvCountNotifRs,
OtaHotelPostEventNotifRq,
OtaHotelPostEventNotifRs,
OtaHotelRatePlanNotifRq,
OtaHotelRatePlanNotifRs,
OtaHotelRatePlanRq,
OtaHotelRatePlanRs,
OtaHotelResNotifRq,
OtaHotelResNotifRs,
OtaNotifReportRq,
OtaNotifReportRs,
OtaPingRq,
OtaPingRs,
OtaReadRq,
OtaResRetrieveRs,
PositionAltitudeUnitOfMeasureCode,
PrerequisiteInventoryInvType,
ProfileProfileType,
RateDescriptionName,
RatePlanRatePlanNotifType,
RatePlanRatePlanType,
RateRateTimeUnit,
RestrictionStatusRestriction,
RestrictionStatusStatus,
RoomTypeRoomType,
ServiceMealPlanCode,
ServiceServiceCategoryCode,
ServiceServicePricingType,
ServiceType,
SpecialRequestCodeContext,
StayRequirementStayContext,
SupplementAddToBasicRateIndicator,
SupplementChargeTypeCode,
SupplementInvType,
TaxPolicyChargeFrequency,
TaxPolicyChargeUnit,
TaxPolicyCode,
TextTextFormat1,
TextTextFormat2,
TimeUnitType,
TypeRoomRoomType,
UniqueIdInstance,
UniqueIdType1,
UniqueIdType2,
UniqueIdType3,
UrlType,
VideoItemCategory,
WarningStatus,
)
__all__ = [
"BaseByGuestAmtType",
"BookingRuleCodeContext",
"CommentName1",
"CommentName2",
"ContactInfoLocation",
"DescriptionName",
"DescriptionTextFormat1",
"DescriptionTextFormat2",
"DiscountPercent",
"ErrorStatus",
"ErrorType",
"EventIdType",
"GuestFirstQualifyingPosition",
"HotelReservationResStatus",
"ImageItemCategory",
"InvCountCountType",
"LengthOfStayMinMaxMessageType1",
"LengthOfStayMinMaxMessageType2",
"LengthOfStayTimeUnit",
"MealsIncludedMealPlanCodes",
"MealsIncludedMealPlanIndicator",
"MultimediaDescriptionInfoCode1",
"MultimediaDescriptionInfoCode2",
"OtaHotelDescriptiveContentNotifRq",
"OtaHotelDescriptiveContentNotifRs",
"OtaHotelDescriptiveInfoRq",
"OtaHotelDescriptiveInfoRs",
"OtaHotelInvCountNotifRq",
"OtaHotelInvCountNotifRs",
"OtaHotelPostEventNotifRq",
"OtaHotelPostEventNotifRs",
"OtaHotelRatePlanNotifRq",
"OtaHotelRatePlanNotifRs",
"OtaHotelRatePlanRq",
"OtaHotelRatePlanRs",
"OtaHotelResNotifRq",
"OtaHotelResNotifRs",
"OtaNotifReportRq",
"OtaNotifReportRs",
"OtaPingRq",
"OtaPingRs",
"OtaReadRq",
"OtaResRetrieveRs",
"OccupancyAgeQualifyingCode",
"PositionAltitudeUnitOfMeasureCode",
"PrerequisiteInventoryInvType",
"ProfileProfileType",
"RateDescriptionName",
"RatePlanRatePlanNotifType",
"RatePlanRatePlanType",
"RateRateTimeUnit",
"RestrictionStatusRestriction",
"RestrictionStatusStatus",
"RoomTypeRoomType",
"ServiceMealPlanCode",
"ServiceServiceCategoryCode",
"ServiceServicePricingType",
"ServiceType",
"SpecialRequestCodeContext",
"StayRequirementStayContext",
"SupplementAddToBasicRateIndicator",
"SupplementChargeTypeCode",
"SupplementInvType",
"TaxPolicyChargeFrequency",
"TaxPolicyChargeUnit",
"TaxPolicyCode",
"TextTextFormat1",
"TextTextFormat2",
"TimeUnitType",
"TypeRoomRoomType",
"UrlType",
"UniqueIdInstance",
"UniqueIdType1",
"UniqueIdType2",
"UniqueIdType3",
"VideoItemCategory",
"WarningStatus",
"DefSendComplete",
]

13613
src/alpine_bits_python/generated/alpinebits.py Normal file
View File

File diff suppressed because it is too large Load Diff

195
src/alpine_bits_python/main.py Normal file
View File

@@ -0,0 +1,195 @@
from .alpinebits_guestrequests import ResGuest, RoomStay
import generated.alpinebits as ab
from io import BytesIO
import sys
from datetime import datetime, timezone
import re
from xsdata_pydantic.bindings import XmlSerializer
from simplified_access import (
CommentData,
CommentsData,
CommentListItemData,
CustomerData,
HotelReservationIdData,
PhoneTechType,
AlpineBitsFactory,
OtaMessageType
)
def main():
# Success - use None instead of object() for cleaner XML output
success = None
# UniqueID
unique_id = ab.OtaResRetrieveRs.ReservationsList.HotelReservation.UniqueId(
type_value=ab.UniqueIdType2.VALUE_14, id="6b34fe24ac2ff811"
)
# TimeSpan - use the actual nested class
start_date_window = ab.OtaResRetrieveRs.ReservationsList.HotelReservation.RoomStays.RoomStay.TimeSpan.StartDateWindow(
earliest_date="2024-10-01", latest_date="2024-10-02"
)
time_span = ab.OtaResRetrieveRs.ReservationsList.HotelReservation.RoomStays.RoomStay.TimeSpan(
start_date_window=start_date_window
)
# RoomStay with TimeSpan
room_stay = (
ab.OtaResRetrieveRs.ReservationsList.HotelReservation.RoomStays.RoomStay(
time_span=time_span
)
)
room_stays = ab.OtaResRetrieveRs.ReservationsList.HotelReservation.RoomStays(
room_stay=[room_stay]
)
customer_data = CustomerData(
given_name="John",
surname="Doe",
name_prefix="Mr.",
phone_numbers=[
("+1234567890", PhoneTechType.MOBILE), # Phone number with type
("+0987654321", None), # Phone number without type
],
email_address="john.doe@example.com",
email_newsletter=True,
address_line="123 Main Street",
city_name="Anytown",
postal_code="12345",
country_code="US",
address_catalog=False,
gender="Male",
birth_date="1980-01-01",
language="en",
)
alpine_bits_factory = AlpineBitsFactory()
res_guests = alpine_bits_factory.create_res_guests(customer_data, OtaMessageType.RETRIEVE)
hotel_res_id_data = HotelReservationIdData(
res_id_type="13",
res_id_value=None,
res_id_source=None,
res_id_source_context="99tales",
)
# Create HotelReservationId using the factory
hotel_res_id = alpine_bits_factory.create(hotel_res_id_data, OtaMessageType.RETRIEVE)
# Use the actual nested HotelReservationIds class
hotel_res_ids = ab.OtaResRetrieveRs.ReservationsList.HotelReservation.ResGlobalInfo.HotelReservationIds(
hotel_reservation_id=[hotel_res_id]
)
# Basic property info
basic_property_info = ab.OtaResRetrieveRs.ReservationsList.HotelReservation.ResGlobalInfo.BasicPropertyInfo(
hotel_code="123", hotel_name="Frangart Inn"
)
comment = CommentData(
name= ab.CommentName2.CUSTOMER_COMMENT,
text="This is a sample comment.",
list_items=[CommentListItemData(
value="Landing page comment",
language="en",
list_item="1",
)],
)
comment2 = CommentData(
name= ab.CommentName2.ADDITIONAL_INFO,
text="This is a special request comment.",
)
comments_data = CommentsData(comments=[comment, comment2])
comments = alpine_bits_factory.create(comments_data, OtaMessageType.RETRIEVE)
# ResGlobalInfo
res_global_info = (
ab.OtaResRetrieveRs.ReservationsList.HotelReservation.ResGlobalInfo(
hotel_reservation_ids=hotel_res_ids, basic_property_info=basic_property_info, comments=comments
)
)
# Hotel Reservation
hotel_reservation = ab.OtaResRetrieveRs.ReservationsList.HotelReservation(
create_date_time=datetime.now(timezone.utc).isoformat(),
res_status=ab.HotelReservationResStatus.REQUESTED,
room_stay_reservation="true",
unique_id=unique_id,
room_stays=room_stays,
res_guests=res_guests,
res_global_info=res_global_info,
)
reservations_list = ab.OtaResRetrieveRs.ReservationsList(
hotel_reservation=[hotel_reservation]
)
# Root element
ota_res_retrieve_rs = ab.OtaResRetrieveRs(
version="7.000", success=success, reservations_list=reservations_list
)
# Serialize using Pydantic's model_dump and convert to XML
try:
# First validate the model
ota_res_retrieve_rs.model_validate(ota_res_retrieve_rs.model_dump())
print("✅ Pydantic validation successful!")
# For XML serialization with Pydantic models, we need to use xsdata-pydantic serializer
from xsdata.formats.dataclass.serializers.config import SerializerConfig
config = SerializerConfig(
pretty_print=True, xml_declaration=True, encoding="UTF-8"
)
serializer = XmlSerializer(config=config)
# Use ns_map to control namespace prefixes - set default namespace
ns_map = {None: "http://www.opentravel.org/OTA/2003/05"}
xml_string = serializer.render(ota_res_retrieve_rs, ns_map=ns_map)
with open("output.xml", "w", encoding="utf-8") as outfile:
outfile.write(xml_string)
print("✅ XML serialization successful!")
print(f"Generated XML written to output.xml")
# Also print the pretty formatted XML to console
print("\n📄 Generated XML:")
print(xml_string)
# Test parsing back
from xsdata_pydantic.bindings import XmlParser
parser = XmlParser()
with open("output.xml", "r", encoding="utf-8") as infile:
xml_content = infile.read()
parsed_result = parser.from_string(xml_content, ab.OtaResRetrieveRs)
print("✅ Round-trip validation successful!")
print(
f"Parsed reservation status: {parsed_result.reservations_list.hotel_reservation[0].res_status}"
)
except Exception as e:
print(f"❌ Validation/Serialization failed: {e}")
if __name__ == "__main__":
main()

56
src/alpine_bits_python/output.xml Normal file
View File

@@ -0,0 +1,56 @@
<?xml version="1.0" encoding="UTF-8"?>
<OTA_ResRetrieveRS xmlns="http://www.opentravel.org/OTA/2003/05" Version="7.000">
<ReservationsList>
<HotelReservation CreateDateTime="2025-09-24T15:07:57.451427+00:00" ResStatus="Requested" RoomStayReservation="true">
<UniqueID Type="14" ID="6b34fe24ac2ff811"/>
<RoomStays>
<RoomStay>
<TimeSpan>
<StartDateWindow EarliestDate="2024-10-01" LatestDate="2024-10-02"/>
</TimeSpan>
</RoomStay>
</RoomStays>
<ResGuests>
<ResGuest>
<Profiles>
<ProfileInfo>
<Profile>
<Customer Gender="Male" BirthDate="1980-01-01" Language="en">
<PersonName>
<NamePrefix>Mr.</NamePrefix>
<GivenName>John</GivenName>
<Surname>Doe</Surname>
</PersonName>
<Telephone PhoneTechType="5" PhoneNumber="+1234567890"/>
<Telephone PhoneNumber="+0987654321"/>
<Email Remark="newsletter:yes">john.doe@example.com</Email>
<Address Remark="catalog:no">
<AddressLine>123 Main Street</AddressLine>
<CityName>Anytown</CityName>
<PostalCode>12345</PostalCode>
<CountryName Code="US"/>
</Address>
</Customer>
</Profile>
</ProfileInfo>
</Profiles>
</ResGuest>
</ResGuests>
<ResGlobalInfo>
<Comments>
<Comment Name="customer comment">
<ListItem ListItem="1" Language="en">Landing page comment</ListItem>
<Text>This is a sample comment.</Text>
</Comment>
<Comment Name="additional info">
<Text>This is a special request comment.</Text>
</Comment>
</Comments>
<HotelReservationIDs>
<HotelReservationID ResID_Type="13" ResID_SourceContext="99tales"/>
</HotelReservationIDs>
<BasicPropertyInfo HotelCode="123" HotelName="Frangart Inn"/>
</ResGlobalInfo>
</HotelReservation>
</ReservationsList>
</OTA_ResRetrieveRS>

740
src/alpine_bits_python/simplified_access.py Normal file
View File

@@ -0,0 +1,740 @@
from typing import Union, Optional, Any, TypeVar
from pydantic import BaseModel, ConfigDict, Field
from dataclasses import dataclass
from enum import Enum
# Import the generated classes
from generated.alpinebits import OtaHotelResNotifRq, OtaResRetrieveRs, CommentName2
# Define type aliases for the two Customer types
NotifCustomer = OtaHotelResNotifRq.HotelReservations.HotelReservation.ResGuests.ResGuest.Profiles.ProfileInfo.Profile.Customer
RetrieveCustomer = OtaResRetrieveRs.ReservationsList.HotelReservation.ResGuests.ResGuest.Profiles.ProfileInfo.Profile.Customer
# Define type aliases for HotelReservationId types
NotifHotelReservationId = OtaHotelResNotifRq.HotelReservations.HotelReservation.ResGlobalInfo.HotelReservationIds.HotelReservationId
RetrieveHotelReservationId = OtaResRetrieveRs.ReservationsList.HotelReservation.ResGlobalInfo.HotelReservationIds.HotelReservationId
# Define type aliases for Comments types
NotifComments = OtaHotelResNotifRq.HotelReservations.HotelReservation.ResGlobalInfo.Comments
RetrieveComments = OtaResRetrieveRs.ReservationsList.HotelReservation.ResGlobalInfo.Comments
NotifComment = OtaHotelResNotifRq.HotelReservations.HotelReservation.ResGlobalInfo.Comments.Comment
RetrieveComment = OtaResRetrieveRs.ReservationsList.HotelReservation.ResGlobalInfo.Comments.Comment
# phonetechtype enum 1,3,5 voice, fax, mobile
class PhoneTechType(Enum):
VOICE = "1"
FAX = "3"
MOBILE = "5"
# Enum to specify which OTA message type to use
class OtaMessageType(Enum):
NOTIF = "notification" # For OtaHotelResNotifRq
RETRIEVE = "retrieve" # For OtaResRetrieveRs
@dataclass
class CustomerData:
"""Simple data class to hold customer information without nested type constraints."""
given_name: str
surname: str
name_prefix: None | str = None
name_title: None | str = None
phone_numbers: list[tuple[str, None | PhoneTechType]] = (
None # (phone_number, phone_tech_type)
)
email_address: None | str = None
email_newsletter: None | bool = (
None # True for "yes", False for "no", None for not specified
)
address_line: None | str = None
city_name: None | str = None
postal_code: None | str = None
country_code: None | str = None # Two-letter country code
address_catalog: None | bool = (
None # True for "yes", False for "no", None for not specified
)
gender: None | str = None # "Unknown", "Male", "Female"
birth_date: None | str = None
language: None | str = None # Two-letter language code
def __post_init__(self):
if self.phone_numbers is None:
self.phone_numbers = []
class CustomerFactory:
"""Factory class to create Customer instances for both OtaHotelResNotifRq and OtaResRetrieveRs."""
@staticmethod
def create_notif_customer(data: CustomerData) -> NotifCustomer:
"""Create a Customer for OtaHotelResNotifRq."""
return CustomerFactory._create_customer(NotifCustomer, data)
@staticmethod
def create_retrieve_customer(data: CustomerData) -> RetrieveCustomer:
"""Create a Customer for OtaResRetrieveRs."""
return CustomerFactory._create_customer(RetrieveCustomer, data)
@staticmethod
def _create_customer(customer_class: type, data: CustomerData) -> Any:
"""Internal method to create a customer of the specified type."""
# Create PersonName
person_name = customer_class.PersonName(
given_name=data.given_name,
surname=data.surname,
name_prefix=data.name_prefix,
name_title=data.name_title,
)
# Create telephone list
telephones = []
for phone_number, phone_tech_type in data.phone_numbers:
telephone = customer_class.Telephone(
phone_number=phone_number,
phone_tech_type=phone_tech_type.value if phone_tech_type else None,
)
telephones.append(telephone)
# Create email if provided
email = None
if data.email_address:
remark = None
if data.email_newsletter is not None:
remark = f"newsletter:{'yes' if data.email_newsletter else 'no'}"
email = customer_class.Email(value=data.email_address, remark=remark)
# Create address if any address fields are provided
address = None
if any(
[data.address_line, data.city_name, data.postal_code, data.country_code]
):
country_name = None
if data.country_code:
country_name = customer_class.Address.CountryName(
code=data.country_code
)
address_remark = None
if data.address_catalog is not None:
address_remark = f"catalog:{'yes' if data.address_catalog else 'no'}"
address = customer_class.Address(
address_line=data.address_line,
city_name=data.city_name,
postal_code=data.postal_code,
country_name=country_name,
remark=address_remark,
)
# Create the customer
return customer_class(
person_name=person_name,
telephone=telephones,
email=email,
address=address,
gender=data.gender,
birth_date=data.birth_date,
language=data.language,
)
@staticmethod
def from_notif_customer(customer: NotifCustomer) -> CustomerData:
"""Convert a NotifCustomer back to CustomerData."""
return CustomerFactory._customer_to_data(customer)
@staticmethod
def from_retrieve_customer(customer: RetrieveCustomer) -> CustomerData:
"""Convert a RetrieveCustomer back to CustomerData."""
return CustomerFactory._customer_to_data(customer)
@staticmethod
def _customer_to_data(customer: Any) -> CustomerData:
"""Internal method to convert any customer type to CustomerData."""
# Extract phone numbers
phone_numbers = []
if customer.telephone:
for tel in customer.telephone:
phone_numbers.append(
(
tel.phone_number,
PhoneTechType(tel.phone_tech_type)
if tel.phone_tech_type
else None,
)
)
# Extract email info
email_address = None
email_newsletter = None
if customer.email:
email_address = customer.email.value
if customer.email.remark:
if "newsletter:yes" in customer.email.remark:
email_newsletter = True
elif "newsletter:no" in customer.email.remark:
email_newsletter = False
# Extract address info
address_line = None
city_name = None
postal_code = None
country_code = None
address_catalog = None
if customer.address:
address_line = customer.address.address_line
city_name = customer.address.city_name
postal_code = customer.address.postal_code
if customer.address.country_name:
country_code = customer.address.country_name.code
if customer.address.remark:
if "catalog:yes" in customer.address.remark:
address_catalog = True
elif "catalog:no" in customer.address.remark:
address_catalog = False
return CustomerData(
given_name=customer.person_name.given_name,
surname=customer.person_name.surname,
name_prefix=customer.person_name.name_prefix,
name_title=customer.person_name.name_title,
phone_numbers=phone_numbers,
email_address=email_address,
email_newsletter=email_newsletter,
address_line=address_line,
city_name=city_name,
postal_code=postal_code,
country_code=country_code,
address_catalog=address_catalog,
gender=customer.gender,
birth_date=customer.birth_date,
language=customer.language,
)
@dataclass
class HotelReservationIdData:
"""Simple data class to hold hotel reservation ID information without nested type constraints."""
res_id_type: str # Required field - pattern: [0-9]+
res_id_value: None | str = None # Max 64 characters
res_id_source: None | str = None # Max 64 characters
res_id_source_context: None | str = None # Max 64 characters
class HotelReservationIdFactory:
"""Factory class to create HotelReservationId instances for both OtaHotelResNotifRq and OtaResRetrieveRs."""
@staticmethod
def create_notif_hotel_reservation_id(
data: HotelReservationIdData,
) -> NotifHotelReservationId:
"""Create a HotelReservationId for OtaHotelResNotifRq."""
return HotelReservationIdFactory._create_hotel_reservation_id(
NotifHotelReservationId, data
)
@staticmethod
def create_retrieve_hotel_reservation_id(
data: HotelReservationIdData,
) -> RetrieveHotelReservationId:
"""Create a HotelReservationId for OtaResRetrieveRs."""
return HotelReservationIdFactory._create_hotel_reservation_id(
RetrieveHotelReservationId, data
)
@staticmethod
def _create_hotel_reservation_id(
hotel_reservation_id_class: type, data: HotelReservationIdData
) -> Any:
"""Internal method to create a hotel reservation id of the specified type."""
return hotel_reservation_id_class(
res_id_type=data.res_id_type,
res_id_value=data.res_id_value,
res_id_source=data.res_id_source,
res_id_source_context=data.res_id_source_context,
)
@staticmethod
def from_notif_hotel_reservation_id(
hotel_reservation_id: NotifHotelReservationId,
) -> HotelReservationIdData:
"""Convert a NotifHotelReservationId back to HotelReservationIdData."""
return HotelReservationIdFactory._hotel_reservation_id_to_data(
hotel_reservation_id
)
@staticmethod
def from_retrieve_hotel_reservation_id(
hotel_reservation_id: RetrieveHotelReservationId,
) -> HotelReservationIdData:
"""Convert a RetrieveHotelReservationId back to HotelReservationIdData."""
return HotelReservationIdFactory._hotel_reservation_id_to_data(
hotel_reservation_id
)
@staticmethod
def _hotel_reservation_id_to_data(
hotel_reservation_id: Any,
) -> HotelReservationIdData:
"""Internal method to convert any hotel reservation id type to HotelReservationIdData."""
return HotelReservationIdData(
res_id_type=hotel_reservation_id.res_id_type,
res_id_value=hotel_reservation_id.res_id_value,
res_id_source=hotel_reservation_id.res_id_source,
res_id_source_context=hotel_reservation_id.res_id_source_context,
)
@dataclass
class CommentListItemData:
"""Simple data class to hold comment list item information."""
value: str # The text content of the list item
list_item: str # Numeric identifier (pattern: [0-9]+)
language: str # Two-letter language code (pattern: [a-z][a-z])
@dataclass
class CommentData:
"""Simple data class to hold comment information without nested type constraints."""
name: CommentName2 # Required: "included services", "customer comment", "additional info"
text: Optional[str] = None # Optional text content
list_items: list[CommentListItemData] = None # Optional list items
def __post_init__(self):
if self.list_items is None:
self.list_items = []
@dataclass
class CommentsData:
"""Simple data class to hold multiple comments (1-3 max)."""
comments: list[CommentData] = None # 1-3 comments maximum
def __post_init__(self):
if self.comments is None:
self.comments = []
class CommentFactory:
"""Factory class to create Comment instances for both OtaHotelResNotifRq and OtaResRetrieveRs."""
@staticmethod
def create_notif_comments(data: CommentsData) -> NotifComments:
"""Create Comments for OtaHotelResNotifRq."""
return CommentFactory._create_comments(NotifComments, NotifComment, data)
@staticmethod
def create_retrieve_comments(data: CommentsData) -> RetrieveComments:
"""Create Comments for OtaResRetrieveRs."""
return CommentFactory._create_comments(RetrieveComments, RetrieveComment, data)
@staticmethod
def _create_comments(comments_class: type, comment_class: type, data: CommentsData) -> Any:
"""Internal method to create comments of the specified type."""
comments_list = []
for comment_data in data.comments:
# Create list items
list_items = []
for item_data in comment_data.list_items:
list_item = comment_class.ListItem(
value=item_data.value,
list_item=item_data.list_item,
language=item_data.language
)
list_items.append(list_item)
# Create comment
comment = comment_class(
name=comment_data.name,
text=comment_data.text,
list_item=list_items
)
comments_list.append(comment)
# Create comments container
return comments_class(comment=comments_list)
@staticmethod
def from_notif_comments(comments: NotifComments) -> CommentsData:
"""Convert NotifComments back to CommentsData."""
return CommentFactory._comments_to_data(comments)
@staticmethod
def from_retrieve_comments(comments: RetrieveComments) -> CommentsData:
"""Convert RetrieveComments back to CommentsData."""
return CommentFactory._comments_to_data(comments)
@staticmethod
def _comments_to_data(comments: Any) -> CommentsData:
"""Internal method to convert any comments type to CommentsData."""
comments_data_list = []
for comment in comments.comment:
# Extract list items
list_items_data = []
if comment.list_item:
for list_item in comment.list_item:
list_items_data.append(CommentListItemData(
value=list_item.value,
list_item=list_item.list_item,
language=list_item.language
))
# Extract comment data
comment_data = CommentData(
name=comment.name,
text=comment.text,
list_items=list_items_data
)
comments_data_list.append(comment_data)
return CommentsData(comments=comments_data_list)
# Define type aliases for ResGuests types
NotifResGuests = OtaHotelResNotifRq.HotelReservations.HotelReservation.ResGuests
RetrieveResGuests = OtaResRetrieveRs.ReservationsList.HotelReservation.ResGuests
class ResGuestFactory:
"""Factory class to create complete ResGuests structures with a primary customer."""
@staticmethod
def create_notif_res_guests(customer_data: CustomerData) -> NotifResGuests:
"""Create a complete ResGuests structure for OtaHotelResNotifRq with primary customer."""
return ResGuestFactory._create_res_guests(
NotifResGuests, NotifCustomer, customer_data
)
@staticmethod
def create_retrieve_res_guests(customer_data: CustomerData) -> RetrieveResGuests:
"""Create a complete ResGuests structure for OtaResRetrieveRs with primary customer."""
return ResGuestFactory._create_res_guests(
RetrieveResGuests, RetrieveCustomer, customer_data
)
@staticmethod
def _create_res_guests(
res_guests_class: type, customer_class: type, customer_data: CustomerData
) -> Any:
"""Internal method to create complete ResGuests structure."""
# Create the customer using the existing CustomerFactory
customer = CustomerFactory._create_customer(customer_class, customer_data)
# Create Profile with the customer
profile = res_guests_class.ResGuest.Profiles.ProfileInfo.Profile(
customer=customer
)
# Create ProfileInfo with the profile
profile_info = res_guests_class.ResGuest.Profiles.ProfileInfo(profile=profile)
# Create Profiles with the profile_info
profiles = res_guests_class.ResGuest.Profiles(profile_info=profile_info)
# Create ResGuest with the profiles
res_guest = res_guests_class.ResGuest(profiles=profiles)
# Create ResGuests with the res_guest
return res_guests_class(res_guest=res_guest)
@staticmethod
def extract_primary_customer(
res_guests: Union[NotifResGuests, RetrieveResGuests],
) -> CustomerData:
"""Extract the primary customer data from a ResGuests structure."""
# Navigate down the nested structure to get the customer
customer = res_guests.res_guest.profiles.profile_info.profile.customer
# Use the existing CustomerFactory conversion method
if isinstance(res_guests, NotifResGuests):
return CustomerFactory.from_notif_customer(customer)
else:
return CustomerFactory.from_retrieve_customer(customer)
class AlpineBitsFactory:
"""Unified factory class for creating AlpineBits objects with a simple interface."""
@staticmethod
def create(data: Union[CustomerData, HotelReservationIdData, CommentsData], message_type: OtaMessageType) -> Any:
"""
Create an AlpineBits object based on the data type and message type.
Args:
data: The data object (CustomerData, HotelReservationIdData, CommentsData, etc.)
message_type: Whether to create for NOTIF or RETRIEVE message types
Returns:
The appropriate AlpineBits object based on the data type and message type
"""
if isinstance(data, CustomerData):
if message_type == OtaMessageType.NOTIF:
return CustomerFactory.create_notif_customer(data)
else:
return CustomerFactory.create_retrieve_customer(data)
elif isinstance(data, HotelReservationIdData):
if message_type == OtaMessageType.NOTIF:
return HotelReservationIdFactory.create_notif_hotel_reservation_id(data)
else:
return HotelReservationIdFactory.create_retrieve_hotel_reservation_id(data)
elif isinstance(data, CommentsData):
if message_type == OtaMessageType.NOTIF:
return CommentFactory.create_notif_comments(data)
else:
return CommentFactory.create_retrieve_comments(data)
else:
raise ValueError(f"Unsupported data type: {type(data)}")
@staticmethod
def create_res_guests(customer_data: CustomerData, message_type: OtaMessageType) -> Union[NotifResGuests, RetrieveResGuests]:
"""
Create a complete ResGuests structure with a primary customer.
Args:
customer_data: The customer data
message_type: Whether to create for NOTIF or RETRIEVE message types
Returns:
The appropriate ResGuests object
"""
if message_type == OtaMessageType.NOTIF:
return ResGuestFactory.create_notif_res_guests(customer_data)
else:
return ResGuestFactory.create_retrieve_res_guests(customer_data)
@staticmethod
def extract_data(obj: Any) -> Union[CustomerData, HotelReservationIdData, CommentsData]:
"""
Extract data from an AlpineBits object back to a simple data class.
Args:
obj: The AlpineBits object to extract data from
Returns:
The appropriate data object
"""
# Check if it's a Customer object
if hasattr(obj, 'person_name') and hasattr(obj.person_name, 'given_name'):
if isinstance(obj, NotifCustomer):
return CustomerFactory.from_notif_customer(obj)
elif isinstance(obj, RetrieveCustomer):
return CustomerFactory.from_retrieve_customer(obj)
# Check if it's a HotelReservationId object
elif hasattr(obj, 'res_id_type'):
if isinstance(obj, NotifHotelReservationId):
return HotelReservationIdFactory.from_notif_hotel_reservation_id(obj)
elif isinstance(obj, RetrieveHotelReservationId):
return HotelReservationIdFactory.from_retrieve_hotel_reservation_id(obj)
# Check if it's a Comments object
elif hasattr(obj, 'comment'):
if isinstance(obj, NotifComments):
return CommentFactory.from_notif_comments(obj)
elif isinstance(obj, RetrieveComments):
return CommentFactory.from_retrieve_comments(obj)
# Check if it's a ResGuests object
elif hasattr(obj, 'res_guest'):
return ResGuestFactory.extract_primary_customer(obj)
else:
raise ValueError(f"Unsupported object type: {type(obj)}")
# Usage examples
if __name__ == "__main__":
# Create customer data using simple data class
customer_data = CustomerData(
given_name="John",
surname="Doe",
name_prefix="Mr.",
phone_numbers=[
("+1234567890", PhoneTechType.MOBILE), # Phone number with type
("+0987654321", None), # Phone number without type
],
email_address="john.doe@example.com",
email_newsletter=True,
address_line="123 Main Street",
city_name="Anytown",
postal_code="12345",
country_code="US",
address_catalog=False,
gender="Male",
birth_date="1980-01-01",
language="en",
)
# Create customer for OtaHotelResNotifRq
notif_customer = CustomerFactory.create_notif_customer(customer_data)
print(
"Created NotifCustomer:",
notif_customer.person_name.given_name,
notif_customer.person_name.surname,
)
# Create customer for OtaResRetrieveRs
retrieve_customer = CustomerFactory.create_retrieve_customer(customer_data)
print(
"Created RetrieveCustomer:",
retrieve_customer.person_name.given_name,
retrieve_customer.person_name.surname,
)
# Convert back to data class
converted_data = CustomerFactory.from_notif_customer(notif_customer)
print("Converted back to data:", converted_data.given_name, converted_data.surname)
# Verify they contain the same information
print("Original and converted data match:", customer_data == converted_data)
print("\n--- HotelReservationIdFactory Examples ---")
# Create hotel reservation ID data
reservation_id_data = HotelReservationIdData(
res_id_type="123",
res_id_value="RESERVATION-456",
res_id_source="HOTEL_SYSTEM",
res_id_source_context="BOOKING_ENGINE",
)
# Create HotelReservationId for both types
notif_res_id = HotelReservationIdFactory.create_notif_hotel_reservation_id(
reservation_id_data
)
retrieve_res_id = HotelReservationIdFactory.create_retrieve_hotel_reservation_id(
reservation_id_data
)
print(
"Created NotifHotelReservationId:",
notif_res_id.res_id_type,
notif_res_id.res_id_value,
)
print(
"Created RetrieveHotelReservationId:",
retrieve_res_id.res_id_type,
retrieve_res_id.res_id_value,
)
# Convert back to data class
converted_res_id_data = HotelReservationIdFactory.from_notif_hotel_reservation_id(
notif_res_id
)
print(
"Converted back to reservation ID data:",
converted_res_id_data.res_id_type,
converted_res_id_data.res_id_value,
)
# Verify they contain the same information
print(
"Original and converted reservation ID data match:",
reservation_id_data == converted_res_id_data,
)
print("\n--- ResGuestFactory Examples ---")
# Create complete ResGuests structure for OtaHotelResNotifRq - much simpler!
notif_res_guests = ResGuestFactory.create_notif_res_guests(customer_data)
print(
"Created NotifResGuests with customer:",
notif_res_guests.res_guest.profiles.profile_info.profile.customer.person_name.given_name,
)
# Create complete ResGuests structure for OtaResRetrieveRs - much simpler!
retrieve_res_guests = ResGuestFactory.create_retrieve_res_guests(customer_data)
print(
"Created RetrieveResGuests with customer:",
retrieve_res_guests.res_guest.profiles.profile_info.profile.customer.person_name.given_name,
)
# Extract primary customer data back from ResGuests structure
extracted_data = ResGuestFactory.extract_primary_customer(retrieve_res_guests)
print("Extracted customer data:", extracted_data.given_name, extracted_data.surname)
# Verify roundtrip conversion
print("Roundtrip conversion successful:", customer_data == extracted_data)
print("\n--- Unified AlpineBitsFactory Examples ---")
# Much simpler approach - single factory with enum parameter!
print("=== Customer Creation ===")
notif_customer = AlpineBitsFactory.create(customer_data, OtaMessageType.NOTIF)
retrieve_customer = AlpineBitsFactory.create(customer_data, OtaMessageType.RETRIEVE)
print("Created customers using unified factory")
print("=== HotelReservationId Creation ===")
reservation_id_data = HotelReservationIdData(
res_id_type="123",
res_id_value="RESERVATION-456",
res_id_source="HOTEL_SYSTEM"
)
notif_res_id = AlpineBitsFactory.create(reservation_id_data, OtaMessageType.NOTIF)
retrieve_res_id = AlpineBitsFactory.create(reservation_id_data, OtaMessageType.RETRIEVE)
print("Created reservation IDs using unified factory")
print("=== Comments Creation ===")
comments_data = CommentsData(comments=[
CommentData(
name=CommentName2.CUSTOMER_COMMENT,
text="This is a customer comment about the reservation",
list_items=[
CommentListItemData(
value="Special dietary requirements: vegetarian",
list_item="1",
language="en"
),
CommentListItemData(
value="Late arrival expected",
list_item="2",
language="en"
)
]
),
CommentData(
name=CommentName2.ADDITIONAL_INFO,
text="Additional information about the stay"
)
])
notif_comments = AlpineBitsFactory.create(comments_data, OtaMessageType.NOTIF)
retrieve_comments = AlpineBitsFactory.create(comments_data, OtaMessageType.RETRIEVE)
print("Created comments using unified factory")
print("=== ResGuests Creation ===")
notif_res_guests = AlpineBitsFactory.create_res_guests(customer_data, OtaMessageType.NOTIF)
retrieve_res_guests = AlpineBitsFactory.create_res_guests(customer_data, OtaMessageType.RETRIEVE)
print("Created ResGuests using unified factory")
print("=== Data Extraction ===")
# Extract data back using unified interface
extracted_customer_data = AlpineBitsFactory.extract_data(notif_customer)
extracted_res_id_data = AlpineBitsFactory.extract_data(notif_res_id)
extracted_comments_data = AlpineBitsFactory.extract_data(retrieve_comments)
extracted_from_res_guests = AlpineBitsFactory.extract_data(retrieve_res_guests)
print("Data extraction successful:")
print("- Customer roundtrip:", customer_data == extracted_customer_data)
print("- ReservationId roundtrip:", reservation_id_data == extracted_res_id_data)
print("- Comments roundtrip:", comments_data == extracted_comments_data)
print("- ResGuests roundtrip:", customer_data == extracted_from_res_guests)
print("\n--- Comparison with old approach ---")
print("Old way required multiple imports and knowing specific factory methods")
print("New way: single import, single factory, enum parameter to specify type!")

22
src/alpine_bits_python/util/handshake_util.py Normal file
View File

@@ -0,0 +1,22 @@
from generated.alpinebits import OtaPingRq, OtaPingRs
def main():
# test parsing a ping request sample
path = "../../AlpineBits-HotelData-2024-10/files/samples/Handshake/Handshake-OTA_PingRQ.xml"
with open(
path, "r", encoding="utf-8") as f:
xml = f.read()
# Parse the XML into the request object
request = OtaPingRq.from_xml(xml)
print("Parsed OTA_PingRQ:", request)
if __name__ == "__main__":
main()