Make xsdata optional and add requests-only fallback in FiscalClient

README.md

@@ -10,7 +10,7 @@ This client is designed to be inherited by specialized clients such as for elect
 invoicing (NFe). But with some extra boiler plate code to deal with the SOAP enveloppe,
 it can still be used alone as you can see in the usage section below.
 
-It uses [xsdata](https://github.com/tefra/xsdata) for the
+It uses [xsdata](https://github.com/tefra/xsdata) as a soft dependency for the
 [databinding](https://xsdata.readthedocs.io/en/latest/data_binding/basics/) and it
 overrides its SOAP
 [client](https://xsdata.readthedocs.io/en/latest/codegen/wsdl_modeling/#client)
@@ -19,6 +19,11 @@ 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 +82,31 @@ 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,42 @@
 from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
 from enum import Enum
-from typing import Any
+from importlib import import_module
+from importlib.util import find_spec
+from typing import TYPE_CHECKING, 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 TYPE_CHECKING:
+    # Import types for type checking only
+    from xsdata.exceptions import ParserError
+    from xsdata.formats.dataclass.client import Client, ClientValueError, Config
+    from xsdata.formats.dataclass.parsers import DictDecoder
+elif XSDATA_AVAILABLE:
+    _xsdata_exceptions = import_module("xsdata.exceptions")
+    ParserError = _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):  # type: ignore[no-redef]
+        """Raised when parsing a SOAP response fails."""
+
+    class ClientValueError(ValueError):  # type: ignore[no-redef]
+        """Raised when a payload does not match client expectations."""
+
+    class Client:  # type: ignore[no-redef]
+        """Fallback base class used when xsdata isn't installed."""
+
 
 _logger = logging.Logger(__name__)
 
@@ -136,14 +164,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 +226,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 +252,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 +279,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 +297,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 +367,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": {}})