Added handling for 'files' and 'file'.

Cyrus Radfar · Cyrus Radfar · commit 4644e782cdab · 2024-12-06T15:07:59.000-10:00
diff --git a/python_anvil/api_resources/payload.py b/python_anvil/api_resources/payload.py
@@ -197,7 +197,7 @@ class CreateEtchFilePayload(BaseModel):
     payloads: Union[str, Dict[str, FillPDFPayload]]
 
 
-class CreateEtchPacketPayload(BaseModel):
+class CreateEtchPacketPayload(FileCompatibleBaseModel):
     """
     Payload for createEtchPacket.
 
diff --git a/python_anvil/models.py b/python_anvil/models.py
@@ -1,12 +1,12 @@
 from typing import Any
-import warnings
-from pydantic import BaseModel, ConfigDict
-from io import IOBase, BytesIO, BufferedReader
+from io import BytesIO, BufferedReader
 
-try:    
+try:        
+    from pydantic import BaseModel
     from pydantic.version import VERSION as PYDANTIC_VERSION
     IS_V2 = not PYDANTIC_VERSION.startswith('1')
 except ImportError:    
+    from pydantic.v1 import BaseModel
     IS_V2 = False
 
 if IS_V2:
@@ -15,17 +15,41 @@ class FileCompatibleBaseModel(BaseModel):
             Patched model_dump to extract file objects from SerializationIterator in V2
             and return as BufferedReader
         """ 
+
+        def _iterator_to_buffered_reader(self, value):
+            content = bytearray()
+            try:
+                while True:
+                    content.extend(next(value))
+            except StopIteration:
+                # Create a BytesIO with the content
+                bio = BytesIO(bytes(content))
+                # Get the total length
+                bio.seek(0, 2)  # Seek to end
+                total_length = bio.tell()
+                bio.seek(0)  # Reset to start
+                
+                # Create a BufferedReader with the content
+                reader = BufferedReader(bio)
+                # Add a length attribute that requests_toolbelt can use
+                reader.len = total_length
+                return reader
+        
+        def _check_if_serialization_iterator(self, value):
+            return str(type(value).__name__) == 'SerializationIterator' and hasattr(value, '__next__')
+
         def model_dump(self, **kwargs):
-            
             data = super().model_dump(**kwargs)
             for key, value in data.items():
-                if str(type(value).__name__) == 'SerializationIterator' and hasattr(value, '__next__'):
-                    content = bytearray()
-                    try:
-                        while True:
-                            content.extend(next(value))
-                    except StopIteration:
-                        data[key] = BufferedReader(BytesIO(bytes(content)))
+                if key == 'file' and self._check_if_serialization_iterator(value):
+                    # Direct file case
+                    data[key] = self._iterator_to_buffered_reader(value)
+                elif key == 'files' and isinstance(value, list):
+                    # List of objects case
+                    for index, item in enumerate(value):
+                        if isinstance(item, dict) and 'file' in item:
+                            if self._check_if_serialization_iterator(item['file']):
+                                data[key][index]['file'] = self._iterator_to_buffered_reader(item['file'])
             return data
 
 else:
diff --git a/python_anvil/tests/test_models.py b/python_anvil/tests/test_models.py
@@ -120,3 +120,140 @@ def test_document_upload_handles_file_objects():
         assert data['title'] == "Test Document"
         assert len(data['fields']) == 1
         assert data['fields'][0]['id'] == "sig1"
+
+def test_create_etch_packet_payload_handles_nested_file_objects():
+    from python_anvil.api_resources.payload import (
+        CreateEtchPacketPayload,
+        DocumentUpload,
+        SignatureField,
+        EtchSigner,
+    )
+    
+    # Create a sample signature field
+    field = SignatureField(
+        id="sig1",
+        type="signature",
+        page_num=1,
+        rect={"x": 100.0, "y": 100.0, "width": 100.0}
+    )
+    
+    # Create a signer
+    signer = EtchSigner(
+        name="Test Signer",
+        email="test@example.com",
+        fields=[{"file_id": "doc1", "field_id": "sig1"}]
+    )
+    
+    # Test with a file object
+    with open(__file__, 'rb') as test_file:
+        # Create a DocumentUpload instance
+        doc = DocumentUpload(
+            id="doc1",
+            title="Test Document",
+            file=test_file,
+            fields=[field]
+        )
+        
+        # Create the packet payload
+        packet = CreateEtchPacketPayload(
+            name="Test Packet",
+            signers=[signer],
+            files=[doc],
+            is_test=True
+        )
+        
+        # Dump the model
+        data = packet.model_dump()
+        
+        # Verify the structure
+        assert data['name'] == "Test Packet"
+        assert len(data['files']) == 1
+        assert len(data['signers']) == 1
+        
+        # Verify file handling in the nested DocumentUpload
+        file_data = data['files'][0]
+        assert file_data['id'] == "doc1"
+        assert file_data['title'] == "Test Document"
+        assert isinstance(file_data['file'], BufferedReader), \
+            f"Expected BufferedReader but got {type(file_data['file'])}"
+        
+        # Verify file is still readable
+        file_data['file'].seek(0)
+        content = file_data['file'].read()
+        assert len(content) > 0, "File should be readable"
+        
+        # Verify the content matches the original file
+        with open(__file__, 'rb') as original_file:
+            assert content == original_file.read(), "File content should match original"            
+
+def test_create_etch_packet_payload_handles_multiple_files():
+    from python_anvil.api_resources.payload import (
+        CreateEtchPacketPayload,
+        DocumentUpload,
+        SignatureField,
+        EtchSigner,
+    )
+    
+    # Create signature fields
+    field1 = SignatureField(
+        id="sig1",
+        type="signature",
+        page_num=1,
+        rect={"x": 100.0, "y": 100.0, "width": 100.0}
+    )
+    
+    field2 = SignatureField(
+        id="sig2",
+        type="signature",
+        page_num=1,
+        rect={"x": 200.0, "y": 200.0, "width": 100.0}
+    )
+    
+    signer = EtchSigner(
+        name="Test Signer",
+        email="test@example.com",
+        fields=[
+            {"file_id": "doc1", "field_id": "sig1"},
+            {"file_id": "doc2", "field_id": "sig2"}
+        ]
+    )
+    
+    # Test with multiple file objects
+    with open(__file__, 'rb') as test_file1, open(__file__, 'rb') as test_file2:
+        doc1 = DocumentUpload(
+            id="doc1",
+            title="Test Document 1",
+            file=test_file1,
+            fields=[field1]
+        )
+        
+        doc2 = DocumentUpload(
+            id="doc2",
+            title="Test Document 2",
+            file=test_file2,
+            fields=[field2]
+        )
+        
+        packet = CreateEtchPacketPayload(
+            name="Test Packet",
+            signers=[signer],
+            files=[doc1, doc2],
+            is_test=True
+        )
+        
+        data = packet.model_dump()
+        
+        # Verify structure
+        assert len(data['files']) == 2
+        
+        # Verify both files are properly handled
+        for i, file_data in enumerate(data['files'], 1):
+            assert file_data['id'] == f"doc{i}"
+            assert file_data['title'] == f"Test Document {i}"
+            assert isinstance(file_data['file'], BufferedReader), \
+                f"File {i}: Expected BufferedReader but got {type(file_data['file'])}"
+            
+            # Verify file is readable
+            file_data['file'].seek(0)
+            content = file_data['file'].read()
+            assert len(content) > 0, f"File {i} should be readable"
