Make xsdata an optional dependency with requests fallback

Author: rvalyi

README.md

@@ -19,6 +19,10 @@ overrides its SOAP
 
 `pip install brazil-fiscal-client`
 
+If you want xsdata databinding/parsing support (recommended for nfelib/Odoo and advanced SOAP use cases), install the optional extra:
+
+`pip install brazil-fiscal-client[xsdata]`
+
 ## Usage
 
 For instance, with an appropriate pkcs12 certificate, you can query the NFe server
@@ -77,3 +81,30 @@ downloaded wsdl file and using the
 [WSDL xsdata generator](https://xsdata.readthedocs.io/en/latest/codegen/wsdl_modeling/).
 All this is usually done in the specialized clients that override this base
 `brazil-fiscal-client`SOAP client.
+
+
+## Usage without xsdata (requests-only mode)
+
+If `xsdata` is not installed, `FiscalClient` still works for basic SOAP calls using plain XML payloads (string/bytes). In this mode, `send()` returns the raw SOAP XML response string.
+
+```python
+from brazil_fiscal_client.fiscal_client import FiscalClient
+
+client = FiscalClient(
+    ambiente="2",
+    versao="4.00",
+    pkcs12_data=pkcs12_data,
+    pkcs12_password=certificate_password,
+    fake_certificate=True,
+)
+
+raw_response_xml = client.send(
+    action_class=None,  # not required in requests-only mode
+    location="https://example.com/service",
+    wrapped_obj="""
+        <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
+          <soapenv:Body>...</soapenv:Body>
+        </soapenv:Envelope>
+    """,
+)
+```

brazil_fiscal_client/fiscal_client.py

@@ -8,14 +8,37 @@
 from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
 from enum import Enum
+from importlib import import_module
+from importlib.util import find_spec
 from typing import Any
 
+import requests
 from requests.adapters import HTTPAdapter, Retry
 from requests.exceptions import RequestException
 from requests_pkcs12 import Pkcs12Adapter
-from xsdata.exceptions import ParserError
-from xsdata.formats.dataclass.client import Client, ClientValueError, Config
-from xsdata.formats.dataclass.parsers import DictDecoder
+
+XSDATA_AVAILABLE = find_spec("xsdata") is not None
+
+if XSDATA_AVAILABLE:
+    ParserError = import_module("xsdata.exceptions").ParserError
+    _client_mod = import_module("xsdata.formats.dataclass.client")
+    _parser_mod = import_module("xsdata.formats.dataclass.parsers")
+    Client = _client_mod.Client
+    ClientValueError = _client_mod.ClientValueError
+    Config = _client_mod.Config
+    DictDecoder = _parser_mod.DictDecoder
+else:
+
+    class ParserError(Exception):
+        """Raised when parsing a SOAP response fails."""
+
+    class ClientValueError(ValueError):
+        """Raised when a payload does not match client expectations."""
+
+    class _ClientBase:
+        """Fallback base class used when xsdata isn't installed."""
+
+    Client = _ClientBase
 
 _logger = logging.Logger(__name__)
 
@@ -136,14 +159,22 @@ def __init__(
         elif uf:
             self.uf = uf.value
 
-        super().__init__(config=kwargs.get("config", {}), **kwargs)
+        self._xsdata_available = XSDATA_AVAILABLE
+        if self._xsdata_available:
+            super().__init__(config=kwargs.get("config", {}), **kwargs)
+        else:
+            self.session = requests.Session()
         self.versao = versao
         self.pkcs12_data = pkcs12_data
         self.pkcs12_password = pkcs12_password
         self.verify_ssl = verify_ssl
         self.service = service
-        self.transport.timeout = timeout
-        self.transport.session.verify = self.verify_ssl
+        if self._xsdata_available:
+            self.transport.timeout = timeout
+            self.transport.session.verify = self.verify_ssl
+        else:
+            self.timeout = timeout
+            self.session.verify = self.verify_ssl
         self.fake_certificate = fake_certificate
         self.soap12_envelope = soap12_envelope
         self.wrap_response = wrap_response
@@ -190,16 +221,17 @@ def send(
             The response model instance.
         """
         server = "https://" + location.split("/")[2]
-        self.config = Config.from_service(action_class, location=location)
+        if self._xsdata_available:
+            self.config = Config.from_service(action_class, location=location)
 
         retries = Retry(  # retry in case of errors
             total=RETRIES,
             backoff_factor=BACKOFF_FACTOR,
             status_forcelist=RETRY_ERRORS,
         )
-        self.transport.session.mount(server, HTTPAdapter(max_retries=retries))
+        self._session.mount(server, HTTPAdapter(max_retries=retries))
         if not self.fake_certificate:
-            self.transport.session.mount(
+            self._session.mount(
                 server,
                 Pkcs12Adapter(
                     pkcs12_data=self.pkcs12_data,
@@ -215,9 +247,7 @@ def send(
         try:
             _logger.debug(f"Sending SOAP request to {location} with headers: {headers}")
             _logger.debug(f"SOAP request payload: {data}")
-            original_response = self.transport.post(
-                location, data=data, headers=headers
-            )
+            original_response = self._post(location, data=data, headers=headers)
             response = original_response.decode()
             _logger.debug(f"SOAP response: {response}")
 
@@ -244,12 +274,15 @@ def send(
                 )
                 response = response.replace(SOAP12_ENV_NS, SOAP11_ENV_NS)
 
-            res = self.parser.from_string(response, action_class.output)
+            if self._xsdata_available:
+                res = self.parser.from_string(response, action_class.output)
+            else:
+                res = response
             if not self.wrap_response:
                 return res
 
             return WrappedResponse(
-                webservice=action_class.soapAction.split("/")[-1],
+                webservice=self._webservice_name(action_class),
                 envio_raiz=placeholder_content,
                 envio_xml=data.encode(),
                 resposta=res,
@@ -259,13 +292,51 @@ def send(
             _logger.error(f"Failed to send SOAP request to {location}: {e}")
             raise
         except ParserError as e:
+            if not self._xsdata_available:
+                raise
             _logger.error(
                 f"Failed to parse SOAP response as {action_class.output}\n"
                 f"SOAP response:\n{response}"
             )
             _logger.error(f"Error: {e}")
             raise
 
+    def prepare_headers(self, headers: dict) -> dict:
+        """Prepare request headers.
+
+        Keep xsdata behavior when available and default to plain HTTP headers
+        when running in lightweight mode without xsdata.
+        """
+        if self._xsdata_available:
+            return super().prepare_headers(headers)
+
+        return {
+            "Content-Type": "text/xml; charset=utf-8",
+            **headers,
+        }
+
+    @property
+    def _session(self):
+        return self.transport.session if self._xsdata_available else self.session
+
+    def _post(self, location: str, data: str, headers: dict) -> bytes:
+        if self._xsdata_available:
+            return self.transport.post(location, data=data, headers=headers)
+
+        response = self.session.post(
+            location,
+            data=data,
+            headers=headers,
+            timeout=self.timeout,
+        )
+        response.raise_for_status()
+        return response.content
+
+    @staticmethod
+    def _webservice_name(action_class: Any) -> str:
+        soap_action = getattr(action_class, "soapAction", "")
+        return soap_action.split("/")[-1] if soap_action else "unknown"
+
     def prepare_payload(
         self,
         obj: Any,
@@ -291,6 +362,18 @@ def prepare_payload(
         Raises:
             ClientValueError: If the config input type doesn't match the given object.
         """
+        if not self._xsdata_available:
+            if isinstance(obj, bytes):
+                return obj.decode()
+            if isinstance(obj, str):
+                return obj
+            if isinstance(obj, dict) and "raw_xml" in obj:
+                return str(obj["raw_xml"])
+            raise ClientValueError(
+                "xsdata is not installed: pass the SOAP payload as `str`, `bytes` "
+                "or a dict containing a `raw_xml` key."
+            )
+
         if isinstance(obj, dict):
             decoder = DictDecoder(context=self.serializer.context)
             obj = decoder.decode(obj, self.config.input)

pyproject.toml

@@ -29,7 +29,6 @@ classifiers = [
 keywords = ["soap", "wsdl", "nfe", "fazenda", "binding", "nfelib", "odoo"]
 requires-python = ">=3.8"
 dependencies = [
-    "xsdata",
     "requests",
     "requests-pkcs12"
 ]
@@ -45,6 +44,9 @@ Source = "https://github.com/akretion/brazil-fiscal-client"
 Documentation = "https://github.com/akretion/brazil-fiscal-client"
 
 [project.optional-dependencies]
+xsdata = [
+    "xsdata",
+]
 test = [
     "pre-commit",
     "pytest",
@@ -53,8 +55,6 @@ test = [
     "lxml",
 ]
 
-[project.scripts]
-xsdata = "xsdata.__main__:main"
 
 [tool.setuptools]
 include-package-data = true

tests/test_fiscal_client.py

@@ -10,7 +10,12 @@
 from xsdata.exceptions import ParserError
 from xsdata.formats.dataclass.transports import DefaultTransport
 
-from brazil_fiscal_client.fiscal_client import FiscalClient, Tamb, TcodUfIbge
+from brazil_fiscal_client.fiscal_client import (
+    ClientValueError,
+    FiscalClient,
+    Tamb,
+    TcodUfIbge,
+)
 from tests.fixtures.nfestatusservico4 import NfeStatusServico4SoapNfeStatusServicoNf
 
 response = """<?xml version="1.0" encoding="utf-8"?>
@@ -341,3 +346,46 @@ def test_send_raise_on_soap_mismatch(self, mock_post):
         self.assertIn(
             SOAP12_ENV_NS, str(cm.exception)
         )  # Check if error mentions the unexpected NS
+
+    @mock.patch("brazil_fiscal_client.fiscal_client.XSDATA_AVAILABLE", False)
+    @mock.patch("requests.Session.post")
+    def test_send_without_xsdata_returns_raw_xml(self, mock_post):
+        mock_post.return_value = mock.Mock(
+            content=response.encode(),
+            raise_for_status=mock.Mock(),
+        )
+
+        client = FiscalClient(
+            ambiente=Tamb.DEV,
+            uf=TcodUfIbge.SC,
+            versao="4.00",
+            pkcs12_data=b"fake_cert",
+            pkcs12_password="123456",
+            fake_certificate=True,
+            service="nfe",
+        )
+
+        raw_payload = "<soapenv:Envelope xmlns:soapenv='http://schemas.xmlsoap.org/soap/envelope/'></soapenv:Envelope>"
+        result = client.send(
+            action_class=None,
+            location="https://nfe-homologacao.svrs.rs.gov.br/ws/NfeStatusServico/NfeStatusServico4.asmx",
+            wrapped_obj=raw_payload,
+        )
+
+        self.assertIn("<soap:Envelope", result)
+        self.assertTrue(mock_post.called)
+
+    @mock.patch("brazil_fiscal_client.fiscal_client.XSDATA_AVAILABLE", False)
+    def test_prepare_payload_without_xsdata_requires_raw_xml(self):
+        client = FiscalClient(
+            ambiente=Tamb.DEV,
+            uf=TcodUfIbge.SC,
+            versao="4.00",
+            pkcs12_data=b"fake_cert",
+            pkcs12_password="123456",
+            fake_certificate=True,
+            service="nfe",
+        )
+
+        with self.assertRaisesRegex(ClientValueError, "xsdata is not installed"):
+            client.prepare_payload({"Body": {}})