Feat get triangles url

ledger_analytics/__about__.py

@@ -1 +1 @@
-__version__ = '0.0.26'
+__version__ = '0.0.27'

ledger_analytics/requester.py

@@ -20,8 +20,11 @@ def _get_stream_chunks(**kwargs):
 
 
 class Requester(object):
-    def __init__(self, api_key: str) -> None:
-        self.headers = {"Authorization": f"Api-Key {api_key}"}
+    def __init__(self, api_key: str | None = None) -> None:
+        if api_key:
+            self.headers = {"Authorization": f"Api-Key {api_key}"}
+        else:
+            self.headers = {}
 
     def post(self, url: str, data: JSONDict):
         return self._factory("post", url, data)

ledger_analytics/triangle.py

@@ -1,10 +1,11 @@
 from __future__ import annotations
 
 import logging
+from tempfile import NamedTemporaryFile
 from typing import Optional
 
+import requests
 from bermuda import Triangle as BermudaTriangle
-from requests import Response
 from requests.exceptions import ChunkedEncodingError
 
 from .config import JSONDict
@@ -29,8 +30,8 @@ def __init__(
         self._id: str = id
         self._name: str = name
         self._data: JSONDict = data
-        self._get_response: Response | None = None
-        self._delete_response: Response | None = None
+        self._get_response: requests.Response | None = None
+        self._delete_response: requests.Response | None = None
         self._captured_stdout: str = ""
 
     id = property(lambda self: self._id)
@@ -56,14 +57,29 @@ def get(cls, id: str, name: str, endpoint: str, requester: Requester) -> Triangl
                 try:
                     retries += 1
                     get_response = requester.get(endpoint, stream=stream)
+                    if get_response.json().get("url") is not None:
+                        bytes = get_response.json().get("triangle_size_bytes")
+                        mb = 1_048_576
+                        console.log(
+                            f"Retrieving triangle from pre-signed URL of size {bytes / mb:.02f}MB."
+                        )
+                        url = get_response.json().get("url")
+                        url_response = requests.get(url)
+                        with NamedTemporaryFile(suffix=".trib") as f:
+                            f.write(url_response.content)
+                            triangle_data = BermudaTriangle.from_binary(
+                                f.name
+                            ).to_dict()
+                    else:
+                        triangle_data = (get_response.json().get("triangle_data"),)
                 except ChunkedEncodingError:
                     stream = True
                     continue
 
         self = cls(
             id,
             name,
-            get_response.json().get("triangle_data"),
+            triangle_data,
             endpoint,
             requester,
         )