food

src/alpine_bits_python/alpinebits_server.py

@@ -18,6 +18,7 @@ from .generated.alpinebits import OtaPingRq, OtaPingRs, WarningStatus
 from xsdata_pydantic.bindings import XmlSerializer
 from xsdata.formats.dataclass.serializers.config import SerializerConfig
 from abc import ABC, abstractmethod
+from xsdata_pydantic.bindings import XmlParser
 
 
 class HttpStatusCode(IntEnum):
@@ -143,6 +144,7 @@ class ServerCapabilities:
     def __init__(self):
         self.action_registry: Dict[str, Type[AlpineBitsAction]] = {}
         self._discover_actions()
+        self.capability_dict = None
     
     def _discover_actions(self):
         """Discover all AlpineBitsAction implementations in the current module."""
@@ -170,12 +172,11 @@ class ServerCapabilities:
             return True
         return False
     
-    def get_capabilities_dict(self) -> Dict:
+
+    def create_capabilities_dict(self) -> None:
         """
         Generate the capabilities dictionary based on discovered actions.
         
-        Returns:
-            Dictionary matching the AlpineBits capabilities format
         """
         versions_dict = {}
         
@@ -206,9 +207,21 @@ class ServerCapabilities:
                 
                 versions_dict[version_str]["actions"].append(action_dict)
 
-        return {
+        self.capability_dict = {"versions": list(versions_dict.values())}
-            "versions": list(versions_dict.values())
+
-        }
+
+        
+        return None
+
+    
+    def get_capabilities_dict(self) -> Dict:
+        """
+        Get capabilities as a dictionary. Generates if not already created.
+        """
+        
+        if self.capability_dict is None:
+            self.create_capabilities_dict()
+        return self.capability_dict
     
     def get_capabilities_json(self) -> str:
         """Get capabilities as formatted JSON string."""
@@ -228,13 +241,34 @@ class PingAction(AlpineBitsAction):
         self.name = AlpineBitsActionName.OTA_PING
         self.version = [Version.V2024_10, Version.V2022_10]  # Supports multiple versions
     
-    async def handle(self, action: str, request_xml: str, version: Version) -> AlpineBitsResponse:
+    async def handle(self, action: str, request_xml: str, version: Version, server_capabilities: None | ServerCapabilities = None) -> AlpineBitsResponse:
         """Handle ping requests."""
-        response_xml = f'''<?xml version="1.0" encoding="UTF-8"?>
+
-<OTA_PingRS xmlns="http://www.opentravel.org/OTA/2003/05" Version="8.000">
+        if server_capabilities is None:
-    <Success/>
+            return AlpineBitsResponse("Error: Something went wrong", HttpStatusCode.INTERNAL_SERVER_ERROR)
-    <EchoData>Ping successful for version {version.value}</EchoData>
+        
-</OTA_PingRS>'''
+
+        # Parse the incoming request XML and extract EchoData
+        parser = XmlParser()
+
+        try:
+            parsed_request = parser.from_string(request_xml, OtaPingRq)
+            echo_data = json.loads(parsed_request.echo_data)
+        except Exception as e:
+            return AlpineBitsResponse(f"Error: Invalid XML request - {str(e)}", HttpStatusCode.BAD_REQUEST)
+
+        
+        
+        
+        
+
+
+        
+
+
+
+
+
         return AlpineBitsResponse(response_xml, HttpStatusCode.OK)
 
 
@@ -291,9 +325,6 @@ class GuestRequestsAction(AlpineBitsAction):
 
 
 
-
-
-
 class AlpineBitsServer:
     """
     Asynchronous AlpineBits server for handling hotel data exchange requests.
@@ -454,12 +485,12 @@ async def main():
     
     # Test different request formats
     test_cases = [
-        ("OTA_Ping", "2024-10"),
+        ("OTA_Ping:Handshaking", "2024-10"),
         ("OTA_Read:GuestRequests", "2024-10"),
-        ("OTA_Read", "2022-10"),
+        ("OTA_Read:GuestRequest", "2022-10"),
         ("OTA_HotelAvailNotif", "2024-10"),
         ("UnknownAction", "2024-10"),
-        ("OTA_Ping", "unsupported-version")
+        ("OTA_Ping:Handshaking", "unsupported-version")
     ]
     
     for request_name, version in test_cases: