MCPcopy Create free account
hub / github.com/sqlalchemy/sqlalchemy / ForUpdateArg

Class ForUpdateArg

lib/sqlalchemy/sql/selectable.py:3256–3323  ·  view source on GitHub ↗

Source from the content-addressed store, hash-verified

3254
3255
3256class ForUpdateArg(ClauseElement):
3257 _traverse_internals: _TraverseInternalsType = [
3258 ("of", InternalTraversal.dp_clauseelement_list),
3259 ("nowait", InternalTraversal.dp_boolean),
3260 ("read", InternalTraversal.dp_boolean),
3261 ("skip_locked", InternalTraversal.dp_boolean),
3262 ("key_share", InternalTraversal.dp_boolean),
3263 ]
3264
3265 of: Optional[Sequence[ClauseElement]]
3266 nowait: bool
3267 read: bool
3268 skip_locked: bool
3269
3270 @classmethod
3271 def _from_argument(
3272 cls, with_for_update: ForUpdateParameter
3273 ) -> Optional[ForUpdateArg]:
3274 if isinstance(with_for_update, ForUpdateArg):
3275 return with_for_update
3276 elif with_for_update in (None, False):
3277 return None
3278 elif with_for_update is True:
3279 return ForUpdateArg()
3280 else:
3281 return ForUpdateArg(**cast("Dict[str, Any]", with_for_update))
3282
3283 def __eq__(self, other: Any) -> bool:
3284 return (
3285 isinstance(other, ForUpdateArg)
3286 and other.nowait == self.nowait
3287 and other.read == self.read
3288 and other.skip_locked == self.skip_locked
3289 and other.key_share == self.key_share
3290 and other.of is self.of
3291 )
3292
3293 def __ne__(self, other: Any) -> bool:
3294 return not self.__eq__(other)
3295
3296 def __hash__(self) -> int:
3297 return id(self)
3298
3299 def __init__(
3300 self,
3301 *,
3302 nowait: bool = False,
3303 read: bool = False,
3304 of: Optional[_ForUpdateOfArgument] = None,
3305 skip_locked: bool = False,
3306 key_share: bool = False,
3307 ):
3308 """Represents arguments specified to
3309 :meth:`_expression.Select.for_update`.
3310
3311 """
3312
3313 self.nowait = nowait

Callers 4

_from_argumentMethod · 0.85
with_for_updateMethod · 0.85
with_for_updateMethod · 0.85

Calls

no outgoing calls

Tested by 1