Alpha/task9 - Done Enigx Service

.env.example

@@ -10,4 +10,4 @@ APP_PORT=8000
 MONGO_URL=mongodb://host:port
 DB_NAME=dbname
 
-JWT_SECRET_KEY=4ab2fcf7a6bc014396215ed36edb1c5d975c6574a2904132c03c
+JWT_SECRET_KEY=jwt_secret_key

.env.test.example

@@ -0,0 +1,3 @@
+TENANT_ID=tenant_id
+PROJECT_ID=project_id
+BEARER_TOKEN=bearer_token

.gitignore

@@ -1,5 +1,6 @@
 __pycache__
 .env
+.env.test
 .mypy_cache
 .pytest_cache
 .ruff_cache

app/schemas/fetch_config.py

@@ -20,8 +20,9 @@ class FetchDataType(StrEnum):
 
 
 class FetchData(BaseModel):
+    name: str | None
     type: FetchDataType
-    data: dict[str, Any] | list[Any] | str | None
+    data: bytes | dict[str, Any] | list[Any] | str | None
 
 
 class FetchTokenType(StrEnum):

app/services/enigx.py

@@ -0,0 +1,75 @@
+import json
+import mimetypes
+from datetime import UTC, datetime
+from http import HTTPStatus
+from typing import Any
+
+import requests
+from loguru import logger
+from requests import Response
+
+from app.core.config.fetch import config as config_fetch
+from app.schemas.fetch_config import FetchData, FetchDataType
+
+
+class EnigxService:
+    def __init__(self, tenant_id: str, project_id: str, bearer_token: str) -> None:
+        self.tenant_id = tenant_id
+        self.project_id = project_id
+        self.bearer_token = bearer_token
+
+    def get_def_headers(self) -> dict[str, str]:
+        return {
+            "x-tenant-id": self.tenant_id,
+            "Authorization": f"Bearer {self.bearer_token}",
+        }
+
+    def put_to_knowledge(self, data: FetchData) -> None:
+        if data.type is FetchDataType.FILE:
+            if isinstance(data.data, bytes):
+                resp = self.put_file(data.name, data.data)
+            else:
+                msg = f"Invalid file data type: {type(data.data)}"
+                raise TypeError(msg)
+        else:
+            resp = self.put_text(json.dumps(data.data, default=str))
+
+        if resp.status_code == HTTPStatus.OK:
+            logger.debug(json.dumps(resp.json(), indent=2))
+        else:
+            msg = f"Enigx error: {resp.status_code}: {resp.text}"
+            raise ValueError(msg)
+
+    def put_file(self, filename: str | None, file_bytes: bytes, options: dict[str, Any] | None = None) -> Response:
+        filename = filename or "unknown"
+
+        mime_type, _ = mimetypes.guess_type(filename)
+        mime_type = mime_type or "application/octet-stream"
+
+        files = {
+            "file": (filename, file_bytes, mime_type),
+            "fileOptionsJson": (None, json.dumps(options), "application/json"),
+        }
+
+        return requests.post(
+            url=f"https://api.enigx.com/v1/projects/{self.project_id}/knowledge/file",
+            headers={
+                **self.get_def_headers(),
+            },
+            files=files,  # type: ignore  # noqa: PGH003
+            timeout=config_fetch.REQUEST_TIMEOUT,
+        )
+
+    def put_text(self, text: str) -> Response:
+        return requests.post(
+            url=f"https://api.enigx.com/v1/projects/{self.project_id}/knowledge/text",
+            headers={
+                **self.get_def_headers(),
+                "Content-Type": "application/json",
+            },
+            json={
+                "name": f"text_{datetime.now(tz=UTC).isoformat()}",
+                "text": text,
+            },
+            timeout=config_fetch.REQUEST_TIMEOUT,
+        )

app/services/fetcher.py

@@ -1,3 +1,5 @@
+from urllib.parse import unquote
+
 import requests
 
 from app.core.config.fetch import config as config_fetch
@@ -36,17 +38,29 @@ def fetch(self) -> FetchData:
             raise ValueError(msg)
 
         resp_data = FetchData(
+            name=None,
             type=self.config.data_type,
             data=None,
         )
 
         if resp.status_code == self.config.success_code:
             if self.config.data_type is FetchDataType.JSON:
                 resp_data.data = resp.json()
+
             elif self.config.data_type is FetchDataType.FILE:
-                msg = "Not implemented for FILE type response"
-                raise NotImplementedError(msg)
+                content_disposition = resp.headers.get("Content-Disposition")
+                filename = "downloaded_file"
+                if content_disposition:
+                    parts = content_disposition.split("filename=")
+                    if len(parts) > 1:
+                        filename = unquote(parts[1].strip('"'))
+                resp_data.name = filename
+                resp_data.data = resp.content
+
             elif self.config.data_type is FetchDataType.HTML:
                 resp_data.data = resp.text
+        else:
+            msg = f"Fetch error: {resp.status_code}: {resp.text}"
+            raise ValueError(msg)
 
         return resp_data

test/core/env.py

@@ -0,0 +1,12 @@
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Config(BaseSettings):
+    model_config = SettingsConfigDict(env_file=".env.test", extra="allow")
+
+    TENANT_ID: str = ""
+    PROJECT_ID: str = ""
+    BEARER_TOKEN: str = ""
+
+
+config = Config()

test/integration_test/service/test_enigx.py

@@ -0,0 +1,38 @@
+import json
+import time
+from http import HTTPStatus
+
+from loguru import logger
+
+from app.services.enigx import EnigxService
+from test.core.env import config as config_env_test
+
+
+def test_enigx_put_file() -> None:
+    service = EnigxService(
+        tenant_id=config_env_test.TENANT_ID,
+        project_id=config_env_test.PROJECT_ID,
+        bearer_token=config_env_test.BEARER_TOKEN,
+    )
+    resp = service.put_file(
+        filename=f"test_{time.time()}.txt",
+        file_bytes=b"test",
+        options={
+            "description": "this is a test file upload",
+        },
+    )
+    logger.info(json.dumps(resp.json(), indent=2))
+
+    assert resp.status_code == HTTPStatus.OK
+
+
+def test_enigx_put_text() -> None:
+    service = EnigxService(
+        tenant_id=config_env_test.TENANT_ID,
+        project_id=config_env_test.PROJECT_ID,
+        bearer_token=config_env_test.BEARER_TOKEN,
+    )
+    resp = service.put_text("this is a test text upload")
+    logger.info(json.dumps(resp.json(), indent=2))
+
+    assert resp.status_code == HTTPStatus.OK