AlpineBitsfactory. Not necessarily a great idea though

This commit is contained in:
Jonas Linter
2025-09-24 16:11:43 +02:00
parent 79ceb72d12
commit 5e4703a6b5
4 changed files with 268 additions and 8 deletions

View File

@@ -8,11 +8,10 @@ from xsdata_pydantic.bindings import XmlSerializer
from simplified_access import (
CustomerData,
CustomerFactory,
HotelReservationIdData,
HotelReservationIdFactory,
PhoneTechType,
ResGuestFactory,
AlpineBitsFactory,
OtaMessageType
)
@@ -65,7 +64,9 @@ def main():
language="en",
)
res_guests = ResGuestFactory.create_retrieve_res_guests(customer_data)
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",
@@ -74,9 +75,7 @@ def main():
res_id_source_context="99tales",
)
# Create HotelReservationId using the factory
hotel_res_id = HotelReservationIdFactory.create_retrieve_hotel_reservation_id(
hotel_res_id_data
)
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(

View File

@@ -1,7 +1,7 @@
<?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-24T13:22:30.892827+00:00" ResStatus="Requested" RoomStayReservation="true">
<HotelReservation CreateDateTime="2025-09-24T13:36:42.168402+00:00" ResStatus="Requested" RoomStayReservation="true">
<UniqueID Type="14" ID="6b34fe24ac2ff811"/>
<RoomStays>
<RoomStay>

View File

@@ -22,6 +22,12 @@ class PhoneTechType(Enum):
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."""
@@ -346,6 +352,86 @@ class ResGuestFactory:
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], message_type: OtaMessageType) -> Any:
"""
Create an AlpineBits object based on the data type and message type.
Args:
data: The data object (CustomerData, HotelReservationIdData, 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)
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]:
"""
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 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
@@ -459,3 +545,41 @@ if __name__ == "__main__":
# 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("=== 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_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("- 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!")

View File

@@ -12,7 +12,9 @@ from simplified_access import (
ResGuestFactory,
HotelReservationIdData,
HotelReservationIdFactory,
AlpineBitsFactory,
PhoneTechType,
OtaMessageType,
NotifCustomer,
RetrieveCustomer,
NotifResGuests,
@@ -381,6 +383,141 @@ class TestPhoneTechType:
assert PhoneTechType.MOBILE.value == "5"
class TestAlpineBitsFactory:
"""Test the unified AlpineBitsFactory class."""
def test_create_customer_notif(self, sample_customer_data):
"""Test creating customer using unified factory for NOTIF."""
customer = AlpineBitsFactory.create(sample_customer_data, OtaMessageType.NOTIF)
assert isinstance(customer, NotifCustomer)
assert customer.person_name.given_name == "John"
assert customer.person_name.surname == "Doe"
def test_create_customer_retrieve(self, sample_customer_data):
"""Test creating customer using unified factory for RETRIEVE."""
customer = AlpineBitsFactory.create(sample_customer_data, OtaMessageType.RETRIEVE)
assert isinstance(customer, RetrieveCustomer)
assert customer.person_name.given_name == "John"
assert customer.person_name.surname == "Doe"
def test_create_hotel_reservation_id_notif(self, sample_hotel_reservation_id_data):
"""Test creating hotel reservation ID using unified factory for NOTIF."""
reservation_id = AlpineBitsFactory.create(sample_hotel_reservation_id_data, OtaMessageType.NOTIF)
assert isinstance(reservation_id, NotifHotelReservationId)
assert reservation_id.res_id_type == "123"
assert reservation_id.res_id_value == "RESERVATION-456"
def test_create_hotel_reservation_id_retrieve(self, sample_hotel_reservation_id_data):
"""Test creating hotel reservation ID using unified factory for RETRIEVE."""
reservation_id = AlpineBitsFactory.create(sample_hotel_reservation_id_data, OtaMessageType.RETRIEVE)
assert isinstance(reservation_id, RetrieveHotelReservationId)
assert reservation_id.res_id_type == "123"
assert reservation_id.res_id_value == "RESERVATION-456"
def test_create_res_guests_notif(self, sample_customer_data):
"""Test creating ResGuests using unified factory for NOTIF."""
res_guests = AlpineBitsFactory.create_res_guests(sample_customer_data, OtaMessageType.NOTIF)
assert isinstance(res_guests, NotifResGuests)
customer = res_guests.res_guest.profiles.profile_info.profile.customer
assert customer.person_name.given_name == "John"
def test_create_res_guests_retrieve(self, sample_customer_data):
"""Test creating ResGuests using unified factory for RETRIEVE."""
res_guests = AlpineBitsFactory.create_res_guests(sample_customer_data, OtaMessageType.RETRIEVE)
assert isinstance(res_guests, RetrieveResGuests)
customer = res_guests.res_guest.profiles.profile_info.profile.customer
assert customer.person_name.given_name == "John"
def test_extract_data_from_customer(self, sample_customer_data):
"""Test extracting data from customer objects."""
# Create both types and extract data back
notif_customer = AlpineBitsFactory.create(sample_customer_data, OtaMessageType.NOTIF)
retrieve_customer = AlpineBitsFactory.create(sample_customer_data, OtaMessageType.RETRIEVE)
notif_extracted = AlpineBitsFactory.extract_data(notif_customer)
retrieve_extracted = AlpineBitsFactory.extract_data(retrieve_customer)
assert notif_extracted == sample_customer_data
assert retrieve_extracted == sample_customer_data
def test_extract_data_from_hotel_reservation_id(self, sample_hotel_reservation_id_data):
"""Test extracting data from hotel reservation ID objects."""
# Create both types and extract data back
notif_res_id = AlpineBitsFactory.create(sample_hotel_reservation_id_data, OtaMessageType.NOTIF)
retrieve_res_id = AlpineBitsFactory.create(sample_hotel_reservation_id_data, OtaMessageType.RETRIEVE)
notif_extracted = AlpineBitsFactory.extract_data(notif_res_id)
retrieve_extracted = AlpineBitsFactory.extract_data(retrieve_res_id)
assert notif_extracted == sample_hotel_reservation_id_data
assert retrieve_extracted == sample_hotel_reservation_id_data
def test_extract_data_from_res_guests(self, sample_customer_data):
"""Test extracting data from ResGuests objects."""
# Create both types and extract data back
notif_res_guests = AlpineBitsFactory.create_res_guests(sample_customer_data, OtaMessageType.NOTIF)
retrieve_res_guests = AlpineBitsFactory.create_res_guests(sample_customer_data, OtaMessageType.RETRIEVE)
notif_extracted = AlpineBitsFactory.extract_data(notif_res_guests)
retrieve_extracted = AlpineBitsFactory.extract_data(retrieve_res_guests)
assert notif_extracted == sample_customer_data
assert retrieve_extracted == sample_customer_data
def test_unsupported_data_type_error(self):
"""Test that unsupported data types raise ValueError."""
with pytest.raises(ValueError, match="Unsupported data type"):
AlpineBitsFactory.create("invalid_data", OtaMessageType.NOTIF)
def test_unsupported_object_type_error(self):
"""Test that unsupported object types raise ValueError in extract_data."""
with pytest.raises(ValueError, match="Unsupported object type"):
AlpineBitsFactory.extract_data("invalid_object")
def test_complete_workflow_with_unified_factory(self):
"""Test a complete workflow using only the unified factory."""
# Original data
customer_data = CustomerData(
given_name="Unified",
surname="Factory",
email_address="unified@factory.com",
phone_numbers=[("+1234567890", PhoneTechType.MOBILE)]
)
reservation_data = HotelReservationIdData(
res_id_type="999",
res_id_value="UNIFIED-TEST"
)
# Create using unified factory
customer_notif = AlpineBitsFactory.create(customer_data, OtaMessageType.NOTIF)
customer_retrieve = AlpineBitsFactory.create(customer_data, OtaMessageType.RETRIEVE)
res_id_notif = AlpineBitsFactory.create(reservation_data, OtaMessageType.NOTIF)
res_id_retrieve = AlpineBitsFactory.create(reservation_data, OtaMessageType.RETRIEVE)
res_guests_notif = AlpineBitsFactory.create_res_guests(customer_data, OtaMessageType.NOTIF)
res_guests_retrieve = AlpineBitsFactory.create_res_guests(customer_data, OtaMessageType.RETRIEVE)
# Extract everything back
extracted_customer_from_notif = AlpineBitsFactory.extract_data(customer_notif)
extracted_customer_from_retrieve = AlpineBitsFactory.extract_data(customer_retrieve)
extracted_res_id_from_notif = AlpineBitsFactory.extract_data(res_id_notif)
extracted_res_id_from_retrieve = AlpineBitsFactory.extract_data(res_id_retrieve)
extracted_from_res_guests_notif = AlpineBitsFactory.extract_data(res_guests_notif)
extracted_from_res_guests_retrieve = AlpineBitsFactory.extract_data(res_guests_retrieve)
# Verify everything matches
assert extracted_customer_from_notif == customer_data
assert extracted_customer_from_retrieve == customer_data
assert extracted_res_id_from_notif == reservation_data
assert extracted_res_id_from_retrieve == reservation_data
assert extracted_from_res_guests_notif == customer_data
assert extracted_from_res_guests_retrieve == customer_data
class TestIntegration:
"""Integration tests combining both factories."""