from dataclasses import dataclass, field
from time import time
from ..private.process.lambda_process import SharedContext
[docs]
@dataclass(slots=True, kw_only=True)
class LambdaContext[UserCtx]:
"""Lambda context"""
userCtx: UserCtx | None = None
lambdaStartTime: float = field(default_factory=time)
jobConfig: dict | None = None
@property
def isServiceLaunch(self):
"""Check whether it is job service launch"""
return self.jobConfig is not None and (self.jobConfig.get("service_launch", 0) == 1)
@dataclass(slots=True, kw_only=True)
class _SharedCounters:
_sharedContext: SharedContext | None
@property
def success(self):
return self._sharedContext.success.value
@property
def failed(self):
return self._sharedContext.failed.value
@property
def inProgress(self):
return self._sharedContext.inProgress.value
[docs]
@dataclass(slots=True)
class CurrentRequestsCount(_SharedCounters):
@property
def all(self):
return self._sharedContext.all.value
[docs]
@dataclass(slots=True, frozen=True)
class PreviousRequestsCount:
all: int
success: int
failed: int
inProgress: int
@classmethod
def getEmpty(cls):
return cls(0, 0, 0, 0)
[docs]
@dataclass(slots=True, kw_only=True)
class LambdaRequestContext(LambdaContext):
previousRequestsCount: PreviousRequestsCount | None = None
requestsCount: CurrentRequestsCount
def getCurrentData(self):
return {
"requests_count": {
"all": self.requestsCount.all,
"success": self.requestsCount.success,
"failed": self.requestsCount.failed,
"in_progress": self.requestsCount.inProgress,
}
}
[docs]
@dataclass
class CurrentSplitCount(_SharedCounters):
@property
def all(self):
return self._sharedContext.all.value
[docs]
@dataclass
class CurrentSubtaskCount(_SharedCounters):
@property
def all(self):
return self._sharedContext.all.value
[docs]
@dataclass(slots=True, frozen=True)
class PreviousSplitCount:
all: int
success: int
failed: int
inProgress: int
@classmethod
def getEmpty(cls):
return cls(0, 0, 0, 0)
[docs]
@dataclass(slots=True, frozen=True)
class PreviousSubtaskCount:
all: int
success: int
failed: int
inProgress: int
@classmethod
def getEmpty(cls):
return cls(0, 0, 0, 0)
[docs]
@dataclass(slots=True, kw_only=True)
class LambdaTaskContext(LambdaContext):
previousSplit: PreviousSplitCount | None = None
previousSubtask: PreviousSubtaskCount | None = None
split: CurrentSplitCount
subtask: CurrentSubtaskCount
def getCurrentData(self):
return {
"split": {
"all": self.split.all,
"success": self.split.success,
"failed": self.split.failed,
"in_progress": self.split.inProgress,
},
"subtask": {
"all": self.subtask.all,
"success": self.subtask.success,
"failed": self.subtask.failed,
"in_progress": self.subtask.inProgress,
},
}
[docs]
@dataclass
class CurrentStreamCount(_SharedCounters):
@property
def count(self):
return self._sharedContext.count.value
[docs]
@dataclass(slots=True, frozen=True)
class PreviousStreamCount:
count: int
success: int
failed: int
inProgress: int
@classmethod
def getEmpty(cls):
return cls(0, 0, 0, 0)
[docs]
@dataclass(slots=True, kw_only=True)
class LambdaAgentContext(LambdaContext):
previousStreamCount: PreviousStreamCount | None = None
streamCount: CurrentStreamCount
def getCurrentData(self):
return {
"stream_count": {
"count": self.streamCount.count,
"success": self.streamCount.success,
"failed": self.streamCount.failed,
"in_progress": self.streamCount.inProgress,
},
}