"""
**File:** ``file.py``
**Region:** ``ds_provider_grasp_py_lib/dataset/file``
Grasp File Dataset
This module implements a generic GRASP file dataset.
The dataset can read and create files via grasp API.
"""
import io
from dataclasses import dataclass, field
from typing import Any, Generic, NoReturn, TypeVar
import pandas as pd
from ds_common_logger_py_lib import Logger
from ds_protocol_http_py_lib.linked_service.http import HttpLinkedService
from ds_resource_plugin_py_lib.common.resource.dataset import (
DatasetSettings,
TabularDataset,
)
from ds_resource_plugin_py_lib.common.resource.errors import ResourceException
from ds_resource_plugin_py_lib.common.resource.linked_service.errors import AuthorizationError
from ds_resource_plugin_py_lib.common.serde.deserialize import PandasDeserializer
from ds_resource_plugin_py_lib.common.serde.serialize import PandasSerializer
from ..enums import ResourceType
logger = Logger.get_logger(__name__, package=True)
[docs]
@dataclass(kw_only=True)
class CreateSettings:
"""
Settings for create operations.
"""
acl: dict[str, Any] | None = field(default_factory=dict)
description: str | None = None
file_path: str | None = None
metadata: dict[str, Any] | None = field(default_factory=dict)
status: str | None = field(default="active")
tags: dict[str, Any] | None = field(default_factory=dict)
version: str | None = field(default="1.0.0")
content: io.BytesIO | None = field(default=None)
"""File content to be uploaded."""
[docs]
@dataclass(kw_only=True)
class ReadSettings:
"""
Settings for read operations.
"""
download_file: bool = True
limit: int = 500
offset: int = 0
order_by: str | None = None
tags: dict[str, str] | None = field(default_factory=dict)
meta: dict[str, str] | None = field(default_factory=dict)
id: str | None = None
file_path: str | None = None
created_at_gte: str | None = None
modified_at_gte: str | None = None
created_at_lte: str | None = None
modified_at_lte: str | None = None
status: str | None = None
[docs]
@dataclass(kw_only=True)
class GraspFileDatasetSettings(DatasetSettings):
"""Settings for Grasp file dataset operations."""
url: str | None = field(default="https://grasp-daas.com/api/file-dev/v2/file/")
create: CreateSettings = field(default_factory=CreateSettings)
read: ReadSettings = field(default_factory=ReadSettings)
GraspFileDatasetSettingsType = TypeVar(
"GraspFileDatasetSettingsType",
bound=GraspFileDatasetSettings,
)
HttpLinkedServiceType = TypeVar(
"HttpLinkedServiceType",
bound=HttpLinkedService[Any],
)
[docs]
@dataclass(kw_only=True)
class GraspFileDataset(
TabularDataset[
HttpLinkedServiceType,
GraspFileDatasetSettingsType,
PandasSerializer,
PandasDeserializer,
],
Generic[HttpLinkedServiceType, GraspFileDatasetSettingsType],
):
linked_service: HttpLinkedServiceType
settings: GraspFileDatasetSettingsType
@property
def type(self) -> ResourceType:
return ResourceType.DATASET_FILE
[docs]
def _details(self, **extra: Any) -> dict[str, Any]:
return {
"type": self.type.value,
"settings": self.settings.serialize(),
**extra,
}
[docs]
def _base_url(self) -> str:
if not self.settings.url:
raise ValueError("File dataset settings.url is required")
return self.settings.url.rstrip("/") + "/"
[docs]
def _read_params(self) -> dict[str, Any]:
"""Build query parameters for file listing from read settings."""
read = self.settings.read
params: dict[str, Any] = {}
for key in (
"limit",
"offset",
"order_by",
"id",
"file_path",
"created_at_gte",
"modified_at_gte",
"created_at_lte",
"modified_at_lte",
"status",
):
value = getattr(read, key)
if value is not None:
params[key] = value
for key, value in (read.tags or {}).items():
params[f"tag.{key}"] = value
for key, value in (read.meta or {}).items():
params[f"meta.{key}"] = value
return params
[docs]
def create(self) -> None:
"""
Write the content of the dataset to the file.
:return: None
"""
metadata = self._create_metadata()
data = self._upload_file_content(metadata)
self.output = pd.DataFrame(data)
[docs]
def read(self) -> None:
"""
Read the file from the file_uri.
:return: None
"""
base_url = self._base_url()
logger.debug(f"Reading files from {base_url}")
response = self.linked_service.connection.request(
method="GET",
url=base_url,
headers=self.linked_service.settings.headers,
params=self._read_params(),
)
files = response.json()["data"]
if self.settings.read.download_file:
for file in files:
file_id = file["id"]
url = f"{base_url}{file_id}/content/"
try:
response = self.linked_service.connection.request(
method="GET",
url=url,
headers=self.linked_service.settings.headers,
)
except ResourceException as exc:
if exc.status_code == 404:
file.update({"content": b""})
continue
file.update({"content": response.content})
self.output = pd.DataFrame(files)
else:
self.output = pd.DataFrame(files)
[docs]
def update(self) -> NoReturn:
raise AuthorizationError(
message="You are not authorized to update a Grasp File dataset",
status_code=403,
details={
"type": self.type.value,
"settings": self.settings.serialize(),
},
)
[docs]
def upsert(self) -> NoReturn:
raise AuthorizationError(
message="You are not authorized to upsert a Grasp File dataset",
status_code=403,
details={
"type": self.type.value,
"settings": self.settings.serialize(),
},
)
[docs]
def delete(self) -> NoReturn:
raise AuthorizationError(
message="You are not authorized to delete a Grasp File dataset",
status_code=403,
details={
"type": self.type.value,
"settings": self.settings.serialize(),
},
)
[docs]
def purge(self) -> NoReturn:
raise AuthorizationError(
message="You are not authorized to purge a Grasp File dataset",
status_code=403,
details={
"type": self.type.value,
"settings": self.settings.serialize(),
},
)
[docs]
def list(self) -> NoReturn:
raise AuthorizationError(
message="You are not authorized to list a Grasp File dataset",
status_code=403,
details={
"type": self.type.value,
"settings": self.settings.serialize(),
},
)
[docs]
def rename(self) -> NoReturn:
raise AuthorizationError(
message="You are not authorized to rename a Grasp File dataset",
status_code=403,
details={
"type": self.type.value,
"settings": self.settings.serialize(),
},
)
[docs]
def close(self) -> None:
"""Close the dataset."""
self.linked_service.close()
[docs]
def _upload_file_content(self, metadata: dict[str, Any]) -> dict[str, Any]:
"""
Upload the file content to the file.
:return: Dict
"""
headers = self.linked_service.settings.headers
headers.update(
{
"Content-Type": "application/octet-stream",
"accept": "*/*",
}
)
base_url = self._base_url()
url = f"{base_url}{metadata['id']}/content/"
response = self.linked_service.connection.request(
method="PUT",
url=url,
headers=headers,
data=self.settings.create.content,
)
data: dict[str, Any] = response.json()
return data