Source code for ds_resource_plugin_py_lib.common.resource.dataset.result

"""
**File:** ``result.py``
**Region:** ``ds_resource_plugin_py_lib/common/resource/dataset``

Description
-----------
Dataclass capturing operation details for every dataset method call.
"""

from dataclasses import dataclass, field
from datetime import datetime
from typing import Any

from ds_common_serde_py_lib import Serializable

from .enums import DatasetMethod


[docs] @dataclass(kw_only=True) class OperationError(Serializable): """Structured error captured from a ``ResourceException``.""" message: str """The error message.""" code: str """The error code.""" status_code: int """The HTTP status code.""" details: dict[str, Any] = field(default_factory=dict) """The error details."""
[docs] @dataclass(kw_only=True) class OperationInfo(Serializable): """ Report produced by every dataset operation. Timing fields (``started_at``, ``ended_at``, ``duration_ms``) are populated automatically by the ``track_result`` decorator. Providers may set ``row_count``, ``schema``, or ``metadata`` inside their method; any value left at its default will be auto-derived from ``self.output`` after the method returns. Accessible on the dataset instance as ``self.operation``. """ method: DatasetMethod | None = None """The method that was called.""" success: bool = False """Whether the method call was successful.""" error: OperationError | None = None """The error captured from a ``ResourceException``.""" row_count: int = 0 """The number of rows read, written, or discovered.""" started_at: datetime | None = None """The timestamp when the method started.""" ended_at: datetime | None = None """The timestamp when the method ended.""" duration_ms: float = 0.0 """The duration of the method in milliseconds.""" schema: dict[str, Any] | None = None """The schema of the data.""" metadata: dict[str, Any] = field(default_factory=dict) """The metadata of the data."""