# Copyright (c) 2025 Ferrosoft GmbH. All rights reserved.
"""HTTP client for the Business Central connector microservice.
Calculated transport emissions are POSTed to the BC connector, which
relays them into a customer's Business Central tenant. The JSON
payload uses the BC field-name conventions (``camelCase``); units are
flattened to ``{"value": ..., "unit": ...}`` shapes by :func:`_decimal`.
"""
import logging
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import List
from urllib.parse import urljoin
import requests
from ferrosoft.apps.emiflow.models import DistanceType
from ferrosoft.apps.emiflow.services.emissions import (
TransportChainResult,
TransportChainElementResult,
)
from ferrosoft.apps.ferrobase.models import Tenant
from ferrosoft.apps.ferrobase.models.decimal import DecimalWithUnit
from ferrosoft.apps.ferrobase.tenant.context import get_current_tenant
def _decimal(input: DecimalWithUnit) -> dict:
"""Render a :class:`DecimalWithUnit` as the BC connector's ``{value, unit}`` shape."""
return {
"value": str(input.value),
"unit": str(input.try_unit().symbol),
}
[docs]
class Error(RuntimeError):
"""Base class for BC connector errors."""
pass
[docs]
class ClientError(Error):
"""Configuration or request-time failure in the BC connector client."""
pass
[docs]
class BcConnectorClient(ABC):
"""Abstract BC connector client. Production singleton is :class:`BcConnectorHTTPClient`."""
[docs]
@classmethod
def get_instance(cls) -> "BcConnectorClient":
"""Return the configured production client instance."""
return BcConnectorHTTPClient.get_instance()
[docs]
@abstractmethod
def post_emission(
self,
emission: TransportChainResult,
element_emissions: List[TransportChainElementResult],
tenant: Tenant | None = None,
):
"""Forward a chain's emission result and its per-element breakdown to BC."""
pass
[docs]
class BcConnectorHTTPClient(BcConnectorClient):
"""HTTP implementation of :class:`BcConnectorClient`.
Uses a long-lived JWT (loaded from Django settings) as the bearer
token; the connector identifies tenants by the Ferrosoft tenant
UUID embedded in the payload.
"""
[docs]
@dataclass
class Config:
"""Endpoint URL and bearer token for the connector."""
base_url: str
jwt: str | None
[docs]
@classmethod
def from_django_settings(cls):
"""Build a config from ``BC_CONNECTOR_BASE_URL`` / ``BC_CONNECTOR_JWT`` settings."""
from django.conf import settings
return cls(
getattr(settings, "BC_CONNECTOR_BASE_URL"),
getattr(settings, "BC_CONNECTOR_JWT"),
)
[docs]
@classmethod
def get_instance(cls):
"""Instantiate the client with config loaded from Django settings."""
return cls(cls.Config.from_django_settings())
def __init__(self, config: Config):
self.config = config
[docs]
def post_emission(
self,
emission: TransportChainResult,
element_emissions: List[TransportChainElementResult],
tenant: Tenant | None = None,
):
"""POST one chain's emissions (plus per-element breakdown) to the connector.
Raises:
ClientError: When the JWT is not configured.
RuntimeError: When the connector returns a non-200 status.
"""
if self.config.jwt is None:
raise ClientError("JWT is not configured")
payload = self._convert_tc_result(emission)
payload["elements"] = list(map(self._convert_tce_result, element_emissions))
response = requests.post(
urljoin(self.config.base_url, "emission"),
json=payload,
headers={
"Authorization": "Bearer " + self.config.jwt,
},
)
if response.status_code != 200:
logging.error("BC connector error response: %s", response.text)
raise RuntimeError(
"Posting to BC connector failed with code %d" % response.status_code
)
@staticmethod
def _convert_tc_result(
emission: TransportChainResult,
tenant: Tenant | None = None,
) -> dict:
"""Build the connector payload for one chain.
Refuses chains with mixed distance types since the connector
currently expects a single bucket per payload, and requires
either an explicit ``tenant`` argument or
:func:`get_current_tenant` to resolve one.
"""
tenant = tenant or get_current_tenant()
if tenant is None:
raise RuntimeError(
"Tenant must either be passed explicitly or defined implicitly through TenantMiddleware"
)
activities = emission.transport_activity
if len(activities) != 1:
raise RuntimeError("Mixed transport activities are not supported")
distance_type = DistanceType(list(activities.keys())[0])
transport_activity = activities[distance_type]
return {
"tenant": str(tenant.id),
"transportChain": {
"id": str(emission.chain.id),
"reference": emission.chain.reference,
},
"distanceType": distance_type.name,
"transportActivity": _decimal(transport_activity),
"emissionIntensity": {
"total": _decimal(emission.emission_intensity_total),
"operation": _decimal(emission.emission_intensity_operation),
},
"emission": {
"total": _decimal(emission.emission_total),
"operation": _decimal(emission.emission_operation),
},
}
@staticmethod
def _convert_tce_result(result: TransportChainElementResult) -> dict:
"""Build the connector payload for one chain element."""
operation = result.element.operation
return {
"transportChainElement": {
"id": str(result.element.id),
},
"emitter": str(operation.emitter),
"energyCarrier": str(operation.energy_carrier),
"distanceType": result.distance_type.name,
"transportActivity": _decimal(result.transport_activity),
"emission": {
"total": _decimal(result.emission_total),
"operation": _decimal(result.emission_vehicle_operation),
},
}