Skip to content

Row context

uptimely.row_context defines the runtime row-window values passed to analytical functions. Python declarations for requesting those windows live in uptimely.spec.python.row_context.

row_context

RowContext is also re-exported from uptimely.spec.

Portable observation-context requirements shared by spec and plans.

UNBOUNDED module-attribute

UNBOUNDED: Literal['unbounded'] = 'unbounded'

CountExtent module-attribute

CountExtent = typing.Annotated[int, pydantic.Field(strict=True, ge=0)] | typing.Literal['unbounded']

TimeExtent module-attribute

TimeExtent = str

ContextExtent

Bases: BaseModel

Context in one direction, expressed using a single measurement.

model_config class-attribute instance-attribute

model_config = pydantic.ConfigDict(extra='forbid', frozen=True)

time class-attribute instance-attribute

time: TimeExtent | None = None

steps class-attribute instance-attribute

steps: CountExtent | None = None

rows class-attribute instance-attribute

rows: CountExtent | None = None

validate_time classmethod

validate_time(value: str | None) -> str | None

Require a fixed duration or the explicit unbounded marker.

reject_null_extents classmethod

reject_null_extents(value: object) -> object

Distinguish omitted measurements from invalid explicit nulls.

validate_measurement

validate_measurement() -> ContextExtent

Require at most one measurement within a directional extent.

merge

merge(other: ContextExtent) -> ContextExtent

Take the maximum extent, requiring compatible measurements.

RowContext

Bases: BaseModel

Required input observations, interpreted using the dataset's logical order.

Empty directions mean current-row-only input. An absent RowContext means unknown requirements, not zero context. Unbounded directions require all observations in that direction; two unbounded directions need no ordering.

model_config class-attribute instance-attribute

model_config = pydantic.ConfigDict(extra='forbid', frozen=True)

dimension class-attribute instance-attribute

dimension: str | None = pydantic.Field(default=None, pattern=f'^{re.escape(contracts.DIMENSION_PREFIX)}[^.]+$')

before class-attribute instance-attribute

before: ContextExtent | Literal['unbounded'] = pydantic.Field(default_factory=ContextExtent)

after class-attribute instance-attribute

after: ContextExtent | Literal['unbounded'] = pydantic.Field(default_factory=ContextExtent)

requires_order property

requires_order: bool

Return whether the context distinguishes preceding from following rows.

validate_extent

validate_extent() -> RowContext

Require one measurement across both directional requirements.

to_dict

to_dict() -> dict[str, object]

Return JSON-compatible metadata without unspecified measurements.

merge

merge(other: RowContext) -> RowContext

Combine requirements for the same input scope without adding lags.

duration_nanoseconds

duration_nanoseconds(value: str) -> int

Parse a fixed duration such as 1h30m without floating-point rounding.

Parameters:

Name Type Description Default
value str

Nonnegative duration using distinct lowercase unit suffixes. Days and weeks mean exactly 24 hours and seven days.

required

Returns:

Type Description
int

The exact duration in nanoseconds.

Raises:

Type Description
ValueError

If syntax is invalid or precision is finer than a nanosecond.