feat: support multiple object storage + parquet_csv + polars -> datafusion" (#3853)

* multiple storage

* all

* rm symlinks

* all

* all
This commit is contained in:
Ruben Fiszel
2024-05-31 11:57:41 +02:00
committed by Ruben Fiszel
parent 78d704b2d6
commit 0e7a25c0bd
21 changed files with 539 additions and 202 deletions
+3 -1
View File
@@ -440,7 +440,7 @@ class Windmill:
print(file_reader.read())
'''
"""
reader = S3BufferedReader(f"{self.workspace}", self.client, s3object["s3"], s3_resource_path)
reader = S3BufferedReader(f"{self.workspace}", self.client, s3object["s3"], s3_resource_path, s3object["storage"])
return reader
def write_s3_file(
@@ -479,6 +479,8 @@ class Windmill:
query_params["file_key"] = s3object["s3"]
if s3_resource_path is not None and s3_resource_path != "":
query_params["s3_resource_path"] = s3_resource_path
if s3object is not None and s3object["storage"] is not None:
query_params["storage"] = s3object["storage"]
try:
# need a vanilla client b/c content-type is not application/json here
+3 -1
View File
@@ -5,12 +5,14 @@ import httpx
class S3BufferedReader(BufferedReader):
def __init__(self, workspace: str, windmill_client: httpx.Client, file_key: str, s3_resource_path: str | None):
def __init__(self, workspace: str, windmill_client: httpx.Client, file_key: str, s3_resource_path: str | None, storage: str | None):
params = {
"file_key": file_key,
}
if s3_resource_path is not None:
params["s3_resource_path"] = s3_resource_path
if storage is not None:
params["storage"] = storage
self._context_manager = windmill_client.stream(
"GET",
f"/w/{workspace}/job_helpers/download_s3_file",
+1
View File
@@ -1,5 +1,6 @@
class S3Object(dict):
s3: str
storage: str | None
def __getattr__(self, attr):
return self[attr]