Differential D8385 Diff 41667 python_modules/dagster/dagster/core/storage/captured_log_manager.py

Changeset View

Standalone View

python_modules/dagster/dagster/core/storage/captured_log_manager.py

This file was added.

				from abc import ABC, abstractmethod
				from contextlib import contextmanager
				from typing import NamedTuple, Optional

				from dagster.core.instance import MayHaveInstanceWeakref


				class CapturedLogData(
				NamedTuple(
				"_CapturedLogData",
				[("chunk", Optional[bytes]), ("cursor", Optional[int])],
				)
				):
				"""
				Object representing captured log data, either a partial chunk of the log data or the full
				capture. Contains the raw bytes and optionally the cursor offset for the partial chunk.
				"""

				def __new__(cls, chunk: Optional[bytes] = None, cursor: Optional[int] = None):
				alangenfeldUnsubmitted Not Done Inline Actions maybe "chunk" or "slice" to indicate its a piece of the full logs alangenfeld: maybe "chunk" or "slice" to indicate its a piece of the full logs
				return super(CapturedLogData, cls).__new__(cls, chunk, cursor)


				class CapturedLogMetadata(
				NamedTuple(
				"_CapturedLogMetadata",
				[("location", Optional[str]), ("download_url", Optional[str])],
				)
				):
				"""
				alangenfeldUnsubmitted Not Done Inline Actions what exactly is location? alangenfeld: what exactly is location?
				Object representing metadata info for the captured log data, containing a display string for
				the location of the log data and a URL for direct download of the captured log data.
				"""

				def __new__(cls, location: Optional[str] = None, download_url: Optional[str] = None):
				return super(CapturedLogMetadata, cls).__new__(cls, location, download_url)

				alangenfeldUnsubmitted Not Done Inline Actions from the compute steps of pipeline solids this should be changed right? alangenfeld: > from the compute steps of pipeline solids this should be changed right?

				class CapturedLogManager(ABC, MayHaveInstanceWeakref):
				"""Abstract base class for capturing the unstructured compute logs (stdout/stderr) in the
				current process, stored / retrieved with a provided log_key and namespace."""

				@abstractmethod
				@contextmanager
				def capture_logs(self, log_key: str, namespace: Optional[str] = None):
				pass

				@abstractmethod
				def is_capture_complete(self, log_key: str, namespace: Optional[str] = None):
				pass

				@abstractmethod
				def get_stdout(
				self,
				log_key: str,
				namespace: Optional[str] = None,
				cursor: str = None,
				max_bytes: int = None,
				) -> CapturedLogData:
				pass

				@abstractmethod
				def get_stderr(
				self,
				log_key: str,
				namespace: Optional[str] = None,
				alangenfeldUnsubmitted Not Done Inline Actions do we anticipate this ever working except when the file is local ? should the names communicate as much? alangenfeld: do we anticipate this ever working except when the file is local ? should the names communicate…
				cursor: str = None,
				max_bytes: int = None,
				) -> CapturedLogData:
				pass

				@abstractmethod
				def get_stdout_metadata(
				self, log_key: str, namespace: Optional[str] = None
				) -> CapturedLogMetadata:
				pass

				@abstractmethod
				alangenfeldUnsubmitted Not Done Inline Actions is there a better name than `_metadata` ? alangenfeld: is there a better name than `_metadata` ?
				def get_stderr_metadata(
				self, log_key: str, namespace: Optional[str] = None
				) -> CapturedLogMetadata:
				alangenfeldUnsubmitted Not Done Inline Actions what is this exactly? alangenfeld: what is this exactly?
				pass

				def should_capture_run_by_step(self) -> bool:
				alangenfeldUnsubmitted Not Done Inline Actions more docs here could be useful alangenfeld: more docs here could be useful
				return False