server.lock.Lock

class documentation

class Lock(UaObject, ParentMixin[UaObject]):

Constructor: Lock(minimum_access_level, **kwargs)

Manages exclusive write/execute access rights to the parent UaObject and its children.

Locks are not hierarchical. A child (or children's child, etc.) with its own Lock will not be influenced by this lock.

OPC UA clients first need to request the lock by calling InitLock(). If granted, they now own the lock and get exclusive write and execution rights. (reading, browsing, etc. is always possible). Ideally, clients release the lock as soon as they are done by calling ExitLock(). Locks are automatically released when the client disconnects or is inactive for too long without calling RenewLock().

The request can also fail when the user does not have the required minimum access level for the lock, or the lock is already hold by another client. If it's hold by another client, the client can request the lock by calling BreakLock(). It will succeed only when the priority of the requesting client are larger than of the lock owner.

Based on OPC UA Locking model (https://reference.opcfoundation.org/specs/OPC-10000-100/7).

Method	`__init__`	Construct new `Lock` instance with given `minimum_access_level`.
Method	`access_allowed`	Check whether `user` is allowed to access this lock.
Async Method	`break_lock`	Request exclusive access to nodes protected by this `Lock`.
Async Method	`exit_lock`	Release exclusive access to nodes protected by this `Lock`.
Async Method	`init_lock`	Request exclusive access to nodes protected by this `Lock`.
Async Method	`renew_lock`	Renew the remaining lock time before the ownership is lost due to inactivity.
Async Method	`update_locking_status`	Update locking status based on `user` unconditionally.
Property	`locked`	Whether this `Lock` is locked.
Property	`locking_user`	`User` instance owning this `Lock` or `None` if unlocked.
Property	`max_inactive_time_ms`	Maximum time in milliseconds a lock owner is allowed to be inactive.
Async Method	`_check_inactivity_loop`	Undocumented
Async Method	`_get_definition`	Undocumented
Async Method	`_init`	Undocumented
Async Method	`_lifecycle_lock`	Undocumented
Async Method	`_on_session_closed`	Handle session closed events.
Async Method	`_shutdown`	Undocumented
Async Method	`_ua_break_lock`	Undocumented
Async Method	`_ua_exit_lock`	Undocumented
Async Method	`_ua_init_lock`	Undocumented
Async Method	`_ua_renew_lock`	Undocumented
Class Variable	`_ua_locked`	Undocumented
Class Variable	`_ua_locking_client`	Undocumented
Class Variable	`_ua_locking_user`	Undocumented
Class Variable	`_ua_remaining_lock_time_ms`	Undocumented
Instance Variable	`_lock`	Used to guarantee atomic updates.
Instance Variable	`_locking_time`	Monotonic timestamp when the lock was granted to a user. Used for inactivity timeout.
Instance Variable	`_locking_user`	Undocumented
Instance Variable	`_task_check_inactivity`	Undocumented

def __init__(self, *, minimum_access_level: int | None = None, **kwargs: dict[str, Any]): ¶

Construct new Lock instance with given minimum_access_level.

@override

def access_allowed(self, user: UserAuthorization, *, lock_required: bool | None = None) -> bool: ¶

Check whether user is allowed to access this lock.

async def break_lock(self, user: UserAuthorization, *, session: SessionProtocol | None = None) -> bool: ¶

Request exclusive access to nodes protected by this Lock.

Request will fail if the lock is already owned by a user with higher User.priority than user or if user lacks necessary access rights. Warning: Will steal lock ownership from its current owner!

Returns
`bool`	Whether `user` now owns this `Lock`.

async def exit_lock(self, user: UserAuthorization) -> bool: ¶

Release exclusive access to nodes protected by this Lock.

Will fail if user lacks necessary access rights or the lock is owned by another user.

Returns
`bool`	Whether the lock was released.

async def init_lock(self, user: UserAuthorization, *, session: SessionProtocol | None = None) -> bool: ¶

Request exclusive access to nodes protected by this Lock.

Request will fail if the lock is already owned or if user lacks necessary access rights.

Returns
`bool`	Whether `user` now owns this `Lock`.

async def renew_lock(self, user: UserAuthorization) -> bool: ¶

Renew the remaining lock time before the ownership is lost due to inactivity.

async def update_locking_status(self, user: UserAuthorization | None, *, session: SessionProtocol | None = None, reason: str = ''): ¶

Update locking status based on user unconditionally.

Parameters
user:`UserAuthorization \| None`	`User` that will be set as the locking user. `None` for clearing the releasing the lock.
session:`SessionProtocol \| None`	(Optional) session instance. Is unavailable for internal user(s).
reason:`str`	Short reason description. Only for logging purposes.

@property

locked: bool = ¶

Whether this Lock is locked.

@property

locking_user: UserAuthorization | None = ¶

User instance owning this Lock or None if unlocked.

@property

max_inactive_time_ms: int = ¶

Maximum time in milliseconds a lock owner is allowed to be inactive.

async def _check_inactivity_loop(self) -> Never: ¶

Undocumented

@override

async def _get_definition(self) -> UaObjectDefinition: ¶

Undocumented

async def _init(self): ¶

Undocumented

@lifecycle

async def _lifecycle_lock(self) -> AsyncGenerator[None]: ¶

Undocumented

async def _on_session_closed(self, session: SessionProtocol): ¶

Handle session closed events.

Used to check if the user of that session is owning a lock. If that is the case, it will be released.

async def _shutdown(self): ¶

Undocumented

@uamethod

async def _ua_break_lock(self, parent: ua.NodeId, session: SessionProtocol) -> int | ua.StatusCode: ¶

Undocumented

@uamethod

async def _ua_exit_lock(self, parent: ua.NodeId, session: SessionProtocol) -> int | ua.StatusCode: ¶

Undocumented

@uamethod

async def _ua_init_lock(self, parent: ua.NodeId, session: SessionProtocol, context: str | None = None) -> int | ua.StatusCode: ¶

Undocumented

@uamethod

async def _ua_renew_lock(self, parent: ua.NodeId, session: SessionProtocol) -> int | ua.StatusCode: ¶

Undocumented

_ua_locked: Node = ¶

Undocumented

_ua_locking_client: Node = ¶

Undocumented

_ua_locking_user: Node = ¶

Undocumented

_ua_remaining_lock_time_ms: Node = ¶

Undocumented

_lock: asyncio.Lock = ¶

Used to guarantee atomic updates.

_locking_time: float = ¶

Monotonic timestamp when the lock was granted to a user. Used for inactivity timeout.

_locking_user: UserAuthorization | None = ¶

Undocumented

_task_check_inactivity: Task[None] = ¶

Undocumented