Add Komens support (#15)

.gitignore

@@ -150,3 +150,5 @@ tmp_cache
 pytest_buckets.txt
 
 src/bakalari_api/first_login.py
+
+.data

CHANGELOG.md

@@ -2,13 +2,14 @@
 
 All notable changes to this project will be documented in this file.
 
-## [Unreleased]
+## [0.0.2]
 
 ### Added
 
 - better exceptions handling and logging
 - `class Komens`
   - count unread messages
+  - get all messages
   - tests and coverage
 
 ### Changed
@@ -18,7 +19,7 @@ All notable changes to this project will be documented in this file.
 - Refactor token handling
 
 ### Fixed
-  
+
 - Invalid refresh token
 - Refactor send_request to better maintenance
 
@@ -30,9 +31,9 @@ All notable changes to this project will be documented in this file.
 
   - supports saving `access token` and `refresh token` localy
   - automatically refreshes access token with refresh token if refresh token is not expired
-  
+
 - `class Schools` in `datastructures.py` lists all schools with their API points
-  
+
   - get_url by school name or index in list
   - search school by town
   - cache list of schools by saving and loading list in JSON format

requirements.txt

@@ -21,4 +21,5 @@ pytest_async==0.1.1
 setuptools==69.5.1
 StrEnum==0.4.15
 typing==3.7.4.3
-yarl==1.9.4
+yarl==1.9.4
+dateutils==0.6.12

src/bakalari_api/bakalari.py

@@ -110,6 +110,7 @@ async def send_unauth_request(
         Args:
             request (EndPoint): endpoint
             headers (str): headers for request
+            **kwargs (dict): kwargs
 
         Returns:
             str: JSON response or None

src/bakalari_api/komens.py

@@ -1,19 +1,155 @@
+"""Module for working with Komens."""
+
+from datetime import datetime as dt
+from typing import Any
+
+import dateutil
+import orjson
+
 from .bakalari import Bakalari
 from .const import EndPoint
+from .logger_api import api_logger
+
+log = api_logger("Bakalari API").get()
+
+
+class MessageContainer:
+    """Messages registry."""
+
+    mid: int
+    title: str
+    text: str
+    sent: dt.date
+    sender: dict[str, str]
+    attachments: dict[str, str]
+
+    def __init__(
+        self,
+        *,
+        mid: int,
+        title: str,
+        text: str,
+        sent: dt,
+        sender: dict[str, str],
+        attachments: dict[str, str] | None = None,
+    ):
+        """Initialize MessagesContainer."""
+
+        _setter = object.__setattr__
+        _setter(self, "mid", mid)
+        _setter(self, "title", title)
+        _setter(self, "text", text)
+        _setter(self, "sent", sent)
+        _setter(self, "sender", sender)
+        _setter(self, "attachments", attachments)
+
+    def __repr__(self) -> str:
+        """Representation of MessageContainer."""
+        return (
+            f"<MessageContainer message_id={self.mid} "
+            f"title={self.title} sender={self.sender}>"
+        )
+
+    def __setattr__(self, key: str, value: Any) -> None:
+        """Set an attribute."""
+        super().__setattr__(key, value)
+
+    def as_json(self) -> orjson.Fragment:
+        """Return JSON fragment of message."""
+        json_repr = {
+            "mid": self.mid,
+            "title": self.title,
+            "text": self.text,
+            "sent": self.sent,
+            "sender": self.sender,
+            "attachments": self.attachments,
+        }
+        return orjson.Fragment(json_repr)
+
+    def __str__(self) -> str:
+        """Return string representation of data."""
+        return f"""Message id: {self.mid}
+            title: {self.title}
+            text: {self.text},
+            sent: {self.sent.date()},
+            sender: {self.sender},
+            attachments: {self.attachments}"""
+
+
+class Messages(list[MessageContainer]):
+    """Messages class holds all messages."""
+
+    def __init__(self) -> None:
+        """Messages class holds all messages."""
+        super().__init__()
+
+    def add_message(self, data: MessageContainer):
+        """Add new message to list."""
+        self.append([data])
+
+    def get_message_by_id(self, id: int) -> MessageContainer:
+        """Get message by id."""
+        for i in self:
+            if i[0].mid == id:
+                return i[0]
+
+    def get_messages_by_date(
+        self, date: dt, to_date: dt | None = None
+    ) -> list[MessageContainer]:
+        """Get messages by date.
+
+        If `to_date` is set, then returns list of range from `date` to `to_date`
+        """
+
+        messages = []
+
+        for i in self:
+            if to_date:
+                if (i[0].sent.date() >= date) and (i[0].sent.date() <= to_date):
+                    messages.append(i[0])
+            elif i[0].sent.date() == date:
+                messages.append(i[0])
+        return None if len(messages) == 0 else messages
+
+    def count_messages(self) -> int:
+        """Count messages."""
+        return len(self)
 
 
 class Komens:
-    """Class for manipulating Komens messages."""
+    """Class for working with Komens messages."""
 
     def __init__(self, bakalari: Bakalari):
         """Initialize class Komens."""
-        self._bakalari = bakalari
+        self.bakalari = bakalari
+        self.messages = Messages()
 
-    async def messages(self):
+    async def get_messages(self) -> Messages:
         """Get unread messages."""
-        return await self._bakalari.send_auth_request(EndPoint.KOMENS_UNREAD)
+        messages = await self.bakalari.send_auth_request(
+            EndPoint.KOMENS_UNREAD,
+        )
+
+        _messages = Messages()
+
+        # with open(".data", "rb") as f:
+        #     messages = orjson.loads(f.read())
+        #     f.close()
+
+        for msg in messages["Messages"]:
+            log.debug(f"Writing message: {msg}")
+            _message = MessageContainer(
+                mid=msg["Id"],
+                title=msg["Title"],
+                text=msg["Text"],
+                sent=dateutil.parser.parse(msg["SentDate"]),
+                sender=msg["Sender"]["Name"],
+                attachments=msg["Attachments"],
+            )
+            self.messages.add_message(_message)
+
+        return _messages
 
-    async def count_unread_messages(self):
+    async def count_unread_messages(self) -> int:
         """Get count of unreaded messages."""
-        return await self._bakalari.send_auth_request(
-            EndPoint.KOMENS_UNREAD_COUNT)
+        return await self.bakalari.send_auth_request(EndPoint.KOMENS_UNREAD_COUNT)

tests/test_komens.py

@@ -0,0 +1,148 @@
+import datetime as dt
+
+from aioresponses import aioresponses
+import orjson
+from src.bakalari_api.bakalari import Bakalari
+from src.bakalari_api.const import EndPoint
+from src.bakalari_api.komens import Komens, Messages
+
+fs = "http://fake_server"
+
+payload = """
+{
+  "Messages": [
+    {
+      "$type": "GeneralMessage",
+      "Id": "fake_id1",
+      "Title": "",
+      "Text": "fake_text_id1",
+      "SentDate": "2024-01-01T13:37:28+02:00",
+      "Sender": {
+        "$type": "Sender",
+        "Id": "fake_sender_ID",
+        "Type": "teacher",
+        "Name": "fake_teacher_name1"
+      },
+      "Attachments": [
+        {
+          "$type": "AttachmentInfo",
+          "Id": "fake_attachment_id1",
+          "Name": "fake_atachement_name1",
+          "Type": "fake_type",
+          "Size": 12345
+        }
+      ],
+      "Read": true,
+      "LifeTime": "ToRead",
+      "DateFrom": null,
+      "DateTo": null,
+      "Confirmed": true,
+      "CanConfirm": false,
+      "Type": "OBECNA",
+      "CanAnswer": true,
+      "Hidden": false,
+      "CanHide": true,
+      "CanDelete": false,
+      "RelevantName": "fake_relevant_name1",
+      "RelevantPersonType": "fake_relevant_person1"
+    },
+    {
+      "$type": "GeneralMessage",
+      "Id": "fake_id2",
+      "Title": "",
+      "Text": "fake_text_id2",
+      "SentDate": "2024-01-05T13:37:28+02:00",
+      "Sender": {
+        "$type": "Sender",
+        "Id": "fake_sender_ID",
+        "Type": "teacher",
+        "Name": "fake_teacher_name2"
+      },
+      "Attachments": [],
+      "Read": true,
+      "LifeTime": "ToRead",
+      "DateFrom": null,
+      "DateTo": null,
+      "Confirmed": true,
+      "CanConfirm": false,
+      "Type": "OBECNA",
+      "CanAnswer": true,
+      "Hidden": false,
+      "CanHide": true,
+      "CanDelete": false,
+      "RelevantName": "fake_relevant_name2",
+      "RelevantPersonType": "fake_relevant_person2"
+    }
+   ]}"""
+
+
+async def test_komens_get_messages():
+
+    bakalari = Bakalari(fs)
+    komens = Komens(bakalari)
+    bakalari.credentials.access_token = "token"
+
+    with aioresponses() as m:
+        m.post(
+            url=fs + EndPoint.KOMENS_UNREAD.get("endpoint"),
+            body=payload,
+            headers={},
+            status=200,
+        )
+
+        await komens.get_messages()
+        msg = komens.messages.get_message_by_id("fake_id1")
+        assert isinstance(komens.messages, Messages)
+        assert komens.messages.count_messages() == 2
+
+        assert msg.mid == "fake_id1"
+        assert msg.sender == "fake_teacher_name1"
+        assert msg.text == "fake_text_id1"
+        assert msg.title == ""
+
+        msg = komens.messages.get_messages_by_date(dt.date(2024, 1, 1))
+        assert msg[0].mid == "fake_id1"
+
+        msg = komens.messages.get_messages_by_date(
+            dt.date(2024, 1, 1), to_date=dt.date(2024, 1, 1) + dt.timedelta(days=+5)
+        )
+        assert isinstance(msg, list)
+        assert len(msg) == 2
+        assert msg[1].mid == "fake_id2"
+
+        assert (
+            str(msg[1])
+            == """Message id: fake_id2
+            title: 
+            text: fake_text_id2,
+            sent: 2024-01-05,
+            sender: fake_teacher_name2,
+            attachments: []"""
+        )
+
+        assert (
+            repr(msg[0])
+            == "<MessageContainer message_id=fake_id1 title= sender=fake_teacher_name1>"
+        )
+
+        assert isinstance(msg[0].as_json(), orjson.Fragment)
+
+        msg[0].title = "new_set_title"
+        assert msg[0].title == "new_set_title"
+
+
+async def test_komens_count_unread_messages():
+
+    bakalari = Bakalari(fs)
+    komens = Komens(bakalari)
+    bakalari.credentials.access_token = "token"
+
+    with aioresponses() as m:
+        m.get(
+            url=fs + EndPoint.KOMENS_UNREAD_COUNT.get("endpoint"),
+            body="50",
+            headers={},
+            status=200,
+        )
+
+        assert await komens.count_unread_messages() == 50