Skip to content

feat: updating storage with updates for version and sync return #37

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
themaherkhalil wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat: updating storage with updates for version and sync return #37

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
74 changes: 73 additions & 1 deletion python/ai-server/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.

Expand Down Expand Up @@ -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
Expand Down
131 changes: 121 additions & 10 deletions python/ai-server/src/ai_server/py_client/gaas/storage.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand All @@ -50,19 +54,45 @@ 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.
"""
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,
Expand All @@ -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)
Expand Down Expand Up @@ -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.
Expand All @@ -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.

Expand All @@ -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)

Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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:
Expand Down
Loading