Transport and Settings

class Transport:
    def __init__(
        self,
        cache=None,
        timeout: int = 300,
        operation_timeout: int = None,
        session=None
    ):
        """
        Create HTTP transport for SOAP communication.

        Parameters:
        - cache: Cache backend instance (InMemoryCache, SqliteCache, etc.)
        - timeout: Timeout for loading WSDL/XSD documents (seconds)
        - operation_timeout: Timeout for SOAP operations (seconds, None=no timeout)
        - session: Custom requests.Session instance
        """

    def get(self, address: str, params: dict, headers: dict):
        """
        Execute HTTP GET request.
        
        Parameters:
        - address: Target URL
        - params: Query parameters
        - headers: HTTP headers
        
        Returns:
        requests.Response object
        """

    def post(self, address: str, message: str, headers: dict):
        """
        Execute HTTP POST request.
        
        Parameters:
        - address: Target URL  
        - message: Request body content
        - headers: HTTP headers
        
        Returns:
        requests.Response object
        """

    def post_xml(self, address: str, envelope, headers: dict):
        """
        POST XML envelope to SOAP endpoint.
        
        Parameters:
        - address: SOAP endpoint URL
        - envelope: XML envelope (lxml.etree._Element)
        - headers: HTTP headers
        
        Returns:
        requests.Response object
        """

class AsyncTransport(Transport):
    """
    Asynchronous HTTP transport using httpx.
    
    Provides same interface as Transport but with async methods.
    Requires 'httpx' package to be installed.
    """
    
    async def get(self, address: str, params: dict, headers: dict):
        """Async HTTP GET request."""

    async def post(self, address: str, message: str, headers: dict):
        """Async HTTP POST request."""

    async def post_xml(self, address: str, envelope, headers: dict):
        """Async XML POST to SOAP endpoint."""

class Settings:
    def __init__(
        self,
        strict: bool = True,
        raw_response: bool = False,
        force_https: bool = True,
        extra_http_headers: list = None,
        xml_huge_tree: bool = False,
        forbid_dtd: bool = False,
        forbid_entities: bool = True,
        forbid_external: bool = True,
        xsd_ignore_sequence_order: bool = False
    ):
        """
        Create settings for XML parsing and transport behavior.

        Parameters:
        - strict: Enable strict XML parsing mode
        - raw_response: Return raw response without XML parsing
        - force_https: Force HTTPS for all connections when WSDL is HTTPS
        - extra_http_headers: Additional HTTP headers for all requests
        - xml_huge_tree: Enable support for very large XML documents
        - forbid_dtd: Forbid DTD processing for security
        - forbid_entities: Forbid XML entities for security  
        - forbid_external: Forbid external resource access for security
        - xsd_ignore_sequence_order: Ignore XSD sequence ordering (workaround)
        """

    def __call__(self, **options):
        """
        Context manager for temporary settings override.
        
        Parameters:
        - **options: Settings to temporarily override
        
        Returns:
        Context manager for scoped settings
        """

class Base:
    """Base cache backend interface."""
    
    def add(self, url: str, content: bytes):
        """Add content to cache for given URL."""
        
    def get(self, url: str):
        """Retrieve cached content for URL."""

class InMemoryCache(Base):
    """In-memory cache backend."""
    
    def __init__(self, timeout: int = 3600):
        """
        Parameters:
        - timeout: Cache expiration time in seconds
        """

class SqliteCache(Base):
    """SQLite-based persistent cache backend."""
    
    def __init__(self, path: str = None, timeout: int = 3600):
        """
        Parameters:
        - path: SQLite database file path
        - timeout: Cache expiration time in seconds  
        """

from zeep import Client, Transport
from zeep.cache import InMemoryCache
import requests

# Custom session with authentication
session = requests.Session()
session.auth = ('username', 'password')
session.headers.update({'User-Agent': 'MyApp/1.0'})

# Transport with caching and custom session
transport = Transport(
    cache=InMemoryCache(timeout=3600),
    timeout=30,
    operation_timeout=120,
    session=session
)

client = Client('http://example.com/service.wsdl', transport=transport)

from zeep import Client, Settings

# Security-focused settings
settings = Settings(
    strict=True,
    forbid_dtd=True,
    forbid_entities=True,
    forbid_external=True,
    force_https=True,
    extra_http_headers=[
        ('X-API-Key', 'your-api-key'),
        ('Accept-Encoding', 'gzip, deflate')
    ]
)

client = Client('https://secure-service.com/service.wsdl', settings=settings)

from zeep import Client, Settings

client = Client('http://example.com/service.wsdl')

# Temporarily disable strict parsing for problematic service
with client.settings(strict=False, raw_response=True):
    raw_result = client.service.ProblematicOperation(param='value')
    # Process raw XML response manually

# Settings return to normal after context
normal_result = client.service.NormalOperation(param='value')

from zeep import Client, Transport
from zeep.cache import SqliteCache
import os

# Use SQLite cache in user cache directory
cache_dir = os.path.expanduser('~/.cache/zeep')
os.makedirs(cache_dir, exist_ok=True)

transport = Transport(
    cache=SqliteCache(
        path=os.path.join(cache_dir, 'soap_cache.db'),
        timeout=86400  # 24 hours
    )
)

client = Client('http://example.com/service.wsdl', transport=transport)

import asyncio
from zeep import AsyncClient
from zeep.transports import AsyncTransport
from zeep.cache import InMemoryCache

async def async_soap_call():
    transport = AsyncTransport(
        cache=InMemoryCache(),
        timeout=30,
        operation_timeout=60
    )
    
    async with AsyncClient('http://example.com/service.wsdl', transport=transport) as client:
        result = await client.service.AsyncOperation(param='value')
        return result

result = asyncio.run(async_soap_call())