gitea_admin/sientia-dataops-laborious_temporal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add laborious

Browse Source
This commit is contained in:
vitor-aignosi
2026-09-01 13:28:13 -03:00
parent 0081b4f2f8
commit 984bbf3bfc
39 changed files with 10967 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
laborious/utils/models/__init__.py Normal file
View File

506
laborious/utils/models/import_errors.py Normal file
View File

@@ -0,0 +1,506 @@
"""The import's failure vocabulary: a closed catalog, typed errors and one reporting channel.
Three things live here and nowhere else:
- **the catalog** — one stable `error_code` and one plain-language `error_reason` per pipeline step.
The reason is written for a person who does not know this system exists: what happened, what state
the model is in, what to do next. It never carries a path, a table name, an object key, a Python
exception class or a traceback, and it is never built from `str(exception)`.
- **the typed errors** — the two failures that must not be retried, because repeating them cannot
change the answer: a record this workflow does not own, and a statement the database refuses.
- **the channel** — `report_import_failure`, the single funnel every failure passes through. It is
the only place in the import path that logs an error or sends a notification, which a test
asserts, so no failure is reported twice or half-reported.
"""
from temporalio import workflow
with workflow.unsafe.imports_passed_through():
import traceback
from collections.abc import Mapping
from dataclasses import dataclass
from enum import StrEnum
from types import MappingProxyType
from typing import Any, Final, Protocol
from temporalio.exceptions import ApplicationError
from laborious import metrics
from laborious.utils.bundle.steps import ImportStep
class ImportErrorCode(StrEnum):
"""The machine-readable half of a failure: the frontend's i18n key and support's triage key.
Written under `import.code` in the record's JSON detail — not in a column of its own, so the
import adds nothing to a table it shares with the training flow.
"""
REQUEST_INVALID = 'IMPORT_REQUEST_INVALID'
FILE_UNREADABLE = 'IMPORT_FILE_UNREADABLE'
BUNDLE_CANNOT_BE_OPENED = 'IMPORT_BUNDLE_CANNOT_BE_OPENED'
BUNDLE_REJECTED = 'IMPORT_BUNDLE_REJECTED'
BUNDLE_UNPACK_FAILED = 'IMPORT_BUNDLE_UNPACK_FAILED'
BUNDLE_INCOMPLETE = 'IMPORT_BUNDLE_INCOMPLETE'
BUNDLE_UNEXPECTED_CONTENT = 'IMPORT_BUNDLE_UNEXPECTED_CONTENT'
STORAGE_PREPARATION_FAILED = 'IMPORT_STORAGE_PREPARATION_FAILED'
MODEL_FILES_NOT_STORED = 'IMPORT_MODEL_FILES_NOT_STORED'
MODEL_NOT_REGISTERED = 'IMPORT_MODEL_NOT_REGISTERED'
MODEL_SETTINGS_NOT_SAVED = 'IMPORT_MODEL_SETTINGS_NOT_SAVED'
UNEXPECTED_ERROR = 'IMPORT_UNEXPECTED_ERROR'
# Codes that never reach a record's `error_message`, and are therefore absent from the catalog:
# one names the failure to own a row, one the failure to write one, one the temporary files left
# on the worker after an import that otherwise succeeded. All three live on the channel — log,
# metric, notification, Temporal error — and the third is additionally flagged in the record's
# detail as `cleanup_failed`, without a code and without changing the status.
LOG_ROW_NOT_CLAIMABLE = 'IMPORT_LOG_ROW_NOT_CLAIMABLE'
STATUS_NOT_RECORDED = 'IMPORT_STATUS_NOT_RECORDED'
CLEANUP_INCOMPLETE = 'IMPORT_CLEANUP_INCOMPLETE'
@dataclass(frozen=True)
class ImportFailure:
"""One classification of one failure: the step, its code and the sentence it stores."""
step: ImportStep | None
code: ImportErrorCode
reason: str
_CONTACT_SUPPORT = "Contact support with this import's identifier."
_RE_EXPORT = 'Export the model again from the origin platform.'
IMPORT_FAILURE_CATALOG: Final[Mapping[ImportStep, tuple[ImportErrorCode, str]]] = MappingProxyType(
{
ImportStep.RECEIVED: (
ImportErrorCode.REQUEST_INVALID,
'The import could not be started because the request was incomplete or '
'contradictory. Nothing was created. Submit the import again.',
),
ImportStep.DOWNLOAD: (
ImportErrorCode.FILE_UNREADABLE,
'The uploaded file could not be read, or it is not the file that was sent. '
'Upload it again.',
),
ImportStep.DECRYPTION: (
ImportErrorCode.BUNDLE_CANNOT_BE_OPENED,
'This file could not be opened: the password is wrong, or the file is damaged or '
'was altered. Confirm the password with whoever exported the model, then try '
'again.',
),
ImportStep.ARCHIVE_INSPECTION: (
ImportErrorCode.BUNDLE_REJECTED,
"The file's contents did not pass the platform's safety checks, so it was not "
f'opened. {_RE_EXPORT}',
),
ImportStep.EXTRACTION: (
ImportErrorCode.BUNDLE_UNPACK_FAILED,
f'The file could not be unpacked safely and nothing from it was kept. {_RE_EXPORT}',
),
ImportStep.STRUCTURE_VALIDATION: (
ImportErrorCode.BUNDLE_INCOMPLETE,
'This file is not a complete model export — part of what the platform needs is '
f'missing from it. {_RE_EXPORT}',
),
ImportStep.CONTENT_POLICY: (
ImportErrorCode.BUNDLE_UNEXPECTED_CONTENT,
'The file contains something a model export should not contain, so it was '
f'rejected. {_RE_EXPORT}',
),
ImportStep.EXPERIMENT_CREATION: (
ImportErrorCode.STORAGE_PREPARATION_FAILED,
'The platform could not prepare a place to keep this model. The import stopped '
f'and no model was created. {_CONTACT_SUPPORT}',
),
ImportStep.ARTIFACT_UPLOAD: (
ImportErrorCode.MODEL_FILES_NOT_STORED,
"The model's files could not be stored on this platform. The import stopped and "
f'the model is not available. {_CONTACT_SUPPORT}',
),
ImportStep.REGISTRATION: (
ImportErrorCode.MODEL_NOT_REGISTERED,
"The model's files were stored, but the model itself could not be registered, so "
f'it cannot be used. {_CONTACT_SUPPORT}',
),
ImportStep.MODEL_DOCUMENT: (
ImportErrorCode.MODEL_SETTINGS_NOT_SAVED,
'The model was registered, but the settings that tell the platform how to run it '
f'could not be saved. {_CONTACT_SUPPORT}',
),
}
)
UNCLASSIFIED_FAILURE: Final[tuple[ImportErrorCode, str]] = (
ImportErrorCode.UNEXPECTED_ERROR,
f'The import stopped for an unexpected reason and no model was created. {_CONTACT_SUPPORT}',
)
# `cleanup` deletes temporary files on the worker. It is never the reported failure: when every
# other step succeeded and only cleanup failed, the import is a success and the leak is a log line.
# So it deliberately has no user-facing sentence, and the catalog-completeness test knows it.
STEPS_WITHOUT_A_USER_FACING_REASON: Final[frozenset[ImportStep]] = frozenset({ImportStep.CLEANUP})
def classify_import_failure(
step: ImportStep | None, exception: BaseException | None = None
) -> ImportFailure:
"""Turn a failure into the one code and the one sentence that will be recorded.
The exception's text is **ignored entirely** — it is what carries paths, table names and type
names, and it belongs in the log. What decides the outcome is the step.
Args:
step: the step that was being attempted, or `None` when none had been entered.
exception: the failure, when there is one to hand. It is accepted so callers read as
"classify this failure" and never absent because the classification needs it: the
catalog is keyed by step, and an exception cannot cross an activity boundary with its
type intact anyway — the terminal write classifies from the step alone.
Returns:
ImportFailure: step, code and the reason to store.
"""
del exception # the catalog is keyed by step; the exception's text never reaches the record
if step is None or step not in IMPORT_FAILURE_CATALOG:
code, reason = UNCLASSIFIED_FAILURE
return ImportFailure(step=step, code=code, reason=reason)
code, reason = IMPORT_FAILURE_CATALOG[step]
return ImportFailure(step=step, code=code, reason=reason)
class ImportLogRowNotClaimableError(ApplicationError):
"""The record named by `import_run_id` is not this workflow's to write.
Raised when the claim affects zero rows: no such row, a `TRAIN` row, or a row already running
under another workflow. Non-retryable on purpose — a second attempt would read the same rows and
reach the same conclusion — and loud everywhere except the row itself, which belongs to someone
else.
"""
def __init__(self, message: str, *, import_run_id: Any = None, observed: str = '') -> None:
super().__init__(
message,
type='ImportLogRowNotClaimableError',
non_retryable=True,
)
self.import_run_id = import_run_id
self.observed = observed
class ImportStatusRejectedError(ApplicationError):
"""The database accepted the connection and refused the statement.
A `CHECK` violation, an undefined column, a value too long for its column. `V12` not being
applied yet lands here, at the terminal write, and it must fail on the first attempt instead of
spending the whole retry envelope to say so.
"""
def __init__(self, message: str) -> None:
super().__init__(message, type='ImportStatusRejectedError', non_retryable=True)
class ImportInputError(ApplicationError):
"""The request is incomplete or contradictory. Fails at `received`, on the claimed row."""
def __init__(self, message: str) -> None:
super().__init__(
message,
{'step': ImportStep.RECEIVED.value, 'gate': None},
type='ImportInputError',
non_retryable=True,
)
class ImportGateRejectedError(ApplicationError):
"""A gate rejected the bundle, and the error says which one.
Gates 2 to 7 all run inside one activity, so the step the workflow was *attempting* is not
precise enough to record. An exception loses its Python type crossing an activity boundary but
keeps its `details`, so the step and the gate travel there and the terminal write can name them.
"""
def __init__(
self,
message: str,
*,
step: ImportStep,
gate: int | None = None,
cause: str | None = None,
) -> None:
super().__init__(
message,
{'step': step.value, 'gate': gate, 'cause': cause},
type='ImportGateRejectedError',
non_retryable=True,
)
self.step = step
self.gate = gate
_SUMMARY_LIMIT = 400
_CHAIN_LIMIT = 10
_UNKNOWN_ERROR_TYPE = 'unavailable'
def _root_cause(exception: BaseException) -> BaseException:
"""The deepest `__cause__` in the chain: the failure that actually happened.
A `MlflowException` reaches this channel wrapped — by an `ActivityError`, by an
`ApplicationError`, by `requests`' own re-raises — and the outermost type is the wrapper's, which
says nothing an operator can act on. The chain is walked to its end, and bounded, because a cycle
in `__cause__` would otherwise hang the reporting of a failure.
"""
candidate = exception
for _ in range(_CHAIN_LIMIT):
if candidate.__cause__ is None:
break
candidate = candidate.__cause__
return candidate
def _type_name(exception: BaseException) -> str:
"""The name to report a failure under: `ApplicationError.type` when there is one.
An activity's own exception type is the one thing worth grouping failures by, and Temporal keeps
it in `ApplicationError.type` rather than in the Python class — a `MlflowException` raised in an
activity reaches the workflow as an `ApplicationError` whose `type` still says `MlflowException`.
Reporting the class name here would say `ApplicationError` for every activity failure there is.
"""
return str(getattr(exception, 'type', None) or type(exception).__name__)
def _one_line(text: str) -> str:
"""The first non-empty line, bounded — a log *field* has to stay one field.
The traceback still reaches the log in full, in the message. This is the value a dashboard groups
by and an alert quotes, so a 40-frame `urllib3` chain would make it useless.
"""
for line in text.splitlines():
stripped = line.strip()
if stripped:
return stripped[:_SUMMARY_LIMIT]
return ''
def summarize_failure(
exception: BaseException | None,
detail_text: str | None = None,
error_type: str | None = None,
) -> tuple[str, str]:
"""`(error_type, error_summary)` — the two technical fields every failure log carries.
Args:
exception: the failure, when the object itself is to hand.
detail_text: the failure as text, for a caller whose exception crossed an activity boundary.
error_type: the type name that travelled with such a text, when it is known.
Returns:
tuple: the exception type name and a single-line summary of what it said.
"""
if exception is not None:
cause = _root_cause(exception)
return _type_name(cause), _one_line(str(cause))
return error_type or _UNKNOWN_ERROR_TYPE, _one_line(detail_text or '')
def describe_failure_chain(exception: BaseException) -> tuple[str, str]:
"""`(error_type, detail)` for a failure that is about to lose its exception object.
The workflow is the one place that holds the whole chain — `ActivityError` wrapping the
`ApplicationError` that carries the activity's own type and message — and it cannot hand that
object to the channel, because the channel runs in an activity. So the chain is flattened here,
**root cause first**, one link per line: the first line is what actually broke, which is what
becomes `error_summary`, and the wrappers follow for whoever reads further.
`ApplicationError.type` is preferred over the Python class name because it is the only place the
activity's original exception type survives the boundary: `MlflowException` arrives as an
`ApplicationError` whose `type` says so.
"""
links: list[str] = []
candidate: BaseException | None = exception
for _ in range(_CHAIN_LIMIT):
if candidate is None:
break
kind = _type_name(candidate)
message = _one_line(str(candidate))
# `ApplicationError.__str__` already prefixes its own type, and `MlflowException:
# MlflowException: …` reads like two failures rather than one.
if not message:
links.append(kind)
elif message.startswith(f'{kind}:'):
links.append(message)
else:
links.append(f'{kind}: {message}')
candidate = candidate.__cause__
links.reverse()
root = links[0].split(':', 1)[0] if links else _UNKNOWN_ERROR_TYPE
return root, '\n'.join(links)
def _log_fields(
*,
failure: ImportFailure,
where: str,
import_run_id: Any,
workflow_id: str,
gate: int | None,
error_type: str,
error_summary: str,
verdict: str | None,
metadata: dict[str, Any] | None,
) -> dict[str, Any]:
"""The workflow metadata plus this failure's own fields, as `logfmt` keys.
Every key is prefixed `import_` or `error_` so none of them can collide with the metadata the
rest of the runtime already logs (`model_id`, `model_name`, `schedule_name`, `workflow_name`) or
with the formatter's own (`at`, `msg`, `timestamp`). Values are single-line by construction: the
traceback stays in the message, where a newline costs nothing.
"""
fields: dict[str, Any] = {
**(metadata or {}),
'import_run_id': import_run_id,
'import_workflow_id': workflow_id,
'import_step': where,
'import_code': failure.code.value,
'error_type': error_type,
}
if error_summary:
fields['error_summary'] = error_summary
if gate is not None:
fields['import_gate'] = gate
if verdict:
fields['import_verdict_not_recorded'] = verdict
return fields
def failure_location(exception: BaseException | None) -> tuple[str | None, int | None]:
"""Read the step and gate an activity's failure carries, if it carried any.
Walks the `__cause__` chain because Temporal wraps an activity failure in an `ActivityError`.
Returns `(None, None)` when the failure did not say where it happened, and the caller then falls
back to the step it was attempting.
"""
seen = 0
candidate: BaseException | None = exception
while candidate is not None and seen < 10:
details = getattr(candidate, 'details', None)
if details and isinstance(details[0], dict) and 'step' in details[0]:
location = details[0]
gate = location.get('gate')
return location.get('step'), int(gate) if isinstance(gate, int) else None
candidate = candidate.__cause__
seen += 1
return None, None
class FailureSinks(Protocol):
"""What the channel needs from its caller: the repository's existing observability surface."""
def error(self, message: str, metadata: dict | None = None) -> None: ...
def get_core_labels(self, metadata: dict | None = None, operation_type: str = '-') -> dict: ...
async def emit_metric(
self,
metric_object: Any,
tags: dict[str, Any],
method: str = 'inc',
value: float | None = None,
) -> None: ...
async def send_notification_async(
self,
metadata: dict,
notification_id: str,
message: str,
block: str,
level: str = 'INFO',
attachment_content: str | None = None,
) -> None: ...
async def report_import_failure(
sinks: FailureSinks,
step: ImportStep | None,
exception: BaseException | None,
*,
import_run_id: Any,
workflow_id: str,
metadata: dict[str, Any] | None = None,
code: ImportErrorCode | None = None,
marker: str | None = None,
verdict: str | None = None,
detail_text: str | None = None,
error_type: str | None = None,
gate: int | None = None,
) -> ImportFailure:
"""Report one failure through every sink, exactly once.
Sinks, in order: a structured log carrying the ids, the step, the code **and** the exception
with its traceback (the only sink that gets the exception text); the import error metric, with
the code in `operation_type`; a notification through the existing handler. The exception itself
reaches Temporal by being raised, and the record is the fifth, optional sink — the terminal
write, when the row can be written at all.
Both log lines this produces — the technical one here and the user-facing sentence the
notification logs — carry the same `logfmt` fields, built once by `_log_fields`. That is what
makes the two greppable and joinable: `import_run_id=29` finds the pair, `import_code=` groups
them, and `error_type=` says what actually broke without reading a traceback. The message text is
kept as it was, ids and all, so a query written against it still matches.
Args:
sinks: the activity providing logger, metrics and notifications.
step: the pipeline step, or `None` for a failure that is not one (see `marker`).
exception: the failure.
import_run_id: the record's id, for correlation.
workflow_id: the identifier a user quotes to support.
metadata: workflow metadata for labels and the notification.
code: an explicit code, for the two failures that are not pipeline steps.
marker: `not_claimable` or `status_write` — what a step name would have been.
verdict: for a failed terminal write, the verdict it could not record, so the outcome
survives in text even when it cannot survive in the row.
detail_text: the failure as text, for the callers whose exception object is already gone —
anything that crossed an activity boundary. Used in place of a traceback.
error_type: the exception type name that travelled with such a text, when it is known.
gate: the gate that rejected the bundle, when one did — a field, never a sentence.
Returns:
ImportFailure: the classification that was reported.
"""
failure = classify_import_failure(step, exception)
if code is not None:
failure = ImportFailure(step=step, code=code, reason=failure.reason)
where = marker or (step.value if step else 'unknown')
if exception is not None:
detail = ''.join(traceback.format_exception(exception)).strip()
else:
detail = detail_text or 'no exception object available'
kind, summary = summarize_failure(exception, detail_text, error_type)
fields = _log_fields(
failure=failure,
where=where,
import_run_id=import_run_id,
workflow_id=workflow_id,
gate=gate,
error_type=kind,
error_summary=summary,
verdict=verdict,
metadata=metadata,
)
verdict_line = f' verdict_not_recorded={verdict}' if verdict else ''
sinks.error(
f'model import failed: step={where} code={failure.code.value} '
f'import_run_id={import_run_id} workflow_id={workflow_id}{verdict_line}\n{detail}',
fields,
)
labels = sinks.get_core_labels(metadata or {}, operation_type=failure.code.value)
await sinks.emit_metric(metric_object=metrics.MODEL_IMPORT_ERROR_COUNT, tags=labels)
await sinks.send_notification_async(
metadata=fields,
notification_id=failure.code.value,
message=failure.reason,
block=where,
level='ERROR',
)
return failure

348
laborious/utils/models/minio_dataframe_payload.py Normal file
View File

@@ -0,0 +1,348 @@
"""
MinIO-backed DataFrame payload for Temporal workflows.
Data is never stored as a pandas ``DataFrame`` field on the dataclass.
Instead, the DataFrame is only provided as an input to:
`from_dataframe` / `from_dataframe_to_dict`.
At build time, the DataFrame is evaluated for its serialized size; if it exceeds
the configured threshold, it is serialized to parquet bytes and uploaded to MinIO.
Otherwise, it is inlined as a Temporal-friendly ``dict``.
"""
import pickle
import re
from collections.abc import Hashable
from dataclasses import dataclass
from datetime import datetime
from io import BytesIO
from os import getenv
from typing import Any, Literal
from pandas import DataFrame, read_parquet
from sientia_do.observability.logger import Logger
from sientia_do.repository.minio_repository import MinioRepository
from sientia_do.temporal.constants import DATETIME_FORMAT_FILENAME, DATETIME_FORMAT_WITH_TZ, now
# Keys that are part of the serialized wire format (not arbitrary metadata).
_SERIALIZED_FIELD_KEYS = frozenset({'data', 'bucket', 'object_key', 'object_prefix', 'uri'})
_OBJECT_TIMESTAMP_PATTERN = re.compile(
r'-(?:initial|transform)-(\d{4}-\d{2}-\d{2}_\d{2}-\d{2}-\d{2})\.parquet$'
)
OFFLOAD_THRESHOLD_BYTES = int(
float(getenv('SIENTIA_MINIO_OFFLOAD_THRESHOLD_MEGABYTES', '1.5')) * 1024 * 1024
)
# Relative prefix used for storing offloaded prediction datasets in MinIO.
# It is also the root directory for retention cleanup listing.
PREDICTION_DATASETS_PREFIX = 'prediction_datasets'
OperationKind = Literal['initial', 'transform', 'predict']
def _build_object_key(
model_name: str, operation: OperationKind, timestamp: str
) -> tuple[str, str | None]:
"""
Build the MinIO object key and the directory prefix used for retention listing.
Args:
model_name: Registered model name used in the pipeline.
operation: Either initial (pre-transform load) or transform (post-MLFlow transform).
timestamp: Filename timestamp segment from DATETIME_FORMAT_FILENAME.
Return:
tuple[str, str | None]: Full object key and normalized prefix (or None if at bucket root).
"""
# Naming convention:
# - Directory is always `prediction_datasets/<model_name>`
# - Filename follows the retention-parsing pattern
basename = f'{model_name}-{operation}-{timestamp}.parquet'
model_dir = model_name.strip().strip('/')
prefix = f'{PREDICTION_DATASETS_PREFIX}/{model_dir}'
return f'{prefix}/{basename}', prefix
@dataclass
class MinioDataFramePayload:
"""
Serializable payload after a DataFrame was evaluated: inline tabular dict and/or MinIO keys.
Build from a live DataFrame only via `from_dataframe` / `from_dataframe_to_dict`.
Rehydrate from Temporal via `from_dict`. The DataFrame is not a field on this class.
"""
last_timestamp: str
status: dict[str, Any] | None = None
data: dict[Hashable, Any] | None = None
bucket: str | None = None
object_key: str | None = None
object_prefix: str | None = None
uri: str | None = None
@staticmethod
def _debug(
logger: Logger | None,
message: str,
metadata: dict[str, Any] | None = None,
) -> None:
"""
Emit debug logs only when logger is provided
Args:
- logger (Logger | None): Logger instance used for debug messages
- message (str): Message to be logged
- metadata (dict[str, Any] | None): Optional workflow metadata context
"""
if logger is None:
return
logger.custom_debug(message, metadata)
@classmethod
def from_dict(cls, raw: 'dict[str, Any] | MinioDataFramePayload') -> 'MinioDataFramePayload':
"""
Reconstruct a MinioDataFramePayload from a plain dict produced by Temporal serialization.
Temporal converts dataclass return values into plain dicts when crossing
workflow/activity boundaries. This method rebuilds the typed instance so
that methods like ``retrieve``, ``cleanup_prefix`` and ``has_data`` are
available on the receiving side.
If the argument is already a MinioDataFramePayload, it is returned as-is.
Args:
raw: Dict with keys matching the dataclass fields
(last_timestamp, status, data, bucket, object_key, object_prefix, uri),
or an existing MinioDataFramePayload instance.
Return:
MinioDataFramePayload: Reconstructed (or original) instance.
"""
if isinstance(raw, MinioDataFramePayload):
return raw
return cls(
last_timestamp=raw['last_timestamp'],
status=raw.get('status'),
data=raw.get('data'),
bucket=raw.get('bucket'),
object_key=raw.get('object_key'),
object_prefix=raw.get('object_prefix'),
uri=raw.get('uri'),
)
@staticmethod
def estimate_size_bytes(
df: DataFrame,
metadata: dict[str, Any] | None = None,
logger: Logger | None = None,
) -> int:
"""
Approximate serialized size of the DataFrame as the default-orient dict.
Args:
df: DataFrame whose tabular content size is estimated.
Return:
int: Estimated size in bytes (pickle of dict representation).
"""
try:
size = len(pickle.dumps(df.to_dict()))
except Exception:
size = len(pickle.dumps(df))
MinioDataFramePayload._debug(
logger,
f'DataFrame size: {size} bytes',
metadata,
)
return size
@staticmethod
def parse_object_timestamp(object_key: str) -> datetime | None:
"""
Parse the timestamp embedded in the object key basename (before .parquet).
Args:
object_key: S3/MinIO object key whose basename follows
``{model}-{initial|transform}-{DATETIME_FORMAT_FILENAME}.parquet``.
Return:
datetime | None: Parsed UTC-naive datetime from the key, or None if not matched.
"""
basename = object_key.rsplit('/', 1)[-1]
match = _OBJECT_TIMESTAMP_PATTERN.search(basename)
if not match:
return None
try:
return datetime.strptime(match.group(1), DATETIME_FORMAT_FILENAME)
except ValueError:
return None
def cleanup_prefix(self) -> str | None:
"""
Return True if cleanup is enabled for this payload.
"""
if self.object_key is not None and self.data is None:
return self.object_prefix
return None
def has_data(self) -> bool:
"""
Return True if the payload has some data internally or in MinIO.
"""
return (self.data is not None and self.data != {}) or self.object_key is not None
@classmethod
async def from_dataframe(
cls,
dataframe: DataFrame | None,
minio_repo: MinioRepository,
model_name: str,
operation: OperationKind,
status: dict[str, Any] | None = None,
workflow_metadata: dict | None = None,
last_timestamp: str | None = None,
logger: Logger | None = None,
) -> 'MinioDataFramePayload':
"""
Evaluate the DataFrame size, then either inline dict or upload parquet to MinIO.
The DataFrame is not stored on the returned instance.
Args:
dataframe: Tabular data to evaluate and persist (inline or MinIO).
metadata: Small metadata dict merged into the payload (e.g. success, message).
minio_repo: sientia_do MinioRepository (or compatible) with `upload_file()`.
workflow_metadata: Metadata passed to MinIO store for logging/metrics.
model_name: Registered model name used in the object basename.
operation: Either ``initial`` (query load) or ``transform`` (post-transform).
key_prefix: Backward-compatible parameter (currently ignored for object naming).
size_threshold_bytes: Byte limit before offload. When None, the module-level
environment-derived default is used.
Return:
MinioDataFramePayload: Instance with data and/or MinIO fields set.
"""
if dataframe is None or dataframe.empty:
cls._debug(
logger,
'MinioDataFramePayload.from_dataframe received empty dataframe, returning empty payload',
workflow_metadata,
)
return cls(
data=None, last_timestamp=now().strftime(DATETIME_FORMAT_WITH_TZ), status=status
)
if last_timestamp is None:
last_timestamp = max(dataframe['timestamp'].values.tolist())
dataframe_size = cls.estimate_size_bytes(dataframe, workflow_metadata, logger)
cls._debug(
logger,
(
f'MinioDataFramePayload.from_dataframe estimated size: {dataframe_size} bytes '
f'(threshold: {OFFLOAD_THRESHOLD_BYTES} bytes)'
),
workflow_metadata,
)
if dataframe_size <= OFFLOAD_THRESHOLD_BYTES:
cls._debug(
logger,
'MinioDataFramePayload.from_dataframe using inline payload',
workflow_metadata,
)
return cls(data=dataframe.to_dict(), last_timestamp=last_timestamp, status=status)
timestamp = now().strftime(DATETIME_FORMAT_FILENAME)
object_key, object_prefix = _build_object_key(model_name, operation, timestamp)
cls._debug(
logger,
(
'MinioDataFramePayload.from_dataframe offloading payload to MinIO '
f'with key {object_key}'
),
workflow_metadata,
)
# Upload using the relative object key. The upstream repository will
# prefix it internally under its MinIO namespace.
parquet_buffer = BytesIO()
dataframe.to_parquet(parquet_buffer, engine='pyarrow', index=True)
file_bytes = parquet_buffer.getvalue()
upload_result = await minio_repo.upload_file(
file_bytes=file_bytes,
relative_key=object_key,
metadata=workflow_metadata,
)
bucket = minio_repo.bucket
object_key_full = upload_result.get('minio_object_name', object_key)
uri = f's3://{bucket}/{object_key_full}' if bucket else None
cls._debug(
logger,
f'MinioDataFramePayload.from_dataframe upload completed: {uri}',
workflow_metadata,
)
return cls(
data=None,
bucket=bucket,
object_key=object_key_full,
object_prefix=object_prefix,
uri=uri,
last_timestamp=last_timestamp,
status=status,
)
async def retrieve(
self,
minio_repo: MinioRepository,
workflow_metadata: dict[str, Any] | None = None,
logger: Logger | None = None,
) -> DataFrame:
"""
Load parquet from MinIO when object_key is set and populate inline data.
Args:
minio_repo: sientia_do MinioRepository (or compatible) with download_file().
workflow_metadata: Metadata passed to MinIO read for logging/metrics.
Return:
dict[str, Any]: Flat dict with data filled (same keys as to_dict after load).
"""
if self.data is not None:
self._debug(
logger,
'MinioDataFramePayload.retrieve using inline payload data',
workflow_metadata,
)
return DataFrame(self.data)
if not self.has_data():
self._debug(
logger,
'MinioDataFramePayload.retrieve found no payload data, returning empty dataframe',
workflow_metadata,
)
return DataFrame()
self._debug(
logger,
f'MinioDataFramePayload.retrieve downloading object from MinIO: {self.object_key}',
workflow_metadata,
)
file_bytes = await minio_repo.download_file(
object_name=self.object_key, metadata=workflow_metadata
)
df = read_parquet(BytesIO(file_bytes))
self._debug(
logger,
f'MinioDataFramePayload.retrieve loaded dataframe from MinIO with shape {df.shape}',
workflow_metadata,
)
return df