modules/connectors/providerInfomaniak/connectorInfomaniak.py

@@ -31,7 +31,6 @@ Path conventions (leading slash, ``ServiceAdapter`` paths always start with
     /{addressBookId}/{contactId}      -- single contact (.vcf download)
 """
 
-import json
 import logging
 import re
 from datetime import datetime, timedelta, timezone
@@ -392,115 +391,8 @@ class KdriveAdapter(ServiceAdapter):
             return DownloadResult()
         return DownloadResult(data=content, fileName=fileName, mimeType=mimeType)
 
-    async def _createDirectory(self, driveId: str, parentId: str, name: str) -> Optional[str]:
-        """Create a single directory and return its ID.
-
-        If the directory already exists (409), lists the parent to find
-        the existing folder's ID -- kDrive directory creation is not
-        idempotent.
-        """
-        url = f"{_API_BASE}/3/drive/{driveId}/files/{parentId}/directory"
-        headers = {
-            "Authorization": f"Bearer {self._token}",
-            "Content-Type": "application/json",
-        }
-        body = json.dumps({"name": name})
-        result = await _http.request("POST", url, headers=headers, data=body)
-
-        if isinstance(result, dict) and not result.get("error"):
-            data = _unwrapData(result)
-            if isinstance(data, dict) and data.get("id"):
-                return str(data["id"])
-
-        errorStr = str(result.get("error", "")) if isinstance(result, dict) else ""
-        if "already_exists" in errorStr or "409" in errorStr:
-            children = await self._listChildren(driveId, fileId=parentId, limit=1000)
-            for child in children:
-                if child.isFolder and child.name == name:
-                    return (child.metadata or {}).get("id") or child.path.strip("/").split("/")[-1]
-
-        logger.warning("kDrive mkdir %s/%s in %s failed: %s", driveId, name, parentId, result)
-        return None
-
-    async def _ensureDirectoryPath(self, driveId: str, parentId: str, pathSegments: List[str]) -> Optional[str]:
-        """Walk *pathSegments* and create each level that does not exist yet.
-
-        Returns the numeric folder ID of the deepest directory, or
-        ``None`` if any step fails.
-        """
-        currentId = parentId
-        for segment in pathSegments:
-            folderId = await self._createDirectory(driveId, currentId, segment)
-            if not folderId:
-                return None
-            currentId = folderId
-        return currentId
-
     async def upload(self, path: str, data: bytes, fileName: str) -> dict:
-        """Upload a file to kDrive.
+        return {"error": "kDrive upload not yet implemented"}
-
-        Path formats:
-          /{driveId}                          -> upload to drive root
-          /{driveId}/{folderId}               -> upload into folder by numeric ID
-          /{driveId}/{folderId}/Sub/Path      -> create Sub/Path under folderId, then upload
-          /{driveId}/Some/Human/Path          -> create path from drive root (id 1), then upload
-
-        Directories are created step-by-step via the v3 mkdir endpoint;
-        existing directories are reused (idempotent). File upload uses
-        the v3 upload endpoint (max 1 GB).
-        """
-        segments = [s for s in (path or "").strip("/").split("/") if s]
-        if not segments:
-            return {"error": "Upload path must include at least a drive ID"}
-        driveId = segments[0]
-
-        targetDirId: Optional[str] = None
-        if len(segments) > 1:
-            subSegments = segments[1:]
-            numericPrefix: List[str] = []
-            nameSegments: List[str] = []
-            for i, seg in enumerate(subSegments):
-                if seg.isdigit() and not nameSegments:
-                    numericPrefix.append(seg)
-                else:
-                    nameSegments = subSegments[i:]
-                    break
-
-            parentId = numericPrefix[-1] if numericPrefix else "1"
-
-            if nameSegments and nameSegments[-1] == fileName:
-                nameSegments = nameSegments[:-1]
-
-            if nameSegments:
-                targetDirId = await self._ensureDirectoryPath(driveId, parentId, nameSegments)
-                if not targetDirId:
-                    return {"error": f"Failed to create directory path: {'/'.join(nameSegments)}"}
-            else:
-                targetDirId = parentId
-
-        params = [
-            f"file_name={quote(fileName)}",
-            f"total_size={len(data)}",
-            "conflict=version",
-        ]
-        if targetDirId:
-            params.append(f"directory_id={targetDirId}")
-
-        endpoint = f"/3/drive/{driveId}/upload?{'&'.join(params)}"
-        url = f"{_API_BASE.rstrip('/')}/{endpoint.lstrip('/')}"
-        headers = {
-            "Authorization": f"Bearer {self._token}",
-            "Content-Type": "application/octet-stream",
-        }
-
-        result = await _http.request(
-            "POST", url, headers=headers, data=data,
-            timeout=aiohttp.ClientTimeout(total=120),
-        )
-        if isinstance(result, dict) and result.get("error"):
-            return result
-        unwrapped = _unwrapData(result) if isinstance(result, dict) else result
-        return unwrapped if isinstance(unwrapped, dict) else {"data": unwrapped}
 
     async def search(
         self,