Neomega

外观

scopes API ​

完整接口目录 · 资源 Scope用法 · 工具包下载

模块说明 ​

text
Authority-owned resource scopes, with one execution of author callbacks.

ScopeCondition ​

python
@dataclass(frozen=True)
class ScopeCondition

字段（包含继承字段）：

ScopeCondition.kind ​

python
kind: str

数据类字段；声明于 scopes.ScopeCondition。

ScopeCondition.arguments_json ​

python
arguments_json: str

数据类字段；声明于 scopes.ScopeCondition。

ScopeCondition.near ​

python
@classmethod
def near(cls, target, distance)

ScopeSpec ​

python
@dataclass(frozen=True)
class ScopeSpec

字段（包含继承字段）：

ScopeSpec.resources ​

python
resources: list[str]

数据类字段；声明于 scopes.ScopeSpec。

ScopeSpec.conditions ​

python
conditions: list[ScopeCondition] = field(default_factory=list)

数据类字段；声明于 scopes.ScopeSpec。

ScopeSpec.lease_ms ​

python
lease_ms: i64 = 30000

数据类字段；声明于 scopes.ScopeSpec。

Resources ​

python
class Resources

SDK 说明：

text
Purpose declarations; dependency closure is computed by the Host.

Resources.inventory_session ​

python
@staticmethod
def inventory_session()

Resources.position_control ​

python
@staticmethod
def position_control()

Resources.container_session ​

python
@staticmethod
def container_session(dimension: int, x: int, y: int, z: int)

ScopeHandle ​

python
@dataclass(frozen=True)
class ScopeHandle

字段（包含继承字段）：

ScopeHandle.id ​

python
id: str

数据类字段；声明于 scopes.ScopeHandle。

ScopeHandle.request_id ​

python
request_id: str

数据类字段；声明于 scopes.ScopeHandle。

ScopeHandle.installation ​

python
installation: str

数据类字段；声明于 scopes.ScopeHandle。

ScopeHandle.generation ​

python
generation: i64

数据类字段；声明于 scopes.ScopeHandle。

ScopeHandle.session_id ​

python
session_id: str

数据类字段；声明于 scopes.ScopeHandle。

ScopeHandle.epoch ​

python
epoch: str

数据类字段；声明于 scopes.ScopeHandle。

ScopeHandle.action_session ​

python
action_session: str

数据类字段；声明于 scopes.ScopeHandle。

ScopeHandle.state ​

python
state: str

数据类字段；声明于 scopes.ScopeHandle。

ScopeHandle.resources ​

python
resources: list[str]

数据类字段；声明于 scopes.ScopeHandle。

ScopeHandle.expires_at ​

python
expires_at: str

数据类字段；声明于 scopes.ScopeHandle。

ScopeHandle.max_expires_at ​

python
max_expires_at: str

数据类字段；声明于 scopes.ScopeHandle。

ScopeHandle.operation_ids ​

python
operation_ids: list[str]

数据类字段；声明于 scopes.ScopeHandle。

ScopeHandle.pending_operation_ids ​

python
pending_operation_ids: list[str]

数据类字段；声明于 scopes.ScopeHandle。

ScopeHandle.frontier ​

python
frontier: i64

数据类字段；声明于 scopes.ScopeHandle。

ScopeHandle.pending ​

python
pending: i64

数据类字段；声明于 scopes.ScopeHandle。

ScopeError ​

python
class ScopeError(BusError)

SDK 说明：

text
Retained IDs and independent business/cleanup failures for reconciliation.

ScopeError.__init__ ​

python
def __init__(
    self,
    code,
    *,
    request_id = None,
    handle = None,
    business_error = None,
    cleanup_error = None,
)

ScopeError.request_id ​

python
self.request_id

公开实例属性，在 __init__ 中设置。

ScopeError.handle ​

python
self.handle

公开实例属性，在 __init__ 中设置。

ScopeError.business_error ​

python
self.business_error

公开实例属性，在 __init__ 中设置。

ScopeError.cleanup_error ​

python
self.cleanup_error

公开实例属性，在 __init__ 中设置。

ScopeResult ​

python
@dataclass(frozen=True)
class ScopeResult

字段（包含继承字段）：

ScopeResult.value ​

python
value: object

数据类字段；声明于 scopes.ScopeResult。

ScopeResult.handle ​

python
handle: ScopeHandle | None

数据类字段；声明于 scopes.ScopeResult。

ScopeResult.business_error ​

python
business_error: BaseException | None = None

数据类字段；声明于 scopes.ScopeResult。

ScopeResult.cleanup_error ​

python
cleanup_error: BaseException | None = None

数据类字段；声明于 scopes.ScopeResult。

Scopes ​

python
class Scopes

Scopes.__init__ ​

python
def __init__(self, bus, *, session_id=None)

Scopes.remote ​

python
@classmethod
async def remote(cls, bus)

SDK 说明：

text
Open one trusted EP session; call aclose when its transport ends.

Scopes.aclose ​

python
async def aclose(self)

Scopes.__aenter__ ​

python
async def __aenter__(self)

Scopes.__aexit__ ​

python
async def __aexit__(self, kind, business, traceback)

Scopes.open ​

python
def open(self, spec, *, request_id=None, timeout=30)

SDK 说明：

text
Use ``async with ctx.scopes.open(spec) as scope`` for automatic cleanup.

Scopes.acquire ​

python
async def acquire(self, spec, *, request_id=None, timeout=30)

Scopes.seal ​

python
def seal(self)

Scopes.join ​

python
async def join(self)

Scopes.activate ​

python
def activate(self)

Scopes.lookup ​

python
async def lookup(self, request_id)

Scopes.read ​

python
async def read(self, scope_id, *, timeout=5)

Scopes.run ​

python
async def run(self, spec, callback, *, request_id=None, timeout=30)

Scopes.bus ​

python
self.bus

公开实例属性，在 __init__ 中设置。

Scopes.session_id ​

python
self.session_id

公开实例属性，在 __init__、remote 中设置。

ResourceScope ​

python
class ResourceScope

ResourceScope.__init__ ​

python
def __init__(self, scopes, handle, *, renew=True)

ResourceScope.id ​

python
@property
def id(self)

ResourceScope.context ​

python
@contextmanager
def context(self)

SDK 说明：

text
Propagate scope and existing cause to all normal SDK child calls.

ResourceScope.__aenter__ ​

python
async def __aenter__(self)

ResourceScope.__aexit__ ​

python
async def __aexit__(self, kind, business, traceback)

ResourceScope.evidence ​

python
async def evidence(self, *resources)

ResourceScope.close ​

python
async def close(self, *, timeout=5)

ResourceScope.scopes ​

python
self.scopes

公开实例属性，在 __init__ 中设置。

ResourceScope.handle ​

python
self.handle

公开实例属性，在 __init__、_renew、_close 中设置。

Neomega · 使用、开发与参考文档