feat: Add get EFS Messages API (#98)

iec_api/const.py

@@ -31,6 +31,7 @@
 GET_ELECTRIC_BILL_URL = IEC_API_BASE_URL + "ElectricBillsDrawers/ElectricBills/{contract_id}/{bp_number}"
 GET_CONTRACTS_URL = IEC_API_BASE_URL + "customer/contract/{bp_number}"
 GET_CHECK_CONTRACT_URL = IEC_API_BASE_URL + "customer/checkContract/{{contract_id}}/6"
+GET_EFS_MESSAGES_URL = IEC_API_BASE_URL + "customer/efs"
 GET_DEFAULT_CONTRACT_URL = GET_CONTRACTS_URL + "?count=1"
 GET_LAST_METER_READING_URL = IEC_API_BASE_URL + "Device/LastMeterReading/{contract_id}/{bp_number}"
 AUTHENTICATE_URL = IEC_API_BASE_URL + "Authentication/{id}/1/-1?customErrorPage=true"

iec_api/data.py

@@ -17,6 +17,7 @@
     GET_DEVICE_BY_DEVICE_ID_URL,
     GET_DEVICE_TYPE_URL,
     GET_DEVICES_URL,
+    GET_EFS_MESSAGES_URL,
     GET_ELECTRIC_BILL_URL,
     GET_INVOICE_PDF_URL,
     GET_KWH_TARIFF_URL,
@@ -35,6 +36,8 @@
 from iec_api.models.device import decoder as devices_decoder
 from iec_api.models.device_type import DeviceType
 from iec_api.models.device_type import decoder as device_type_decoder
+from iec_api.models.efs import EfsMessage, EfsRequestAllServices, EfsRequestSingleService
+from iec_api.models.efs import decoder as efs_decoder
 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
@@ -77,6 +80,37 @@ async def _get_response_with_descriptor(
     return response_with_descriptor.data
 
 
+async def _post_response_with_descriptor(
+    session: ClientSession,
+    jwt_token: JWT,
+    request_url: str,
+    json_data: Optional[dict],
+    decoder: BasicDecoder[ResponseWithDescriptor[T]],
+) -> T:
+    """
+    A function to retrieve a response with a descriptor using a JWT token and a URL.
+
+    Args:
+        jwt_token (JWT): The JWT token used for authentication.
+        request_url (str): The URL to send the request to.
+        json_data (dict): POST content
+
+    Returns:
+        T: The response with a descriptor, with its type specified by the return type annotation.
+    """
+    headers = commons.add_auth_bearer_to_headers(HEADERS_WITH_AUTH, jwt_token.id_token)
+    response = await commons.send_post_request(session=session, url=request_url, headers=headers, json_data=json_data)
+
+    response_with_descriptor = decoder.decode(response)
+
+    if not response_with_descriptor.data and not response_with_descriptor.response_descriptor.is_success:
+        raise IECError(
+            response_with_descriptor.response_descriptor.code, response_with_descriptor.response_descriptor.description
+        )
+
+    return response_with_descriptor.data
+
+
 async def get_accounts(session: ClientSession, token: JWT) -> Optional[list[Account]]:
     """Get Accounts response from IEC API."""
     return await _get_response_with_descriptor(session, token, GET_ACCOUNTS_URL, account_decoder)
@@ -117,6 +151,22 @@ async def get_remote_reading(
     return RemoteReadingResponse.from_dict(response)
 
 
+async def get_efs_messages(
+    session: ClientSession, token: JWT, contract_id: str, service_code: Optional[int] = None
+) -> Optional[list[EfsMessage]]:
+    """Get EFS Messages response from IEC API."""
+    if service_code:
+        req = EfsRequestSingleService(
+            contract_number=contract_id, process_type=1, service_code="EFS" + str(service_code).zfill(3)
+        )
+    else:
+        req = EfsRequestAllServices(contract_number=contract_id, process_type=1)
+
+    url = GET_EFS_MESSAGES_URL
+
+    return await _post_response_with_descriptor(session, token, url, json_data=req.to_dict(), decoder=efs_decoder)
+
+
 async def get_electric_bill(
     session: ClientSession, token: JWT, bp_number: str, contract_id: str
 ) -> Optional[ElectricBill]:

iec_api/iec_client.py

@@ -14,6 +14,7 @@
 from iec_api.models.customer import Account, Customer
 from iec_api.models.device import Device, Devices
 from iec_api.models.device_type import DeviceType
+from iec_api.models.efs import EfsMessage
 from iec_api.models.electric_bill import ElectricBill
 from iec_api.models.exceptions import IECLoginError
 from iec_api.models.invoice import GetInvoicesBody
@@ -171,7 +172,6 @@ async def get_contract_check(self, contract_id: Optional[str] = None) -> Optiona
 
         return await data.get_contract_check(self._session, self._token, contract_id)
 
-
     async def get_last_meter_reading(
         self, bp_number: Optional[str] = None, contract_id: Optional[str] = None
     ) -> Optional[MeterReadings]:
@@ -391,6 +391,27 @@ async def get_kwh_tariff(self) -> float:
 
         return self._kwh_tariff
 
+    async def get_efs_messages(
+        self, contract_id: Optional[str] = None, service_code: Optional[int] = None
+    ) -> Optional[list[EfsMessage]]:
+        """
+        Get EFS Messages for the contract
+        Args:
+            self: The instance of the class.
+            contract_id (str): The Contract ID of the meter.
+            service_code (str): Specific EFS Service code
+        Returns:
+            list[EfsMessage]: List of EFS Messages
+        """
+        await self.check_token()
+
+        if not contract_id:
+            contract_id = self._contract_id
+
+        assert contract_id, "Contract Id must be provided"
+
+        return await data.get_efs_messages(self._session, self._token, contract_id, service_code)
+
     # ----------------
     # Login/Token Flow
     # ----------------

iec_api/models/efs.py

@@ -0,0 +1,82 @@
+from dataclasses import dataclass, field
+from datetime import datetime
+
+from mashumaro import DataClassDictMixin, field_options
+from mashumaro.codecs import BasicDecoder
+from mashumaro.config import BaseConfig
+
+from iec_api.models.response_descriptor import ResponseWithDescriptor
+
+#   EFS stands for EFS service (Email, Fax, SMS) Messages
+#
+#   POST https://iecapi.iec.co.il//api/customer/efs
+#   BODY {
+#       "contractNumber":"123456",
+#       "processType":1,
+#       "serviceCode": "EFS004" <- optional
+#   }
+#
+#   Response:
+#
+# {
+#     "data": [
+#         {
+#             "serviceDescription": "הודעה על הפסקה מתוכננת",
+#             "partner": "123456",
+#             "contractNumber": "123456",
+#             "service": "EFS004",
+#             "subscribeDate": "2023-06-30T00:00:00",
+#             "subscribeTime": "0001-01-01T13:00:01",
+#             "isActive": true,
+#             "bpSubscription": false,
+#             "isSms": false,
+#             "communicationMethod": 1,
+#             "email": "",
+#             "sms": "055555555",
+#             "fax": "0000000000",
+#             "unsubscribeDate": "0001-01-01T00:00:00",
+#             "registrationStatus": 0
+#         }
+#     ],
+#     "reponseDescriptor": {
+#         "isSuccess": true,
+#         "code": "300",
+#         "description": ""
+#     }
+# }
+
+
+@dataclass
+class EfsRequestAllServices(DataClassDictMixin):
+    contract_number: str = field(metadata=field_options(alias="contractNumber"))
+    process_type: int = field(metadata=field_options(alias="processType"))
+
+    class Config(BaseConfig):
+        serialize_by_alias = True
+
+
+@dataclass
+class EfsRequestSingleService(EfsRequestAllServices):
+    service_code: str = field(metadata=field_options(alias="serviceCode"))
+
+
+@dataclass
+class EfsMessage(DataClassDictMixin):
+    service_description: str = field(metadata=field_options(alias="serviceDescription"))
+    partner: str = field(metadata=field_options(alias="partner"))
+    contract_number: str = field(metadata=field_options(alias="contractNumber"))
+    service: str = field(metadata=field_options(alias="service"))
+    subscribe_date: datetime = field(metadata=field_options(alias="subscribeDate"))
+    subscribe_time: datetime = field(metadata=field_options(alias="subscribeTime"))
+    is_active: bool = field(metadata=field_options(alias="isActive"))
+    bp_subscription: bool = field(metadata=field_options(alias="bpSubscription"))
+    is_sms: bool = field(metadata=field_options(alias="isSms"))
+    communication_method: int = field(metadata=field_options(alias="communicationMethod"))
+    email: str = field(metadata=field_options(alias="email"))
+    sms: str = field(metadata=field_options(alias="sms"))
+    fax: str = field(metadata=field_options(alias="fax"))
+    unsubscribe_date: datetime = field(metadata=field_options(alias="unsubscribeDate"))
+    registration_status: int = field(metadata=field_options(alias="registrationStatus"))
+
+
+decoder = BasicDecoder(ResponseWithDescriptor[list[EfsMessage]])