feat: Create a Exporter class to allow export data from LandingLens

landingai/data_management/__init__.py

@@ -1,2 +1,3 @@
 from .media import Media  # noqa
 from .metadata import Metadata  # noqa
+from .export import Exporter  # noqa

landingai/data_management/client.py

@@ -26,6 +26,7 @@
 GET_DEFECTS = "get_defects"
 GET_PROJECT_MODEL_INFO = "get_project_model_info"
 GET_FAST_TRAINING_EXPORT = "get_fast_training_export"
+EVENT_LOGS = "event_logs"
 
 
 ROUTES = {
@@ -94,6 +95,11 @@
         "endpoint": "api/{version}/project/with_users",
         "method": requests.get,
     },
+    EVENT_LOGS: {
+        "root_url": "LANDING_API",
+        "endpoint": "api/event/export",
+        "method": requests.post,
+    },
 }
 
 _URL_ROOTS = {

landingai/data_management/export.py

@@ -0,0 +1,70 @@
+from typing import Optional
+
+from landingai.data_management.client import EVENT_LOGS, LandingLens
+from datetime import datetime, timezone
+import requests
+import logging
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class Exporter:
+    """Export management API client.
+    This class provides a set of APIs to export data from LandingLens.
+    For example, you can use this class to export all the available event logs of your organization.
+
+    Example
+    -------
+    >>> client = Exporter(project_id, api_key)
+    >>> client.export_event_logs("2023-06-01", '/path/to/save/file.csv')
+    >>> # The csv file will be saved in the desired path.
+
+    Parameters
+    ----------
+    project_id: int
+        LandingLens project id.  Can override this default in individual commands.
+    api_key: Optional[str]
+        LandingLens API Key. If it's not provided, it will be read from the environment variable LANDINGAI_API_KEY, or from .env file on your project root directory.
+    """
+
+    def __init__(self, project_id: int, api_key: Optional[str] = None):
+        self._client = LandingLens(project_id=project_id, api_key=api_key)
+
+    def export_event_logs(self, from_date: str, save_path: str) -> None:
+        """Exports the event logs of the organization from the given time range.
+
+        Parameters
+        ----------
+        from_date: str
+            In date following the format "YYYY-MM-DD"
+        save_path:
+            Desired path to save the csv file to: '/path/to/save/file.csv'
+
+        Returns
+        ----------
+        None
+            The csv file will be saved in the desired path.
+        """
+        try:
+            datetime.strptime(from_date, "%Y-%m-%d")
+        except ValueError:
+            raise ValueError("from_date must be in YYYY-MM-DD format")
+        from_timestamp = datetime.strptime(
+            f"{from_date} 00:00:00.00000", "%Y-%m-%d %H:%M:%S.%f"
+        )
+        to_timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S.%f")
+
+        _LOGGER.info("Exporting event logs...")
+        resp = self._client._api(
+            EVENT_LOGS,
+            params={"fromTimestamp": from_timestamp, "toTimestamp": to_timestamp},
+        )
+        signed_url = resp["data"].get("signedUrl")
+        _LOGGER.debug("Signed URL: ", signed_url)
+        self._download_file_from_signed_url(signed_url, save_path)
+        print(f"Event logs exported successfully to path: {save_path}")
+
+    def _download_file_from_signed_url(self, signed_url: str, save_path: str) -> None:
+        response = requests.get(signed_url)
+        with open(save_path, "wb") as file:
+            file.write(response.content)

pyproject.toml

@@ -33,6 +33,7 @@ tqdm = ">=4.64.0,<5.0.0"
 tenacity = "^8.2.3"
 pandas = "2.0"
 ruff = "^0.1.8"
+freezegun = "^1.4.0"
 
 [tool.poetry.group.dev.dependencies]
 autoflake = "1.*"

tests/data/responses/test_export_event_logs.yaml

@@ -0,0 +1,8 @@
+responses:
+- response:
+    auto_calculate_content_length: false
+    body: '{"code":0,"message":"","data":{"s3Path":"s3://.../1622505600-1706572800.csv","signedUrl":"https://landinglens-bucket.s3.somepath/1622505600-1706572800.csv?X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential...-SignedHeaders=host"}}'
+    content_type: text/plain
+    method: POST
+    status: 200
+    url: https://app.landing.ai/api/event/export?fromTimestamp=2021-06-01 00:00:00&toTimestamp=2024-01-30 00:00:00.000000

tests/unit/landingai/data_management/test_export.py

@@ -0,0 +1,19 @@
+import responses
+
+from landingai.data_management.export import Exporter
+from freezegun import freeze_time
+from unittest import mock
+
+
+@responses.activate
+@mock.patch("landingai.data_management.export.Exporter._download_file_from_signed_url")
+@freeze_time("2024-01-30 00:00:00.000000")  # Mock the current date and time
+def test_export_event_logs(mocked_method):
+    responses._add_from_file(
+        file_path="tests/data/responses/test_export_event_logs.yaml"
+    )
+    project_id = 12345
+    api_key = "land_sk_12345"
+    client = Exporter(project_id, api_key)
+    res = client.export_event_logs("2021-06-01", "./local/path/to/save/file.csv")
+    assert res is None