feat: Add Get PDF of Invoice to API

iec_api/commons.py

@@ -8,7 +8,7 @@
 from typing import Any, Optional
 
 import pytz
-from aiohttp import ClientError, ClientResponse, ClientSession
+from aiohttp import ClientError, ClientResponse, ClientSession, StreamReader
 
 from iec_api.const import ERROR_FIELD_NAME, ERROR_SUMMARY_FIELD_NAME, TIMEZONE
 from iec_api.models.error_response import IecErrorResponse
@@ -186,6 +186,37 @@ async def send_post_request(
     return json_resp
 
 
+async def send_non_json_post_request(
+    session: ClientSession,
+    url: str,
+    timeout: Optional[int] = 60,
+    headers: Optional[dict] = None,
+    data: Optional[dict] = None,
+    json_data: Optional[dict] = None,
+) -> StreamReader:
+    try:
+        if not headers:
+            headers = session.headers
+
+        if not timeout:
+            headers = session.timeout
+
+        logger.debug(f"HTTP POST: {url}")
+        logger.debug(f"HTTP Content: {data or json_data}")
+
+        resp = await session.post(url=url, data=data, json=json_data, headers=headers, timeout=timeout)
+    except TimeoutError as ex:
+        raise IECError(-1, f"Failed to communicate with IEC API due to time out: ({str(ex)})")
+    except ClientError as ex:
+        raise IECError(-1, f"Failed to communicate with IEC API due to ClientError: ({str(ex)})")
+    except JSONDecodeError as ex:
+        raise IECError(-1, f"Received invalid response from IEC API: {str(ex)}")
+
+    if resp.status != http.HTTPStatus.OK:
+        raise IECError(resp.status, resp.reason)
+    return resp.content
+
+
 def convert_to_tz_aware_datetime(dt: Optional[datetime]) -> Optional[datetime]:
     """
     Convert a datetime object to a timezone aware datetime object.

iec_api/const.py

@@ -36,7 +36,8 @@
 GET_DEVICES_URL = IEC_API_BASE_URL + "Device/{contract_id}"
 GET_DEVICE_BY_DEVICE_ID_URL = GET_DEVICES_URL + "/{device_id}"
 GET_DEVICE_TYPE_URL = IEC_API_BASE_URL + "Device/type/{bp_number}/{contract_id}/false"
-GET_BILLING_INVOICES = IEC_API_BASE_URL + "billingCollection/invoices/{contract_id}/{bp_number}"
+GET_BILLING_INVOICES_URL = IEC_API_BASE_URL + "BillingCollection/invoices/{contract_id}/{bp_number}"
+GET_INVOICE_PDF_URL = IEC_API_BASE_URL + "BillingCollection/pdf"
 GET_KWH_TARIFF_URL = IEC_API_BASE_URL + "content/en-US/content/tariffs/contentpages/homeelectricitytariff"
 ERROR_FIELD_NAME = "Error"
 ERROR_SUMMARY_FIELD_NAME = "errorSummary"

iec_api/data.py

@@ -9,14 +9,15 @@
 from iec_api import commons
 from iec_api.const import (
     GET_ACCOUNTS_URL,
-    GET_BILLING_INVOICES,
+    GET_BILLING_INVOICES_URL,
     GET_CONSUMER_URL,
     GET_CONTRACTS_URL,
     GET_DEFAULT_CONTRACT_URL,
     GET_DEVICE_BY_DEVICE_ID_URL,
     GET_DEVICE_TYPE_URL,
     GET_DEVICES_URL,
     GET_ELECTRIC_BILL_URL,
+    GET_INVOICE_PDF_URL,
     GET_KWH_TARIFF_URL,
     GET_LAST_METER_READING_URL,
     GET_REQUEST_READING_URL,
@@ -34,6 +35,7 @@
 from iec_api.models.electric_bill import ElectricBill
 from iec_api.models.electric_bill import decoder as electric_bill_decoder
 from iec_api.models.exceptions import IECError
+from iec_api.models.get_pdf import GetPdfRequest
 from iec_api.models.invoice import GetInvoicesBody
 from iec_api.models.invoice import decoder as invoice_decoder
 from iec_api.models.jwt import JWT
@@ -187,10 +189,27 @@ async def get_billing_invoices(
 ) -> Optional[GetInvoicesBody]:
     """Get Device Type data response from IEC API."""
     return await _get_response_with_descriptor(
-        session, token, GET_BILLING_INVOICES.format(bp_number=bp_number, contract_id=contract_id), invoice_decoder
+        session, token, GET_BILLING_INVOICES_URL.format(bp_number=bp_number, contract_id=contract_id), invoice_decoder
     )
 
 
+async def get_invoice_pdf(
+    session: ClientSession, token: JWT, bp_number: int | str, contract_id: int | str, invoice_number: int | str
+) -> bytes:
+    """Get Device Type data response from IEC API."""
+    headers = commons.add_auth_bearer_to_headers(HEADERS_WITH_AUTH, token.id_token)
+    headers = headers.copy()  # don't modify original headers
+    headers.update({"accept": "application/pdf", "content-type": "application/json"})
+
+    request = GetPdfRequest(
+        invoice_number=str(invoice_number), contract_id=str(contract_id), bp_number=str(bp_number)
+    ).to_dict()
+    response = await commons.send_non_json_post_request(
+        session, url=GET_INVOICE_PDF_URL, headers=headers, json_data=request
+    )
+    return await response.read()
+
+
 async def get_kwh_tariff(session: ClientSession) -> float:
     """Get Device Type data response from IEC API."""
     response = await commons.send_get_request(session=session, url=GET_KWH_TARIFF_URL)

iec_api/iec_client.py

@@ -195,6 +195,40 @@ async def get_electric_bill(
 
         return await data.get_electric_bill(self._session, self._token, bp_number, contract_id)
 
+    async def save_invoice_pdf_to_file(
+        self,
+        file_path: str,
+        invoice_number: str | int,
+        bp_number: Optional[str | int] = None,
+        contract_id: Optional[str | int] = None,
+    ):
+        """
+        Get PDF of invoice from IEC api
+        Args:
+            self: The instance of the class.
+            file_path (str): Path to save the bill to
+            invoice_number (str): The requested invoice number
+            bp_number (str): The BP number of the meter.
+            contract_id (str): The contract ID associated with the meter.
+        """
+        await self.check_token()
+
+        if not bp_number:
+            bp_number = self._bp_number
+
+        assert bp_number, "BP number must be provided"
+
+        if not contract_id:
+            contract_id = self._contract_id
+
+        assert contract_id, "Contract ID must be provided"
+
+        response_bytes = await data.get_invoice_pdf(self._session, self._token, bp_number, contract_id, invoice_number)
+        if response_bytes:
+            f = open(file_path, "wb")
+            f.write(response_bytes)
+            f.close()
+
     async def get_devices(self, contract_id: Optional[str] = None) -> Optional[list[Device]]:
         """
         Get a list of devices for the user

iec_api/models/get_pdf.py

@@ -0,0 +1,18 @@
+from dataclasses import dataclass, field
+
+from mashumaro import DataClassDictMixin, field_options
+from mashumaro.config import BaseConfig
+
+
+@dataclass
+class GetPdfRequest(DataClassDictMixin):
+    """
+    Get PDF Request dataclass.
+    """
+
+    invoice_number: str = field(metadata=field_options(alias="invoiceNumber"))
+    contract_id: str = field(metadata=field_options(alias="contractId"))
+    bp_number: str = field(metadata=field_options(alias="bpNumber"))
+
+    class Config(BaseConfig):
+        serialize_by_alias = True