diff --git a/README.md b/README.md
index b1e6bd1..7f2ac42 100644
--- a/README.md
+++ b/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="""
+
+ ...
+
+ """,
+)
+```
diff --git a/brazil_fiscal_client/fiscal_client.py b/brazil_fiscal_client/fiscal_client.py
index 1f716d9..2091f01 100644
--- a/brazil_fiscal_client/fiscal_client.py
+++ b/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,6 +297,8 @@ 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}"
@@ -266,6 +306,42 @@ def send(
_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)
diff --git a/pyproject.toml b/pyproject.toml
index fed950a..7680e7c 100644
--- a/pyproject.toml
+++ b/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
diff --git a/tests/test_fiscal_client.py b/tests/test_fiscal_client.py
index cfa201e..6bba8ed 100644
--- a/tests/test_fiscal_client.py
+++ b/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 = """
@@ -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 = ""
+ result = client.send(
+ action_class=None,
+ location="https://nfe-homologacao.svrs.rs.gov.br/ws/NfeStatusServico/NfeStatusServico4.asmx",
+ wrapped_obj=raw_payload,
+ )
+
+ self.assertIn("