From 4374b661b0a0646864e87423c63ca54771fc9bb5 Mon Sep 17 00:00:00 2001 From: Maher Khalil Date: Sat, 22 Aug 2026 22:11:29 -0400 Subject: [PATCH] feat: updating storage with updates for version and sync return --- python/ai-server/README.md | 74 +++++++- .../src/ai_server/py_client/gaas/storage.py | 131 ++++++++++++-- .../src/ai_server/tests/test_storage.py | 160 ++++++++++++++++-- 3 files changed, 343 insertions(+), 22 deletions(-) diff --git a/python/ai-server/README.md b/python/ai-server/README.md index 9b00cbb..2d46c33 100644 --- a/python/ai-server/README.md +++ b/python/ai-server/README.md @@ -6,7 +6,7 @@ _ai-server-sdk_ is a python client SDK to connect to the AI Server - Inference with Models you have acces to within the server - Create Pandas DataFrame from Databases connections -- Pull Storage objects +- Push files to, and pull files from, Storage engines - Run pixel and get the direct output or full json response. - Pull data products from an existing insight using REST API. @@ -154,6 +154,78 @@ langhchain_db.updateQuery(query = 'UPDATE table_name set column1=value1 WHERE co langhchain_db.removeQuery(query = 'DELETE FROM table_name WHERE condition') ``` +### Move files in and out of Storage engines + +```python +# import the storage engine class for the ai_server package +from ai_server import StorageEngine + +storage = StorageEngine(engine_id="68b7e856-2312-4106-ab7a-7d7bb006173a", insight_id=server_connection.cur_insight) + +# list the root of the engine +storage.list(storagePath="/") +# example output +# ['reports/', 'college.csv'] + +# the same listing with details on each entry +storage.listDetails(storagePath="/reports") +# example output +# [{'Path': '/reports/q1.csv', 'Name': 'q1.csv', 'Size': 51049, 'MimeType': 'text/csv', +# 'ModTime': '2026-07-17T20:54:33.767Z', 'IsDir': False, 'Metadata': {'author': 'me'}}] +``` + +Paths are always relative to the root of the engine, and the `Path` on a listing entry can be handed straight back to any other method. Azure is the one engine where the root is the account rather than a single container, so its paths start with the container name, for example `mycontainer/reports/q1.csv`. Listing `/` there shows the containers you can reach. + +```python +# copy one file, or a folder, up to storage +storage.copyToStorage(storagePath="/reports", localPath="/local/path/q1.csv", metadata={"author": "me"}) + +# and back down again +storage.copyToLocal(storagePath="/reports/q1.csv", localPath="/local/path") + +# sync a whole local folder up, skipping files that are already there and unchanged +result = storage.syncLocalToStorage(storagePath="/reports", localPath="/local/reports") +# example output +# {'storagePath': 'reports', 'status': 'SUCCESS', +# 'uploadedFiles': ['reports/q1.csv'], 'skippedFiles': ['reports/q2.csv'], 'failedFiles': []} +``` + +`syncLocalToStorage` returns the outcome of the sync. A sync where some files failed comes back with a `status` of `PARTIAL` and the names in `failedFiles` **without raising**, so check the status to know that every file arrived: + +```python +if result["status"] != "SUCCESS": + print(f"{len(result['failedFiles'])} files did not make it: {result['failedFiles']}") +``` + +Engines that hand the whole transfer off in one call cannot name individual files, and report `SUCCESS` with empty lists. An empty `uploadedFiles` means "not reported", not "nothing uploaded". + +```python +# pull a whole folder down +storage.syncStorageToLocal(storagePath="/reports", localPath="/local/reports") + +# read a file without writing it to the insight workspace first +storage.getFileAsBase64(storagePath="/reports/q1.csv") + +# replace the metadata on a file already in storage. This overwrites rather than +# merges, so pass every key the file should end up with +storage.updateFileMetadata(storagePath="/reports/q1.csv", metadata={"author": "someone else"}) + +# delete a file or a folder. leaveFolderStructure keeps the folder itself visible +storage.deleteFromStorage(storagePath="/reports/q1.csv") +storage.deleteFromStorage(storagePath="/reports", leaveFolderStructure=True) +``` + +On engines that keep versions (S3 style, with bucket versioning turned on) you can list them and pull a specific one: + +```python +versions = storage.listVersions(storagePath="/reports/q1.csv") +# example output +# [{'versionId': 'CvV6ZQ...', 'lastModified': '2026-07-17T20:54:33.767Z', +# 'size': 51049, 'isLatest': True, 'key': 'reports/q1.csv'}] + +storage.copyToLocal(storagePath="/reports/q1.csv", localPath="/local/path", version=versions[1]["versionId"]) +``` + ### Run Function Engines ```python diff --git a/python/ai-server/src/ai_server/py_client/gaas/storage.py b/python/ai-server/src/ai_server/py_client/gaas/storage.py index 6643086..9dc88ee 100644 --- a/python/ai-server/src/ai_server/py_client/gaas/storage.py +++ b/python/ai-server/src/ai_server/py_client/gaas/storage.py @@ -31,12 +31,16 @@ def list(self, storagePath: str, insight_id: Optional[str] = None): """Lists the files and folders in a given storage path. Args: - storagePath: The path in the storage engine to list. + storagePath: The path in the storage engine to list. Use "/" for the root. + On Azure the root lists the containers in the account, and + every path below it starts with a container name, for example + "mycontainer/myfolder". insight_id: Optional; The unique identifier for the temporal workspace. If None, the session's default insight_id is used. Returns: - A list of files and folders in the specified path. + A list of names in the specified path. Folders come back with a trailing + slash, files without one. Raises: RuntimeError: If the server returns an error. @@ -50,12 +54,14 @@ def listDetails(self, storagePath: str, insight_id: Optional[str] = None): """Lists the files and folders in a given storage path with additional details. Args: - storagePath: The path in the storage engine to list. + storagePath: The path in the storage engine to list. Use "/" for the root. insight_id: Optional; The unique identifier for the temporal workspace. If None, the session's default insight_id is used. Returns: - A list of files and folders with additional details. + A list of dicts, one per entry, each with the keys "Path", "Name", "Size", + "MimeType", "ModTime", "IsDir" and "Metadata". "Path" is absolute within + the engine, so it can be handed straight back to any other method here. Raises: RuntimeError: If the server returns an error. @@ -63,6 +69,30 @@ def listDetails(self, storagePath: str, insight_id: Optional[str] = None): pixel = f'Storage("{self.engine_id}")|ListStoragePathDetails(storagePath="{storagePath}");' return self.__execute_pixel(pixel, insight_id) + def listVersions(self, storagePath: str, insight_id: Optional[str] = None): + """Lists the stored versions of a single file. + + Only meaningful on engines that keep versions, which today means S3 style + engines with bucket versioning turned on. + + Args: + storagePath: The path of the file in the storage engine. This has to name + a file, not a folder. + insight_id: Optional; The unique identifier for the temporal workspace. + If None, the session's default insight_id is used. + + Returns: + A list of dicts, newest first, each with "versionId", "lastModified", + "size", "isLatest" and "key". A "versionId" from here can be passed to + copyToLocal to pull that specific version. + + Raises: + RuntimeError: If the server returns an error, including when the engine + does not support versioning. + """ + pixel = f'Storage("{self.engine_id}")|ListStorageVersions(storagePath="{storagePath}");' + return self.__execute_pixel(pixel, insight_id) + def syncLocalToStorage( self, storagePath: str, @@ -83,13 +113,31 @@ def syncLocalToStorage( If None, the session's default insight_id is used. Returns: - True if the sync is successful, False otherwise. + A dict describing the outcome of the sync: + + { + "storagePath": "your/storage/path", + "status": "SUCCESS", + "uploadedFiles": ["your/storage/path/a.csv"], + "skippedFiles": ["your/storage/path/b.csv"], + "failedFiles": [], + } + + "status" is "SUCCESS" when nothing failed, "PARTIAL" when some files made + it and others did not, and "FAILED" when none did. A partial sync does not + raise, so check the status to know that every file arrived. + "skippedFiles" were already in storage and unchanged, so they were not + rewritten. + + Engines that hand the whole transfer off in a single call cannot name + individual files and report "SUCCESS" with empty lists, so an empty + "uploadedFiles" means "not reported", not "nothing uploaded". Raises: RuntimeError: If the server returns an error. """ spaceStr = f',space="{space}"' if space is not None else "" - metadataStr = f",metadata=[{metadata}]" if metadata is not None else "" + metadataStr = f",metadata=[{metadata}]" if metadata else "" pixel = f'Storage("{self.engine_id}")|SyncLocalToStorage(storagePath="{storagePath}",filePath="{localPath}"{spaceStr}{metadataStr});' return self.__execute_pixel(pixel, insight_id) @@ -127,6 +175,7 @@ def copyToLocal( storagePath: str, localPath: str, space: Optional[str] = None, + version: Optional[str] = None, insight_id: Optional[str] = None, ): """Copies files from a storage path to a local path. @@ -136,6 +185,9 @@ def copyToLocal( localPath: The destination path in the local application. space: Optional; The space to use (e.g., project ID, "user"). If None, the current insight space is used. + version: Optional; A version id from listVersions, to pull that version + instead of the current one. Only engines that keep versions + accept this. insight_id: Optional; The unique identifier for the temporal workspace. If None, the session's default insight_id is used. @@ -146,7 +198,8 @@ def copyToLocal( RuntimeError: If the server returns an error. """ spaceStr = f',space="{space}"' if space is not None else "" - pixel = f'Storage("{self.engine_id}")|PullFromStorage(storagePath="{storagePath}",filePath="{localPath}"{spaceStr});' + versionStr = f',version="{version}"' if version else "" + pixel = f'Storage("{self.engine_id}")|PullFromStorage(storagePath="{storagePath}",filePath="{localPath}"{spaceStr}{versionStr});' return self.__execute_pixel(pixel, insight_id) @@ -176,7 +229,7 @@ def copyToStorage( RuntimeError: If the server returns an error. """ spaceStr = f',space="{space}"' if space is not None else "" - metadataStr = f",metadata=[{metadata}]" if metadata is not None else "" + metadataStr = f",metadata=[{metadata}]" if metadata else "" pixel = f'Storage("{self.engine_id}")|PushToStorage(storagePath="{storagePath}",filePath="{localPath}"{spaceStr}{metadataStr});' return self.__execute_pixel(pixel, insight_id) @@ -207,6 +260,64 @@ def deleteFromStorage( return self.__execute_pixel(pixel, insight_id) + def updateFileMetadata( + self, + storagePath: str, + metadata: Dict, + insight_id: Optional[str] = None, + ): + """Replaces the metadata on a file already in storage. + + This rewrites the metadata rather than merging into it, so pass every key the + file should end up with. SMB/CIFS and SFTP have nowhere to keep user metadata + and ignore it. + + Args: + storagePath: The path of the file in the storage engine. + metadata: A dictionary of metadata to set on the file. Values are stored + as strings. + insight_id: Optional; The unique identifier for the temporal workspace. + If None, the session's default insight_id is used. + + Returns: + True if the update is successful. + + Raises: + RuntimeError: If the server returns an error. + """ + pixel = f'Storage("{self.engine_id}")|UpdateStorageFileMetadata(storagePath="{storagePath}",metadata=[{metadata}]);' + + return self.__execute_pixel(pixel, insight_id) + + def getFileAsBase64( + self, + storagePath: str, + convertToPdf: Optional[bool] = False, + insight_id: Optional[str] = None, + ): + """Reads a single file out of storage as a base64 string. + + Useful for handing a file to something that wants its bytes without writing it + to the insight workspace first. + + Args: + storagePath: The path of the file in the storage engine. + convertToPdf: Optional; If True, convert the file to a PDF before encoding + it. Defaults to False. + insight_id: Optional; The unique identifier for the temporal workspace. + If None, the session's default insight_id is used. + + Returns: + The file contents as a base64 encoded string. + + Raises: + RuntimeError: If the server returns an error. + """ + convertToPdfStr = ",convertToPdf=true" if convertToPdf else "" + pixel = f'Storage("{self.engine_id}")|GetStorageFileAsBase64(storagePath="{storagePath}"{convertToPdfStr});' + + return self.__execute_pixel(pixel, insight_id) + def to_langchain_storage(self): """Transform the storage engine into a langchain BaseStore object so that it can be used with langchain code""" from langchain_core.stores import BaseStore @@ -241,10 +352,10 @@ def syncStorageToLocal(self, localPath: str, storagePath: str) -> any: localPath=localPath, storagePath=storagePath ) - def copyToLocal(self, storageFilePath: str, localFolderPath: str) -> any: + def copyToLocal(self, storagePath: str, localPath: str) -> any: """Copy a specific file from the storage to the local system.""" return self.storage_engine.copyToLocal( - storageFilePath=storageFilePath, localFolderPath=localFolderPath + storagePath=storagePath, localPath=localPath ) def deleteFromStorage(self, storagePath: str) -> any: diff --git a/python/ai-server/src/ai_server/tests/test_storage.py b/python/ai-server/src/ai_server/tests/test_storage.py index db27c9f..dfa5fa4 100644 --- a/python/ai-server/src/ai_server/tests/test_storage.py +++ b/python/ai-server/src/ai_server/tests/test_storage.py @@ -1,17 +1,56 @@ +import os +import tempfile import unittest +import uuid + from ai_server.py_client.gaas.storage import StorageEngine from test_base_connection import TestServerClient from variables import STORAGE_ENGINE_ID +# every entry a listing hands back carries these, whether it is a file or a folder +LISTING_KEYS = {'Path', 'Name', 'Size', 'MimeType', 'ModTime', 'IsDir', 'Metadata'} + +# what a sync reports when it finishes +SYNC_STATUS_KEYS = {'storagePath', 'status', + 'uploadedFiles', 'skippedFiles', 'failedFiles'} + class StorageTests(TestServerClient): - def test_storage_list(self): - storage = StorageEngine( + storage = None + # each run works in its own folder so a failed run cannot leave anything + # behind that the next one trips over + test_folder = None + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.storage = StorageEngine( engine_id=STORAGE_ENGINE_ID, ) + cls.test_folder = f"sdk-tests/{uuid.uuid4()}" - storage_list = storage.list(storagePath='/my-new-test-folder/') + @classmethod + def tearDownClass(cls): + try: + cls.storage.deleteFromStorage(storagePath=cls.test_folder) + except RuntimeError as e: + # not worth failing the run over, but say so rather than leaving + # someone to wonder where the folder came from + print(f"Could not clean up {cls.test_folder}: {e}") + + def _write_local_files(self, directory, names): + """Writes a file per name and returns their full paths.""" + paths = [] + for name in names: + path = os.path.join(directory, name) + with open(path, 'w') as f: + f.write(f"contents of {name}") + paths.append(path) + return paths + + def test_storage_list(self): + storage_list = self.storage.list(storagePath="/") self.assertIsInstance(storage_list, list) @@ -19,21 +58,120 @@ def test_storage_list(self): self.assertIsInstance(storage_list[0], str) def test_storage_list_details(self): - storage = StorageEngine( - engine_id=STORAGE_ENGINE_ID, - ) - - storage_list = storage.listDetails(storagePath='/my-new-test-folder/') + storage_list = self.storage.listDetails(storagePath="/") self.assertIsInstance(storage_list, list) if len(storage_list) > 0: self.assertIsInstance(storage_list[0], dict) + self.assertCountEqual(storage_list[0].keys(), LISTING_KEYS) + + def test_storage_copy_round_trip(self): + with tempfile.TemporaryDirectory() as local_dir: + [local_file] = self._write_local_files(local_dir, ['round-trip.txt']) + storage_path = f"{self.test_folder}/copy" + + self.storage.copyToStorage( + storagePath=storage_path, localPath=local_file) + + details = self.storage.listDetails(storagePath=storage_path) + names = [item['Name'] for item in details] + self.assertIn('round-trip.txt', names) + self.assertCountEqual(details[0].keys(), LISTING_KEYS) + + # pull it back into a fresh directory so we know it really came down + with tempfile.TemporaryDirectory() as download_dir: + self.storage.copyToLocal( + storagePath=f"{storage_path}/round-trip.txt", + localPath=download_dir, + ) + self.assertIn('round-trip.txt', os.listdir(download_dir)) + + def test_storage_sync_reports_status(self): + storage_path = f"{self.test_folder}/sync" + + with tempfile.TemporaryDirectory() as local_dir: + self._write_local_files(local_dir, ['a.txt', 'b.txt']) + + status = self.storage.syncLocalToStorage( + storagePath=storage_path, localPath=local_dir) + + self.assertIsInstance(status, dict) + self.assertCountEqual(status.keys(), SYNC_STATUS_KEYS) + self.assertEqual(status['status'], 'SUCCESS') + self.assertEqual(status['failedFiles'], []) + self.assertEqual(len(status['uploadedFiles']), 2) + self.assertEqual(status['skippedFiles'], []) + + # nothing changed locally, so a second sync should upload none of it + second = self.storage.syncLocalToStorage( + storagePath=storage_path, localPath=local_dir) + + self.assertEqual(second['status'], 'SUCCESS') + self.assertEqual(second['uploadedFiles'], []) + self.assertEqual(len(second['skippedFiles']), 2) + + def test_storage_sync_to_local(self): + storage_path = f"{self.test_folder}/down" + + with tempfile.TemporaryDirectory() as local_dir: + self._write_local_files(local_dir, ['c.txt', 'd.txt']) + self.storage.syncLocalToStorage( + storagePath=storage_path, localPath=local_dir) + + with tempfile.TemporaryDirectory() as download_dir: + self.storage.syncStorageToLocal( + storagePath=storage_path, localPath=download_dir) + + self.assertCountEqual(os.listdir(download_dir), ['c.txt', 'd.txt']) + + def test_storage_metadata(self): + storage_path = f"{self.test_folder}/metadata" + + with tempfile.TemporaryDirectory() as local_dir: + [local_file] = self._write_local_files(local_dir, ['tagged.txt']) + self.storage.copyToStorage( + storagePath=storage_path, + localPath=local_file, + metadata={'author': 'sdk-tests'}, + ) + + details = self.storage.listDetails(storagePath=storage_path) + tagged = next(item for item in details if item['Name'] == 'tagged.txt') + + # SMB/CIFS and SFTP have nowhere to keep metadata, so only assert on it + # when the engine actually kept some + if tagged['Metadata']: + self.assertEqual(tagged['Metadata'].get('author'), 'sdk-tests') + + self.storage.updateFileMetadata( + storagePath=f"{storage_path}/tagged.txt", + metadata={'author': 'someone-else'}, + ) + + updated = self.storage.listDetails(storagePath=storage_path) + tagged = next( + item for item in updated if item['Name'] == 'tagged.txt') + # the update replaces the metadata rather than merging into it + self.assertEqual(tagged['Metadata'].get('author'), 'someone-else') + + def test_storage_delete_does_not_reach_siblings(self): + """A delete of "dir" must not take "dir-other" with it.""" + storage_path = f"{self.test_folder}/delete-me" + sibling_path = f"{self.test_folder}/delete-me-not" + + with tempfile.TemporaryDirectory() as local_dir: + [local_file] = self._write_local_files(local_dir, ['file.txt']) + self.storage.copyToStorage( + storagePath=storage_path, localPath=local_file) + self.storage.copyToStorage( + storagePath=sibling_path, localPath=local_file) - expected_keys = {'Path', 'Name', 'Size', - 'MimeType', 'ModTime', 'IsDir', 'Tier'} + self.storage.deleteFromStorage(storagePath=storage_path) - self.assertCountEqual(storage_list[0].keys(), expected_keys) + survivors = [item['Name'] + for item in self.storage.listDetails(storagePath=sibling_path)] + self.assertIn('file.txt', survivors) if __name__ == '__main__':