add Notify (#1)

notify/__init__.py

@@ -0,0 +1,81 @@
+from io import BufferedReader, BytesIO
+from typing import Any, Dict, Union
+
+import requests
+from pydantic import BaseModel, StrictStr
+from requests import Response
+
+from notify.models.rate_limit import RateLimit
+from notify.models.status import Status
+
+
+class Notify(BaseModel):
+    token: StrictStr
+    host: StrictStr = "https://notify-api.line.me"
+
+    def model_post_init(self, __context: Any) -> None:
+        try:
+            self.get_status()
+        except KeyError:
+            raise Exception("Invalid access token")
+
+    @property
+    def headers(self) -> Dict[str, str]:
+        return {
+            "Authorization": f"Bearer {self.token}",
+        }
+
+    def send_image_with_url(self, text: StrictStr, url: StrictStr) -> Response:
+        image_data = BytesIO()
+        r = requests.get(url)
+        r.raise_for_status()
+        image_data.write(r.content)
+        image_data.seek(0)
+        return self.send_image(text, image_data)
+
+    def send_image_with_local_path(
+        self, text: StrictStr, path: StrictStr
+    ) -> Response:
+        return self.send_image(text, open(path, "rb"))
+
+    def send_image(
+        self, text: StrictStr, image: Union[BufferedReader, BytesIO]
+    ) -> Response:
+        return self.send(
+            {
+                "message": (None, text),
+                "imageFile": image,
+            }
+        )
+
+    def send_text(self, text: str) -> Response:
+        return self.send({"message": (None, text)})
+
+    def send(self, files: Dict[str, Any]) -> Response:
+        return requests.post(
+            f"{self.host}/api/notify",
+            headers=self.headers,
+            files=files,
+        )
+
+    def get_status(self) -> Status:
+        j: Dict[str, Any] = requests.get(
+            f"{self.host}/api/status", headers=self.headers
+        ).json()
+        j["target_type"] = j.pop("targetType")
+        return Status(**j)
+
+    def revoke(self) -> Response:
+        return requests.post(f"{self.host}/api/revoke", headers=self.headers)
+
+    def get_rate_limit(self) -> RateLimit:
+        headers = requests.head(
+            f"{self.host}/api/status", headers=self.headers
+        ).headers
+        return RateLimit(
+            limit=int(headers["X-RateLimit-Limit"]),
+            remaining=int(headers["X-RateLimit-Remaining"]),
+            image_limit=int(headers["X-RateLimit-ImageLimit"]),
+            image_remaining=int(headers["X-RateLimit-ImageRemaining"]),
+            reset=int(headers["X-RateLimit-Reset"]),
+        )

notify/models/rate_limit.py

@@ -0,0 +1,15 @@
+from datetime import datetime
+
+from pydantic import BaseModel, StrictInt
+
+
+class RateLimit(BaseModel):
+    limit: StrictInt
+    remaining: StrictInt
+    image_limit: StrictInt
+    image_remaining: StrictInt
+    reset: StrictInt
+
+    @property
+    def reset_date(self) -> datetime:
+        return datetime.fromtimestamp(self.reset)

notify/models/status.py

@@ -0,0 +1,8 @@
+from pydantic import BaseModel
+
+
+class Status(BaseModel):
+    status: int
+    message: str
+    target_type: str
+    target: str

requirements-dev.txt

@@ -3,3 +3,5 @@ flake8==7.0.0
 isort==5.13.2
 mypy==1.10.0
 mypy-extensions==1.0.0
+python-dotenv==1.0.1
+types-requests==2.32.0.20240523

requirements.txt

@@ -0,0 +1,2 @@
+pydantic==2.7.1
+requests==2.32.2