Files
alpinebits_python/src/alpine_bits_python/simplified_access.py
2025-09-25 09:47:38 +02:00

741 lines
28 KiB
Python

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!")