From 05770fa40170ef80ca9002da43f5665121d755c0 Mon Sep 17 00:00:00 2001 From: zhao vistart Date: Thu, 9 Apr 2026 00:02:13 +0800 Subject: [PATCH 1/4] feat(postgres): add row-level lock enhancement and advisory lock support Row-level lock enhancement: - FOR SHARE: Shared lock, allows reads but blocks updates (PG 9.0+) - FOR NO KEY UPDATE: Weaker exclusive lock (PG 9.0+) - FOR KEY SHARE: Weakest shared lock (PG 9.3+) - Add LockStrength enum in core query_parts - Extend ForUpdateClause to support different lock strengths - Add capability detection methods in PostgreSQLDialect Advisory lock support: - Session-level locks: pg_advisory_lock, pg_advisory_lock_shared - Transaction-level locks: pg_advisory_xact_lock, pg_advisory_xact_lock_shared - Non-blocking try variants: pg_try_advisory_lock, pg_try_advisory_lock_shared - Unlock functions: pg_advisory_unlock, pg_advisory_unlock_all - Context manager support for automatic lock release - Support for both single-key and two-key variants Implementation: - Add PostgreSQLAdvisoryLockSupport protocol - Add AdvisoryLockMixin for backend methods - Add expression classes in core advisory module - Implement format_for_update_clause with lock strength - Add version-aware capability detection Tests: - test_row_level_locks.py: Row-level lock strength tests - test_advisory_lock.py: Advisory lock functionality tests - Both sync and async versions covered Official Documentation: - Row-level locks: https://www.postgresql.org/docs/current/explicit-locking.html - Advisory locks: https://www.postgresql.org/docs/current/functions-admin.html#FUNCTIONS-ADVISORY-LOCKS --- .gitignore | 3 + changelog.d/row_level_lock.added.md | 1 + .../impl/postgres/backend/async_backend.py | 153 +++++++++++- .../backend/impl/postgres/backend/sync.py | 146 +++++++++++- .../backend/impl/postgres/dialect.py | 44 ++++ .../backend/impl/postgres/mixins/__init__.py | 2 + .../impl/postgres/mixins/dml/advisory_lock.py | 166 +++++++++++++ .../impl/postgres/protocols/__init__.py | 2 + .../postgres/protocols/dml/advisory_lock.py | 96 ++++++++ .../feature/backend/test_advisory_lock.py | 212 +++++++++++++++++ .../feature/backend/test_row_level_locks.py | 219 ++++++++++++++++++ 11 files changed, 1039 insertions(+), 5 deletions(-) create mode 100644 changelog.d/row_level_lock.added.md create mode 100644 src/rhosocial/activerecord/backend/impl/postgres/mixins/dml/advisory_lock.py create mode 100644 src/rhosocial/activerecord/backend/impl/postgres/protocols/dml/advisory_lock.py create mode 100644 tests/rhosocial/activerecord_postgres_test/feature/backend/test_advisory_lock.py create mode 100644 tests/rhosocial/activerecord_postgres_test/feature/backend/test_row_level_locks.py diff --git a/.gitignore b/.gitignore index 82e1c780..958a51d0 100644 --- a/.gitignore +++ b/.gitignore @@ -175,3 +175,6 @@ cython_debug/ *.crt .env*coverage*.json htmlcov/ + +# Claude Code plans (local only) +.claude/plans/ diff --git a/changelog.d/row_level_lock.added.md b/changelog.d/row_level_lock.added.md new file mode 100644 index 00000000..fac5e03b --- /dev/null +++ b/changelog.d/row_level_lock.added.md @@ -0,0 +1 @@ +Added PostgreSQL row-level lock strength support: FOR SHARE, FOR NO KEY UPDATE, and FOR KEY SHARE. These lock strengths provide finer control over row-level locking in PostgreSQL 9.0+ (KEY SHARE requires 9.3+). \ No newline at end of file diff --git a/src/rhosocial/activerecord/backend/impl/postgres/backend/async_backend.py b/src/rhosocial/activerecord/backend/impl/postgres/backend/async_backend.py index 2daa393e..1798f42b 100644 --- a/src/rhosocial/activerecord/backend/impl/postgres/backend/async_backend.py +++ b/src/rhosocial/activerecord/backend/impl/postgres/backend/async_backend.py @@ -61,7 +61,12 @@ from ..transaction import AsyncPostgresTransactionManager -class AsyncPostgresBackend(AsyncExplainBackendMixin, IntrospectorBackendMixin, PostgresBackendMixin, AsyncStorageBackend): +class AsyncPostgresBackend( + AsyncExplainBackendMixin, + IntrospectorBackendMixin, + PostgresBackendMixin, + AsyncStorageBackend, +): """Asynchronous PostgreSQL-specific backend implementation.""" def __init__(self, **kwargs): @@ -663,7 +668,15 @@ async def _reconnect(self) -> bool: self.log(logging.ERROR, f"Failed to reconnect: {str(e)}") return False - async def execute(self, sql: str, params: Optional[Tuple] = None, *, options=None, max_retries: int = 2, **kwargs) -> QueryResult: + async def execute( + self, + sql: str, + params: Optional[Tuple] = None, + *, + options=None, + max_retries: int = 2, + **kwargs, + ) -> QueryResult: """ Execute a SQL statement with optional parameters asynchronously. @@ -725,7 +738,7 @@ async def execute(self, sql: str, params: Optional[Tuple] = None, *, options=Non if await self._reconnect(): continue # If reconnection fails, break and raise the error - self.log(logging.ERROR, f"Failed to reconnect after connection error") + self.log(logging.ERROR, "Failed to reconnect after connection error") break raise @@ -839,5 +852,139 @@ def _parse_explain_result(self, raw_rows, sql, duration): rows = [PostgresExplainPlanLine(line=r.get("QUERY PLAN", "")) for r in raw_rows] return PostgresExplainResult(raw_rows=raw_rows, sql=sql, duration=duration, rows=rows) + # region Advisory Lock Methods + + async def execute_advisory_lock( + self, + key, + shared: bool = False, + session: bool = True + ) -> None: + """ + Acquire an advisory lock asynchronously. + + This method blocks until the lock is acquired. + + Args: + key: Lock key - either a single 64-bit integer or tuple of two 32-bit integers + shared: If True, acquire shared lock; otherwise exclusive lock + session: If True (default), session-level lock; otherwise transaction-level + + Raises: + DatabaseError: If the lock operation fails + """ + from rhosocial.activerecord.backend.expression.advisory import AdvisoryLockExpression + + expr = AdvisoryLockExpression(self.dialect, key=key, shared=shared, session=session) + sql, params = expr.to_sql() + await self.execute(sql, params) + + async def execute_advisory_unlock(self, key, shared: bool = False) -> bool: + """ + Release an advisory lock asynchronously. + + Args: + key: Lock key - either a single 64-bit integer or tuple of two 32-bit integers + shared: If True, release shared lock; otherwise exclusive lock + + Returns: + True if the lock was released, False if it was not held + """ + from rhosocial.activerecord.backend.expression.advisory import AdvisoryUnlockExpression + + expr = AdvisoryUnlockExpression(self.dialect, key=key, shared=shared) + sql, params = expr.to_sql() + result = await self.execute(sql, params) + # pg_advisory_unlock returns boolean in first column + if result.data and len(result.data) > 0: + row = result.data[0] + # Result is a dict with column name as key + return bool(list(row.values())[0]) + return False + + async def execute_advisory_unlock_all(self) -> None: + """ + Release all advisory locks held by the current session asynchronously. + """ + from rhosocial.activerecord.backend.expression.advisory import AdvisoryUnlockAllExpression + + expr = AdvisoryUnlockAllExpression(self.dialect) + sql, params = expr.to_sql() + await self.execute(sql, params) + + async def try_advisory_lock( + self, + key, + shared: bool = False, + session: bool = True + ) -> bool: + """ + Non-blocking advisory lock acquisition asynchronously. + + Args: + key: Lock key - either a single 64-bit integer or tuple of two 32-bit integers + shared: If True, acquire shared lock; otherwise exclusive lock + session: If True (default), session-level lock; otherwise transaction-level + + Returns: + True if the lock was acquired, False if it was not available + """ + from rhosocial.activerecord.backend.expression.advisory import TryAdvisoryLockExpression + + expr = TryAdvisoryLockExpression(self.dialect, key=key, shared=shared, session=session) + sql, params = expr.to_sql() + result = await self.execute(sql, params) + # pg_try_advisory_lock returns boolean in first column + if result.data and len(result.data) > 0: + row = result.data[0] + # Result is a dict with column name as key + return bool(list(row.values())[0]) + return False + + def advisory_lock( + self, + key, + shared: bool = False, + session: bool = True + ): + """ + Async context manager for advisory locks. + + Automatically acquires the lock on entry and releases it on exit. + + Args: + key: Lock key - either a single 64-bit integer or tuple of two 32-bit integers + shared: If True, acquire shared lock; otherwise exclusive lock + session: If True (default), session-level lock; otherwise transaction-level + + Returns: + Async context manager that holds the lock + + Example: + async with backend.advisory_lock(key=12345): + # Critical section - lock is held + await process_data() + # Lock automatically released + + # Shared lock for read-only access + async with backend.advisory_lock(key=12345, shared=True): + await read_data() + """ + from contextlib import asynccontextmanager + + @asynccontextmanager + async def _advisory_lock_context(): + await self.execute_advisory_lock(key=key, shared=shared, session=session) + try: + yield + finally: + # Only release session-level locks; transaction-level are auto-released + if session: + await self.execute_advisory_unlock(key=key) + + return _advisory_lock_context() + + # endregion + __all__ = ["AsyncPostgresBackend"] diff --git a/src/rhosocial/activerecord/backend/impl/postgres/backend/sync.py b/src/rhosocial/activerecord/backend/impl/postgres/backend/sync.py index 7aa5c2c8..d770c2ca 100644 --- a/src/rhosocial/activerecord/backend/impl/postgres/backend/sync.py +++ b/src/rhosocial/activerecord/backend/impl/postgres/backend/sync.py @@ -664,7 +664,15 @@ def _is_connection_error_message(self, error_msg: str) -> bool: return True return False - def execute(self, sql: str, params: Optional[Tuple] = None, *, options=None, max_retries: int = 2, **kwargs) -> QueryResult: + def execute( + self, + sql: str, + params: Optional[Tuple] = None, + *, + options=None, + max_retries: int = 2, + **kwargs, + ) -> QueryResult: """ Execute a SQL statement with optional parameters. @@ -726,7 +734,7 @@ def execute(self, sql: str, params: Optional[Tuple] = None, *, options=None, max if self._reconnect(): continue # If reconnection fails, break and raise the error - self.log(logging.ERROR, f"Failed to reconnect after connection error") + self.log(logging.ERROR, "Failed to reconnect after connection error") break raise @@ -840,5 +848,139 @@ def _parse_explain_result(self, raw_rows, sql, duration): rows = [PostgresExplainPlanLine(line=r.get("QUERY PLAN", "")) for r in raw_rows] return PostgresExplainResult(raw_rows=raw_rows, sql=sql, duration=duration, rows=rows) + # region Advisory Lock Methods + + def execute_advisory_lock( + self, + key, + shared: bool = False, + session: bool = True + ) -> None: + """ + Acquire an advisory lock. + + This method blocks until the lock is acquired. + + Args: + key: Lock key - either a single 64-bit integer or tuple of two 32-bit integers + shared: If True, acquire shared lock; otherwise exclusive lock + session: If True (default), session-level lock; otherwise transaction-level + + Raises: + DatabaseError: If the lock operation fails + """ + from rhosocial.activerecord.backend.expression.advisory import AdvisoryLockExpression + + expr = AdvisoryLockExpression(self.dialect, key=key, shared=shared, session=session) + sql, params = expr.to_sql() + self.execute(sql, params) + + def execute_advisory_unlock(self, key, shared: bool = False) -> bool: + """ + Release an advisory lock. + + Args: + key: Lock key - either a single 64-bit integer or tuple of two 32-bit integers + shared: If True, release shared lock; otherwise exclusive lock + + Returns: + True if the lock was released, False if it was not held + """ + from rhosocial.activerecord.backend.expression.advisory import AdvisoryUnlockExpression + + expr = AdvisoryUnlockExpression(self.dialect, key=key, shared=shared) + sql, params = expr.to_sql() + result = self.execute(sql, params) + # pg_advisory_unlock returns boolean in first column + if result.data and len(result.data) > 0: + row = result.data[0] + # Result is a dict with column name as key + return bool(list(row.values())[0]) + return False + + def execute_advisory_unlock_all(self) -> None: + """ + Release all advisory locks held by the current session. + """ + from rhosocial.activerecord.backend.expression.advisory import AdvisoryUnlockAllExpression + + expr = AdvisoryUnlockAllExpression(self.dialect) + sql, params = expr.to_sql() + self.execute(sql, params) + + def try_advisory_lock( + self, + key, + shared: bool = False, + session: bool = True + ) -> bool: + """ + Non-blocking advisory lock acquisition. + + Args: + key: Lock key - either a single 64-bit integer or tuple of two 32-bit integers + shared: If True, acquire shared lock; otherwise exclusive lock + session: If True (default), session-level lock; otherwise transaction-level + + Returns: + True if the lock was acquired, False if it was not available + """ + from rhosocial.activerecord.backend.expression.advisory import TryAdvisoryLockExpression + + expr = TryAdvisoryLockExpression(self.dialect, key=key, shared=shared, session=session) + sql, params = expr.to_sql() + result = self.execute(sql, params) + # pg_try_advisory_lock returns boolean in first column + if result.data and len(result.data) > 0: + row = result.data[0] + # Result is a dict with column name as key + return bool(list(row.values())[0]) + return False + + def advisory_lock( + self, + key, + shared: bool = False, + session: bool = True + ): + """ + Context manager for advisory locks. + + Automatically acquires the lock on entry and releases it on exit. + + Args: + key: Lock key - either a single 64-bit integer or tuple of two 32-bit integers + shared: If True, acquire shared lock; otherwise exclusive lock + session: If True (default), session-level lock; otherwise transaction-level + + Returns: + Context manager that holds the lock + + Example: + with backend.advisory_lock(key=12345): + # Critical section - lock is held + process_data() + # Lock automatically released + + # Shared lock for read-only access + with backend.advisory_lock(key=12345, shared=True): + read_data() + """ + from contextlib import contextmanager + + @contextmanager + def _advisory_lock_context(): + self.execute_advisory_lock(key=key, shared=shared, session=session) + try: + yield + finally: + # Only release session-level locks; transaction-level are auto-released + if session: + self.execute_advisory_unlock(key=key) + + return _advisory_lock_context() + + # endregion + __all__ = ["PostgresBackend"] diff --git a/src/rhosocial/activerecord/backend/impl/postgres/dialect.py b/src/rhosocial/activerecord/backend/impl/postgres/dialect.py index 37dfcb58..218fcaba 100644 --- a/src/rhosocial/activerecord/backend/impl/postgres/dialect.py +++ b/src/rhosocial/activerecord/backend/impl/postgres/dialect.py @@ -99,6 +99,7 @@ # DDL/DML operation mixins (new) PostgresExtendedStatisticsMixin, PostgresStoredProcedureMixin, + PostgresAdvisoryLockMixin, # Introspection capability mixin PostgresIntrospectionCapabilityMixin, ) @@ -137,6 +138,7 @@ PostgresParallelQuerySupport, PostgresStoredProcedureSupport, PostgresExtendedStatisticsSupport, + PostgresAdvisoryLockSupport, ) if TYPE_CHECKING: @@ -222,6 +224,7 @@ class PostgresDialect( # DDL/DML operation mixins (new) PostgresExtendedStatisticsMixin, PostgresStoredProcedureMixin, + PostgresAdvisoryLockMixin, # Protocol supports SetOperationSupport, TruncateSupport, @@ -286,6 +289,7 @@ class PostgresDialect( PostgresParallelQuerySupport, PostgresStoredProcedureSupport, PostgresExtendedStatisticsSupport, + PostgresAdvisoryLockSupport, ): """ PostgreSQL dialect implementation that adapts to the PostgreSQL version. @@ -461,6 +465,46 @@ def supports_for_update_skip_locked(self) -> bool: """Whether FOR UPDATE SKIP LOCKED is supported.""" return self.version >= (9, 5, 0) # Supported since 9.5 + def supports_for_share(self) -> bool: + """Whether FOR SHARE lock strength is supported (PostgreSQL 9.0+).""" + return self.version >= (9, 0, 0) + + def supports_for_no_key_update(self) -> bool: + """Whether FOR NO KEY UPDATE lock strength is supported (PostgreSQL 9.0+).""" + return self.version >= (9, 0, 0) + + def supports_for_key_share(self) -> bool: + """Whether FOR KEY SHARE lock strength is supported (PostgreSQL 9.3+).""" + return self.version >= (9, 3, 0) + + def supports_lock_strength(self, strength) -> bool: + """ + Check if a specific lock strength is supported. + + PostgreSQL lock strength support by version: + - FOR UPDATE: All versions + - FOR NO KEY UPDATE: PostgreSQL 9.0+ + - FOR SHARE: PostgreSQL 9.0+ + - FOR KEY SHARE: PostgreSQL 9.3+ + + Args: + strength: The LockStrength enum value to check + + Returns: + True if the lock strength is supported, False otherwise + """ + from rhosocial.activerecord.backend.expression.query_parts import LockStrength + + if strength == LockStrength.UPDATE: + return True # All PostgreSQL versions support FOR UPDATE + elif strength == LockStrength.NO_KEY_UPDATE: + return self.version >= (9, 0, 0) + elif strength == LockStrength.SHARE: + return self.version >= (9, 0, 0) + elif strength == LockStrength.KEY_SHARE: + return self.version >= (9, 3, 0) + return False + def supports_merge_statement(self) -> bool: """Whether MERGE statement is supported.""" return self.version >= (15, 0, 0) # MERGE added in PostgreSQL 15 diff --git a/src/rhosocial/activerecord/backend/impl/postgres/mixins/__init__.py b/src/rhosocial/activerecord/backend/impl/postgres/mixins/__init__.py index 470dcf16..fe2be7ba 100644 --- a/src/rhosocial/activerecord/backend/impl/postgres/mixins/__init__.py +++ b/src/rhosocial/activerecord/backend/impl/postgres/mixins/__init__.py @@ -26,6 +26,7 @@ from .dml.vacuum import PostgresVacuumMixin from .dml.stored_procedure import PostgresStoredProcedureMixin from .dml.extended_statistics import PostgresExtendedStatisticsMixin +from .dml.advisory_lock import PostgresAdvisoryLockMixin # Type mixins from .types.enum import EnumTypeMixin @@ -67,6 +68,7 @@ "PostgresVacuumMixin", "PostgresStoredProcedureMixin", "PostgresExtendedStatisticsMixin", + "PostgresAdvisoryLockMixin", # Type mixins "EnumTypeMixin", "TypesDataTypeMixin", diff --git a/src/rhosocial/activerecord/backend/impl/postgres/mixins/dml/advisory_lock.py b/src/rhosocial/activerecord/backend/impl/postgres/mixins/dml/advisory_lock.py new file mode 100644 index 00000000..ad493681 --- /dev/null +++ b/src/rhosocial/activerecord/backend/impl/postgres/mixins/dml/advisory_lock.py @@ -0,0 +1,166 @@ +# src/rhosocial/activerecord/backend/impl/postgres/mixins/dml/advisory_lock.py +""" +PostgreSQL advisory lock mixin implementation. + +Provides methods for working with PostgreSQL advisory locks, which are +application-level locks managed by the database server. +""" + +from typing import Tuple, TYPE_CHECKING + +if TYPE_CHECKING: + from rhosocial.activerecord.backend.expression.advisory import ( + AdvisoryLockExpression, + AdvisoryUnlockExpression, + AdvisoryUnlockAllExpression, + TryAdvisoryLockExpression, + ) + + +class PostgresAdvisoryLockMixin: + """ + Mixin providing PostgreSQL advisory lock support. + + Advisory locks are application-level locks that provide a way for + applications to coordinate activities across multiple sessions. + + The lock functions available in PostgreSQL: + - pg_advisory_lock(key) - acquire exclusive session-level lock + - pg_advisory_lock_shared(key) - acquire shared session-level lock + - pg_advisory_unlock(key) - release session-level lock + - pg_advisory_unlock_all() - release all session-level locks + - pg_try_advisory_lock(key) - non-blocking acquire exclusive lock + - pg_try_advisory_lock_shared(key) - non-blocking acquire shared lock + - pg_advisory_xact_lock(key) - acquire exclusive transaction-level lock + - pg_advisory_xact_lock_shared(key) - acquire shared transaction-level lock + - pg_try_advisory_xact_lock(key) - non-blocking acquire transaction-level lock + + Keys can be either: + - A single 64-bit integer (bigint) + - Two 32-bit integers (int, int) + """ + + def supports_advisory_lock(self) -> bool: + """ + Whether advisory locks are supported. + + Advisory locks are supported in PostgreSQL 9.0 and later. + + Returns: + True if advisory locks are supported + """ + return True # PostgreSQL 9.0+ supports advisory locks + + def format_advisory_lock(self, expr: "AdvisoryLockExpression") -> Tuple[str, tuple]: + """ + Format SQL for acquiring an advisory lock. + + Generates: + - SELECT pg_advisory_lock(key) for exclusive session lock + - SELECT pg_advisory_lock_shared(key) for shared session lock + - SELECT pg_advisory_xact_lock(key) for exclusive transaction lock + - SELECT pg_advisory_xact_lock_shared(key) for shared transaction lock + + Args: + expr: AdvisoryLockExpression with lock parameters + + Returns: + Tuple of (SQL string, parameters) + """ + # Build function name based on options + if expr.session: + # Session-level lock + if expr.shared: + func_name = "pg_advisory_lock_shared" + else: + func_name = "pg_advisory_lock" + else: + # Transaction-level lock + if expr.shared: + func_name = "pg_advisory_xact_lock_shared" + else: + func_name = "pg_advisory_xact_lock" + + # Handle key format + if isinstance(expr.key, tuple): + key1, key2 = expr.key + return f"SELECT {func_name}(%s, %s)", (key1, key2) + else: + return f"SELECT {func_name}(%s)", (expr.key,) + + def format_advisory_unlock(self, expr: "AdvisoryUnlockExpression") -> Tuple[str, tuple]: + """ + Format SQL for releasing an advisory lock. + + Generates: + - SELECT pg_advisory_unlock(key) for exclusive lock + - SELECT pg_advisory_unlock_shared(key) for shared lock + + Args: + expr: AdvisoryUnlockExpression with lock parameters + + Returns: + Tuple of (SQL string, parameters) + """ + # Build function name based on shared flag + # Note: Shared locks must use pg_advisory_unlock_shared + func_name = "pg_advisory_unlock_shared" if expr.shared else "pg_advisory_unlock" + + # Handle key format + if isinstance(expr.key, tuple): + key1, key2 = expr.key + return f"SELECT {func_name}(%s, %s)", (key1, key2) + else: + return f"SELECT {func_name}(%s)", (expr.key,) + + def format_advisory_unlock_all(self, expr: "AdvisoryUnlockAllExpression") -> Tuple[str, tuple]: + """ + Format SQL for releasing all advisory locks. + + Generates: + - SELECT pg_advisory_unlock_all() + + Args: + expr: AdvisoryUnlockAllExpression + + Returns: + Tuple of (SQL string, parameters) + """ + return "SELECT pg_advisory_unlock_all()", () + + def format_try_advisory_lock(self, expr: "TryAdvisoryLockExpression") -> Tuple[str, tuple]: + """ + Format SQL for non-blocking advisory lock acquisition. + + Generates: + - SELECT pg_try_advisory_lock(key) for exclusive session lock + - SELECT pg_try_advisory_lock_shared(key) for shared session lock + - SELECT pg_try_advisory_xact_lock(key) for exclusive transaction lock + - SELECT pg_try_advisory_xact_lock_shared(key) for shared transaction lock + + Args: + expr: TryAdvisoryLockExpression with lock parameters + + Returns: + Tuple of (SQL string, parameters) + """ + # Build function name based on options + if expr.session: + # Session-level lock + if expr.shared: + func_name = "pg_try_advisory_lock_shared" + else: + func_name = "pg_try_advisory_lock" + else: + # Transaction-level lock + if expr.shared: + func_name = "pg_try_advisory_xact_lock_shared" + else: + func_name = "pg_try_advisory_xact_lock" + + # Handle key format + if isinstance(expr.key, tuple): + key1, key2 = expr.key + return f"SELECT {func_name}(%s, %s)", (key1, key2) + else: + return f"SELECT {func_name}(%s)", (expr.key,) \ No newline at end of file diff --git a/src/rhosocial/activerecord/backend/impl/postgres/protocols/__init__.py b/src/rhosocial/activerecord/backend/impl/postgres/protocols/__init__.py index d659cf6b..fa6d1adf 100644 --- a/src/rhosocial/activerecord/backend/impl/postgres/protocols/__init__.py +++ b/src/rhosocial/activerecord/backend/impl/postgres/protocols/__init__.py @@ -25,6 +25,7 @@ from .dml.vacuum import PostgresVacuumSupport from .dml.stored_procedure import PostgresStoredProcedureSupport from .dml.extended_statistics import PostgresExtendedStatisticsSupport +from .dml.advisory_lock import PostgresAdvisoryLockSupport # Type protocols from .types.data_type import PostgresDataTypeSupport @@ -65,6 +66,7 @@ "PostgresVacuumSupport", "PostgresStoredProcedureSupport", "PostgresExtendedStatisticsSupport", + "PostgresAdvisoryLockSupport", # Type protocols "PostgresDataTypeSupport", "MultirangeSupport", diff --git a/src/rhosocial/activerecord/backend/impl/postgres/protocols/dml/advisory_lock.py b/src/rhosocial/activerecord/backend/impl/postgres/protocols/dml/advisory_lock.py new file mode 100644 index 00000000..95d1b2d9 --- /dev/null +++ b/src/rhosocial/activerecord/backend/impl/postgres/protocols/dml/advisory_lock.py @@ -0,0 +1,96 @@ +# src/rhosocial/activerecord/backend/impl/postgres/protocols/dml/advisory_lock.py +""" +PostgreSQL advisory lock protocol definitions. + +Advisory locks are application-level locks managed by the database server. +They provide a way for applications to coordinate activities across sessions +without relying on table-level or row-level locks. +""" + +from typing import Protocol, Tuple, Union, TYPE_CHECKING + +if TYPE_CHECKING: + from rhosocial.activerecord.backend.expression.advisory import ( + AdvisoryLockExpression, + AdvisoryUnlockExpression, + AdvisoryUnlockAllExpression, + TryAdvisoryLockExpression, + ) + + +class PostgresAdvisoryLockSupport(Protocol): + """ + Protocol for PostgreSQL advisory lock support. + + Advisory locks are application-level locks that provide a way for + applications to coordinate activities across multiple sessions. + + Features: + - Session-level locks: Released explicitly or when session ends + - Transaction-level locks: Released when transaction ends + - Exclusive locks: Only one session can hold + - Shared locks: Multiple sessions can hold simultaneously + - Non-blocking try variants: Return boolean instead of waiting + + PostgreSQL version support: + - All advisory lock functions: PostgreSQL 9.0+ + """ + + def supports_advisory_lock(self) -> bool: + """ + Whether advisory locks are supported. + + Advisory locks are supported in PostgreSQL 9.0 and later. + + Returns: + True if advisory locks are supported + """ + ... # pragma: no cover + + def format_advisory_lock(self, expr: "AdvisoryLockExpression") -> Tuple[str, tuple]: + """ + Format SQL for acquiring an advisory lock. + + Args: + expr: AdvisoryLockExpression with lock parameters + + Returns: + Tuple of (SQL string, parameters) + """ + ... # pragma: no cover + + def format_advisory_unlock(self, expr: "AdvisoryUnlockExpression") -> Tuple[str, tuple]: + """ + Format SQL for releasing an advisory lock. + + Args: + expr: AdvisoryUnlockExpression with lock parameters + + Returns: + Tuple of (SQL string, parameters) + """ + ... # pragma: no cover + + def format_advisory_unlock_all(self, expr: "AdvisoryUnlockAllExpression") -> Tuple[str, tuple]: + """ + Format SQL for releasing all advisory locks. + + Args: + expr: AdvisoryUnlockAllExpression + + Returns: + Tuple of (SQL string, parameters) + """ + ... # pragma: no cover + + def format_try_advisory_lock(self, expr: "TryAdvisoryLockExpression") -> Tuple[str, tuple]: + """ + Format SQL for non-blocking advisory lock acquisition. + + Args: + expr: TryAdvisoryLockExpression with lock parameters + + Returns: + Tuple of (SQL string, parameters) + """ + ... # pragma: no cover \ No newline at end of file diff --git a/tests/rhosocial/activerecord_postgres_test/feature/backend/test_advisory_lock.py b/tests/rhosocial/activerecord_postgres_test/feature/backend/test_advisory_lock.py new file mode 100644 index 00000000..d7924a7c --- /dev/null +++ b/tests/rhosocial/activerecord_postgres_test/feature/backend/test_advisory_lock.py @@ -0,0 +1,212 @@ +# tests/rhosocial/activerecord_postgres_test/feature/backend/test_advisory_lock.py +""" +PostgreSQL advisory lock tests. + +This module tests PostgreSQL's advisory lock functionality: +- Session-level locks +- Transaction-level locks +- Exclusive locks +- Shared locks +- Non-blocking try variants +- Context manager usage +""" + +from rhosocial.activerecord.backend.expression.advisory import ( + AdvisoryLockExpression, + AdvisoryUnlockExpression, + AdvisoryUnlockAllExpression, + TryAdvisoryLockExpression, +) + + +class TestAdvisoryLockExpression: + """Test advisory lock expression SQL generation.""" + + def test_advisory_lock_expression_basic(self, postgres_backend_single): + """Test basic advisory lock expression.""" + dialect = postgres_backend_single.dialect + + expr = AdvisoryLockExpression(dialect, key=12345) + sql, params = expr.to_sql() + assert sql == "SELECT pg_advisory_lock(%s)" + assert params == (12345,) + + def test_advisory_lock_expression_shared(self, postgres_backend_single): + """Test shared advisory lock expression.""" + dialect = postgres_backend_single.dialect + + expr = AdvisoryLockExpression(dialect, key=12345, shared=True) + sql, params = expr.to_sql() + assert sql == "SELECT pg_advisory_lock_shared(%s)" + assert params == (12345,) + + def test_advisory_lock_expression_transaction_level(self, postgres_backend_single): + """Test transaction-level advisory lock expression.""" + dialect = postgres_backend_single.dialect + + expr = AdvisoryLockExpression(dialect, key=12345, session=False) + sql, params = expr.to_sql() + assert sql == "SELECT pg_advisory_xact_lock(%s)" + assert params == (12345,) + + def test_advisory_lock_expression_two_keys(self, postgres_backend_single): + """Test advisory lock with two keys.""" + dialect = postgres_backend_single.dialect + + expr = AdvisoryLockExpression(dialect, key=(123, 456)) + sql, params = expr.to_sql() + assert sql == "SELECT pg_advisory_lock(%s, %s)" + assert params == (123, 456) + + def test_advisory_unlock_expression(self, postgres_backend_single): + """Test advisory unlock expression.""" + dialect = postgres_backend_single.dialect + + expr = AdvisoryUnlockExpression(dialect, key=12345) + sql, params = expr.to_sql() + assert sql == "SELECT pg_advisory_unlock(%s)" + assert params == (12345,) + + def test_advisory_unlock_all_expression(self, postgres_backend_single): + """Test advisory unlock all expression.""" + dialect = postgres_backend_single.dialect + + expr = AdvisoryUnlockAllExpression(dialect) + sql, params = expr.to_sql() + assert sql == "SELECT pg_advisory_unlock_all()" + assert params == () + + def test_try_advisory_lock_expression(self, postgres_backend_single): + """Test try advisory lock expression.""" + dialect = postgres_backend_single.dialect + + expr = TryAdvisoryLockExpression(dialect, key=12345) + sql, params = expr.to_sql() + assert sql == "SELECT pg_try_advisory_lock(%s)" + assert params == (12345,) + + def test_try_advisory_lock_expression_shared(self, postgres_backend_single): + """Test try shared advisory lock expression.""" + dialect = postgres_backend_single.dialect + + expr = TryAdvisoryLockExpression(dialect, key=12345, shared=True) + sql, params = expr.to_sql() + assert sql == "SELECT pg_try_advisory_lock_shared(%s)" + assert params == (12345,) + + +class TestAdvisoryLockBackend: + """Test advisory lock backend methods.""" + + def test_execute_advisory_lock(self, postgres_backend_single): + """Test acquiring and releasing advisory lock.""" + # Acquire lock + postgres_backend_single.execute_advisory_lock(key=12345) + # Release lock + result = postgres_backend_single.execute_advisory_unlock(key=12345) + assert result is True + + def test_execute_advisory_lock_two_keys(self, postgres_backend_single): + """Test advisory lock with two keys.""" + # Acquire lock + postgres_backend_single.execute_advisory_lock(key=(123, 456)) + # Release lock + result = postgres_backend_single.execute_advisory_unlock(key=(123, 456)) + assert result is True + + def test_execute_advisory_lock_shared(self, postgres_backend_single): + """Test shared advisory lock.""" + # Acquire shared lock + postgres_backend_single.execute_advisory_lock(key=12345, shared=True) + # Release shared lock (must use shared=True) + result = postgres_backend_single.execute_advisory_unlock(key=12345, shared=True) + assert result is True + + def test_try_advisory_lock_success(self, postgres_backend_single): + """Test try advisory lock when available.""" + result = postgres_backend_single.try_advisory_lock(key=12345) + assert result is True + # Clean up + postgres_backend_single.execute_advisory_unlock(key=12345) + + def test_try_advisory_lock_conflict(self, postgres_backend_single, postgres_control_backend): + """Test try advisory lock when already locked.""" + # Acquire lock in main backend + postgres_backend_single.execute_advisory_lock(key=99999) + + try: + # Try to acquire same lock in control backend (should fail) + result = postgres_control_backend.try_advisory_lock(key=99999) + assert result is False + finally: + # Clean up + postgres_backend_single.execute_advisory_unlock(key=99999) + + def test_advisory_unlock_all(self, postgres_backend_single): + """Test releasing all advisory locks.""" + # Acquire multiple locks + postgres_backend_single.execute_advisory_lock(key=11111) + postgres_backend_single.execute_advisory_lock(key=22222) + postgres_backend_single.execute_advisory_lock(key=33333) + + # Release all + postgres_backend_single.execute_advisory_unlock_all() + + # Verify locks are released + result = postgres_backend_single.try_advisory_lock(key=11111) + assert result is True + postgres_backend_single.execute_advisory_unlock(key=11111) + + def test_advisory_lock_context_manager(self, postgres_backend_single): + """Test advisory lock context manager.""" + with postgres_backend_single.advisory_lock(key=12345): + # Lock should be held + # Try to acquire from same session - should succeed + # But if we try from another session it would block + pass + # Lock should be released after context exit + result = postgres_backend_single.try_advisory_lock(key=12345) + assert result is True + postgres_backend_single.execute_advisory_unlock(key=12345) + + +class TestAsyncAdvisoryLockExpression: + """Async tests for advisory lock expression SQL generation.""" + + async def test_async_advisory_lock_expression_basic(self, async_postgres_backend_single): + """Test basic advisory lock expression (async).""" + dialect = async_postgres_backend_single.dialect + + expr = AdvisoryLockExpression(dialect, key=12345) + sql, params = expr.to_sql() + assert sql == "SELECT pg_advisory_lock(%s)" + assert params == (12345,) + + +class TestAsyncAdvisoryLockBackend: + """Async tests for advisory lock backend methods.""" + + async def test_async_execute_advisory_lock(self, async_postgres_backend_single): + """Test acquiring and releasing advisory lock (async).""" + # Acquire lock + await async_postgres_backend_single.execute_advisory_lock(key=12345) + # Release lock + result = await async_postgres_backend_single.execute_advisory_unlock(key=12345) + assert result is True + + async def test_async_try_advisory_lock(self, async_postgres_backend_single): + """Test try advisory lock (async).""" + result = await async_postgres_backend_single.try_advisory_lock(key=12345) + assert result is True + # Clean up + await async_postgres_backend_single.execute_advisory_unlock(key=12345) + + async def test_async_advisory_lock_context_manager(self, async_postgres_backend_single): + """Test advisory lock async context manager.""" + async with async_postgres_backend_single.advisory_lock(key=12345): + # Lock should be held + pass + # Lock should be released after context exit + result = await async_postgres_backend_single.try_advisory_lock(key=12345) + assert result is True + await async_postgres_backend_single.execute_advisory_unlock(key=12345) \ No newline at end of file diff --git a/tests/rhosocial/activerecord_postgres_test/feature/backend/test_row_level_locks.py b/tests/rhosocial/activerecord_postgres_test/feature/backend/test_row_level_locks.py new file mode 100644 index 00000000..dd527e36 --- /dev/null +++ b/tests/rhosocial/activerecord_postgres_test/feature/backend/test_row_level_locks.py @@ -0,0 +1,219 @@ +# tests/rhosocial/activerecord_postgres_test/feature/query/test_row_level_locks.py +""" +PostgreSQL row-level lock tests. + +This module tests PostgreSQL's advanced row-level locking features: +- FOR UPDATE (exclusive lock) +- FOR NO KEY UPDATE (weaker exclusive lock) +- FOR SHARE (shared lock) +- FOR KEY SHARE (weakest shared lock) + +Version requirements: +- FOR UPDATE: All PostgreSQL versions +- FOR NO KEY UPDATE: PostgreSQL 9.0+ +- FOR SHARE: PostgreSQL 9.0+ +- FOR KEY SHARE: PostgreSQL 9.3+ +- SKIP LOCKED: PostgreSQL 9.5+ +""" +import pytest +import pytest_asyncio + +from rhosocial.activerecord.backend.expression.query_parts import ( + ForUpdateClause, + LockStrength, +) +from rhosocial.activerecord.backend.dialect.exceptions import UnsupportedFeatureError + + +class TestRowLevelLockStrength: + """Test PostgreSQL row-level lock strength support.""" + + def test_supports_for_update(self, postgres_backend_single): + """Test that FOR UPDATE is always supported.""" + dialect = postgres_backend_single.dialect + assert dialect.supports_for_update() is True + assert dialect.supports_lock_strength(LockStrength.UPDATE) is True + + def test_supports_for_share(self, postgres_backend_single): + """Test FOR SHARE support (PostgreSQL 9.0+).""" + dialect = postgres_backend_single.dialect + if dialect.version >= (9, 0, 0): + assert dialect.supports_for_share() is True + assert dialect.supports_lock_strength(LockStrength.SHARE) is True + else: + assert dialect.supports_for_share() is False + + def test_supports_for_no_key_update(self, postgres_backend_single): + """Test FOR NO KEY UPDATE support (PostgreSQL 9.0+).""" + dialect = postgres_backend_single.dialect + if dialect.version >= (9, 0, 0): + assert dialect.supports_for_no_key_update() is True + assert dialect.supports_lock_strength(LockStrength.NO_KEY_UPDATE) is True + else: + assert dialect.supports_for_no_key_update() is False + + def test_supports_for_key_share(self, postgres_backend_single): + """Test FOR KEY SHARE support (PostgreSQL 9.3+).""" + dialect = postgres_backend_single.dialect + if dialect.version >= (9, 3, 0): + assert dialect.supports_for_key_share() is True + assert dialect.supports_lock_strength(LockStrength.KEY_SHARE) is True + else: + assert dialect.supports_for_key_share() is False + + +class TestForUpdateClauseFormatting: + """Test FOR UPDATE clause SQL formatting.""" + + def test_for_update_basic(self, postgres_backend_single): + """Test basic FOR UPDATE clause.""" + dialect = postgres_backend_single.dialect + clause = ForUpdateClause(dialect) + sql, params = clause.to_sql() + assert sql == "FOR UPDATE" + assert params == () + + def test_for_share_basic(self, postgres_backend_single): + """Test FOR SHARE clause.""" + dialect = postgres_backend_single.dialect + if not dialect.supports_for_share(): + pytest.skip("FOR SHARE not supported on this PostgreSQL version") + + clause = ForUpdateClause(dialect, strength=LockStrength.SHARE) + sql, params = clause.to_sql() + assert sql == "FOR SHARE" + assert params == () + + def test_for_no_key_update_basic(self, postgres_backend_single): + """Test FOR NO KEY UPDATE clause.""" + dialect = postgres_backend_single.dialect + if not dialect.supports_for_no_key_update(): + pytest.skip("FOR NO KEY UPDATE not supported on this PostgreSQL version") + + clause = ForUpdateClause(dialect, strength=LockStrength.NO_KEY_UPDATE) + sql, params = clause.to_sql() + assert sql == "FOR NO KEY UPDATE" + assert params == () + + def test_for_key_share_basic(self, postgres_backend_single): + """Test FOR KEY SHARE clause.""" + dialect = postgres_backend_single.dialect + if not dialect.supports_for_key_share(): + pytest.skip("FOR KEY SHARE not supported on this PostgreSQL version") + + clause = ForUpdateClause(dialect, strength=LockStrength.KEY_SHARE) + sql, params = clause.to_sql() + assert sql == "FOR KEY SHARE" + assert params == () + + def test_for_update_with_nowait(self, postgres_backend_single): + """Test FOR UPDATE with NOWAIT.""" + dialect = postgres_backend_single.dialect + clause = ForUpdateClause(dialect, nowait=True) + sql, params = clause.to_sql() + assert sql == "FOR UPDATE NOWAIT" + assert params == () + + def test_for_share_with_nowait(self, postgres_backend_single): + """Test FOR SHARE with NOWAIT.""" + dialect = postgres_backend_single.dialect + if not dialect.supports_for_share(): + pytest.skip("FOR SHARE not supported on this PostgreSQL version") + + clause = ForUpdateClause(dialect, strength=LockStrength.SHARE, nowait=True) + sql, params = clause.to_sql() + assert sql == "FOR SHARE NOWAIT" + assert params == () + + def test_for_update_with_skip_locked(self, postgres_backend_single): + """Test FOR UPDATE with SKIP LOCKED.""" + dialect = postgres_backend_single.dialect + if not dialect.supports_for_update_skip_locked(): + pytest.skip("SKIP LOCKED not supported on this PostgreSQL version") + + clause = ForUpdateClause(dialect, skip_locked=True) + sql, params = clause.to_sql() + assert sql == "FOR UPDATE SKIP LOCKED" + assert params == () + + def test_for_share_with_skip_locked(self, postgres_backend_single): + """Test FOR SHARE with SKIP LOCKED.""" + dialect = postgres_backend_single.dialect + if not dialect.supports_for_share(): + pytest.skip("FOR SHARE not supported on this PostgreSQL version") + if not dialect.supports_for_update_skip_locked(): + pytest.skip("SKIP LOCKED not supported on this PostgreSQL version") + + clause = ForUpdateClause(dialect, strength=LockStrength.SHARE, skip_locked=True) + sql, params = clause.to_sql() + assert sql == "FOR SHARE SKIP LOCKED" + assert params == () + + def test_for_update_with_of_columns(self, postgres_backend_single): + """Test FOR UPDATE with OF columns.""" + dialect = postgres_backend_single.dialect + clause = ForUpdateClause(dialect, of_columns=["users", "orders"]) + sql, params = clause.to_sql() + assert sql == 'FOR UPDATE OF "users", "orders"' + assert params == () + + def test_for_share_with_of_columns(self, postgres_backend_single): + """Test FOR SHARE with OF columns.""" + dialect = postgres_backend_single.dialect + if not dialect.supports_for_share(): + pytest.skip("FOR SHARE not supported on this PostgreSQL version") + + clause = ForUpdateClause(dialect, strength=LockStrength.SHARE, of_columns=["users"]) + sql, params = clause.to_sql() + assert sql == 'FOR SHARE OF "users"' + assert params == () + + def test_unsupported_lock_strength_raises_error(self, postgres_backend_single): + """Test that unsupported lock strength raises error.""" + dialect = postgres_backend_single.dialect + + # Create a mock dialect that doesn't support FOR SHARE + if dialect.supports_for_share(): + # Skip if this version actually supports it + pytest.skip("This test requires a PostgreSQL version < 9.0") + + clause = ForUpdateClause(dialect, strength=LockStrength.SHARE) + with pytest.raises(UnsupportedFeatureError): + clause.to_sql() + + +class TestAsyncRowLevelLockStrength: + """Async tests for PostgreSQL row-level lock strength support.""" + + async def test_async_supports_for_update(self, async_postgres_backend_single): + """Test that FOR UPDATE is always supported (async).""" + dialect = async_postgres_backend_single.dialect + assert dialect.supports_for_update() is True + assert dialect.supports_lock_strength(LockStrength.UPDATE) is True + + async def test_async_supports_for_share(self, async_postgres_backend_single): + """Test FOR SHARE support (async).""" + dialect = async_postgres_backend_single.dialect + if dialect.version >= (9, 0, 0): + assert dialect.supports_for_share() is True + else: + assert dialect.supports_for_share() is False + + async def test_async_for_update_basic(self, async_postgres_backend_single): + """Test basic FOR UPDATE clause (async).""" + dialect = async_postgres_backend_single.dialect + clause = ForUpdateClause(dialect) + sql, params = clause.to_sql() + assert sql == "FOR UPDATE" + assert params == () + + async def test_async_for_share_basic(self, async_postgres_backend_single): + """Test FOR SHARE clause (async).""" + dialect = async_postgres_backend_single.dialect + if not dialect.supports_for_share(): + pytest.skip("FOR SHARE not supported on this PostgreSQL version") + + clause = ForUpdateClause(dialect, strength=LockStrength.SHARE) + sql, params = clause.to_sql() + assert sql == "FOR SHARE" + assert params == () \ No newline at end of file From 1613d61375f3ac7d355bd82f7741be9e8a046d9d Mon Sep 17 00:00:00 2001 From: zhao vistart Date: Thu, 9 Apr 2026 11:38:51 +0800 Subject: [PATCH 2/4] refactor(postgres): migrate advisory lock and lock strength to extension package - Move advisory lock expressions from core to postgres/expression/advisory/ - Define LockStrength enum and PostgresForUpdateClause in postgres/expression/ - PostgresForUpdateClause.to_sql() handles PostgreSQL-specific lock strengths - Update import paths in backend, mixin, and protocol files - Reorganize test directory: move activerecord_test/ to activerecord_postgres_test/ - Add postgres_dialect fixture to backend conftest.py - Add .claude/tmp/ to .gitignore Co-Authored-By: Claude Opus 4.6 --- .gitignore | 3 +- .../impl/postgres/backend/async_backend.py | 8 +- .../backend/impl/postgres/backend/sync.py | 8 +- .../backend/impl/postgres/dialect.py | 2 +- .../impl/postgres/expression/__init__.py | 26 +++ .../postgres/expression/advisory/__init__.py | 23 ++ .../impl/postgres/expression/advisory/lock.py | 215 ++++++++++++++++++ .../impl/postgres/expression/locking.py | 178 +++++++++++++++ .../impl/postgres/mixins/dml/advisory_lock.py | 2 +- .../postgres/protocols/dml/advisory_lock.py | 2 +- .../feature/backend/conftest.py | 7 + .../test_expressions_transaction.py | 0 .../feature/backend/test_advisory_lock.py | 2 +- .../feature/backend/test_row_level_locks.py | 32 +-- .../feature/backend/postgres/conftest.py | 11 - 15 files changed, 479 insertions(+), 40 deletions(-) create mode 100644 src/rhosocial/activerecord/backend/impl/postgres/expression/__init__.py create mode 100644 src/rhosocial/activerecord/backend/impl/postgres/expression/advisory/__init__.py create mode 100644 src/rhosocial/activerecord/backend/impl/postgres/expression/advisory/lock.py create mode 100644 src/rhosocial/activerecord/backend/impl/postgres/expression/locking.py rename tests/rhosocial/{activerecord_test/feature/backend/postgres => activerecord_postgres_test/feature/backend/expression}/test_expressions_transaction.py (100%) delete mode 100644 tests/rhosocial/activerecord_test/feature/backend/postgres/conftest.py diff --git a/.gitignore b/.gitignore index 958a51d0..f2be2d67 100644 --- a/.gitignore +++ b/.gitignore @@ -176,5 +176,6 @@ cython_debug/ .env*coverage*.json htmlcov/ -# Claude Code plans (local only) +# Claude Code temporary files (local only) +.claude/tmp/ .claude/plans/ diff --git a/src/rhosocial/activerecord/backend/impl/postgres/backend/async_backend.py b/src/rhosocial/activerecord/backend/impl/postgres/backend/async_backend.py index 1798f42b..ffd65598 100644 --- a/src/rhosocial/activerecord/backend/impl/postgres/backend/async_backend.py +++ b/src/rhosocial/activerecord/backend/impl/postgres/backend/async_backend.py @@ -873,7 +873,7 @@ async def execute_advisory_lock( Raises: DatabaseError: If the lock operation fails """ - from rhosocial.activerecord.backend.expression.advisory import AdvisoryLockExpression + from rhosocial.activerecord.backend.impl.postgres.expression.advisory import AdvisoryLockExpression expr = AdvisoryLockExpression(self.dialect, key=key, shared=shared, session=session) sql, params = expr.to_sql() @@ -890,7 +890,7 @@ async def execute_advisory_unlock(self, key, shared: bool = False) -> bool: Returns: True if the lock was released, False if it was not held """ - from rhosocial.activerecord.backend.expression.advisory import AdvisoryUnlockExpression + from rhosocial.activerecord.backend.impl.postgres.expression.advisory import AdvisoryUnlockExpression expr = AdvisoryUnlockExpression(self.dialect, key=key, shared=shared) sql, params = expr.to_sql() @@ -906,7 +906,7 @@ async def execute_advisory_unlock_all(self) -> None: """ Release all advisory locks held by the current session asynchronously. """ - from rhosocial.activerecord.backend.expression.advisory import AdvisoryUnlockAllExpression + from rhosocial.activerecord.backend.impl.postgres.expression.advisory import AdvisoryUnlockAllExpression expr = AdvisoryUnlockAllExpression(self.dialect) sql, params = expr.to_sql() @@ -929,7 +929,7 @@ async def try_advisory_lock( Returns: True if the lock was acquired, False if it was not available """ - from rhosocial.activerecord.backend.expression.advisory import TryAdvisoryLockExpression + from rhosocial.activerecord.backend.impl.postgres.expression.advisory import TryAdvisoryLockExpression expr = TryAdvisoryLockExpression(self.dialect, key=key, shared=shared, session=session) sql, params = expr.to_sql() diff --git a/src/rhosocial/activerecord/backend/impl/postgres/backend/sync.py b/src/rhosocial/activerecord/backend/impl/postgres/backend/sync.py index d770c2ca..322d42d2 100644 --- a/src/rhosocial/activerecord/backend/impl/postgres/backend/sync.py +++ b/src/rhosocial/activerecord/backend/impl/postgres/backend/sync.py @@ -869,7 +869,7 @@ def execute_advisory_lock( Raises: DatabaseError: If the lock operation fails """ - from rhosocial.activerecord.backend.expression.advisory import AdvisoryLockExpression + from rhosocial.activerecord.backend.impl.postgres.expression.advisory import AdvisoryLockExpression expr = AdvisoryLockExpression(self.dialect, key=key, shared=shared, session=session) sql, params = expr.to_sql() @@ -886,7 +886,7 @@ def execute_advisory_unlock(self, key, shared: bool = False) -> bool: Returns: True if the lock was released, False if it was not held """ - from rhosocial.activerecord.backend.expression.advisory import AdvisoryUnlockExpression + from rhosocial.activerecord.backend.impl.postgres.expression.advisory import AdvisoryUnlockExpression expr = AdvisoryUnlockExpression(self.dialect, key=key, shared=shared) sql, params = expr.to_sql() @@ -902,7 +902,7 @@ def execute_advisory_unlock_all(self) -> None: """ Release all advisory locks held by the current session. """ - from rhosocial.activerecord.backend.expression.advisory import AdvisoryUnlockAllExpression + from rhosocial.activerecord.backend.impl.postgres.expression.advisory import AdvisoryUnlockAllExpression expr = AdvisoryUnlockAllExpression(self.dialect) sql, params = expr.to_sql() @@ -925,7 +925,7 @@ def try_advisory_lock( Returns: True if the lock was acquired, False if it was not available """ - from rhosocial.activerecord.backend.expression.advisory import TryAdvisoryLockExpression + from rhosocial.activerecord.backend.impl.postgres.expression.advisory import TryAdvisoryLockExpression expr = TryAdvisoryLockExpression(self.dialect, key=key, shared=shared, session=session) sql, params = expr.to_sql() diff --git a/src/rhosocial/activerecord/backend/impl/postgres/dialect.py b/src/rhosocial/activerecord/backend/impl/postgres/dialect.py index 218fcaba..47289d70 100644 --- a/src/rhosocial/activerecord/backend/impl/postgres/dialect.py +++ b/src/rhosocial/activerecord/backend/impl/postgres/dialect.py @@ -493,7 +493,7 @@ def supports_lock_strength(self, strength) -> bool: Returns: True if the lock strength is supported, False otherwise """ - from rhosocial.activerecord.backend.expression.query_parts import LockStrength + from rhosocial.activerecord.backend.impl.postgres.expression.locking import LockStrength if strength == LockStrength.UPDATE: return True # All PostgreSQL versions support FOR UPDATE diff --git a/src/rhosocial/activerecord/backend/impl/postgres/expression/__init__.py b/src/rhosocial/activerecord/backend/impl/postgres/expression/__init__.py new file mode 100644 index 00000000..16bf2c00 --- /dev/null +++ b/src/rhosocial/activerecord/backend/impl/postgres/expression/__init__.py @@ -0,0 +1,26 @@ +# src/rhosocial/activerecord/backend/impl/postgres/expression/__init__.py +""" +PostgreSQL-specific expression classes. + +This module provides expression classes that are specific to PostgreSQL, +such as advisory lock expressions and row-level locking support. +""" + +from .advisory import ( + AdvisoryLockExpression, + AdvisoryUnlockExpression, + AdvisoryUnlockAllExpression, + TryAdvisoryLockExpression, + AdvisoryLockType, +) +from .locking import LockStrength, PostgresForUpdateClause + +__all__ = [ + "AdvisoryLockExpression", + "AdvisoryUnlockExpression", + "AdvisoryUnlockAllExpression", + "TryAdvisoryLockExpression", + "AdvisoryLockType", + "LockStrength", + "PostgresForUpdateClause", +] \ No newline at end of file diff --git a/src/rhosocial/activerecord/backend/impl/postgres/expression/advisory/__init__.py b/src/rhosocial/activerecord/backend/impl/postgres/expression/advisory/__init__.py new file mode 100644 index 00000000..1390e904 --- /dev/null +++ b/src/rhosocial/activerecord/backend/impl/postgres/expression/advisory/__init__.py @@ -0,0 +1,23 @@ +# src/rhosocial/activerecord/backend/expression/advisory/__init__.py +""" +Advisory lock expression classes. + +This module provides expression classes for database advisory locks, +which are application-level locks managed by the database server. +""" + +from .lock import ( + AdvisoryLockExpression, + AdvisoryUnlockExpression, + AdvisoryUnlockAllExpression, + TryAdvisoryLockExpression, + AdvisoryLockType, +) + +__all__ = [ + "AdvisoryLockExpression", + "AdvisoryUnlockExpression", + "AdvisoryUnlockAllExpression", + "TryAdvisoryLockExpression", + "AdvisoryLockType", +] \ No newline at end of file diff --git a/src/rhosocial/activerecord/backend/impl/postgres/expression/advisory/lock.py b/src/rhosocial/activerecord/backend/impl/postgres/expression/advisory/lock.py new file mode 100644 index 00000000..e303abe2 --- /dev/null +++ b/src/rhosocial/activerecord/backend/impl/postgres/expression/advisory/lock.py @@ -0,0 +1,215 @@ +# src/rhosocial/activerecord/backend/impl/postgres/expression/advisory/lock.py +""" +Advisory lock expression classes for PostgreSQL. + +Advisory locks are application-level locks that provide a way for +applications to coordinate activities across multiple sessions. +Unlike regular locks, advisory locks are not tied to specific table rows +and can be used for arbitrary application-defined purposes. + +PostgreSQL advisory locks: +- pg_advisory_lock(key) - acquire exclusive session-level lock +- pg_advisory_lock_shared(key) - acquire shared session-level lock +- pg_advisory_unlock(key) - release session-level lock +- pg_advisory_unlock_all(key) - release all session-level locks +- pg_try_advisory_lock(key) - non-blocking acquire exclusive lock +- pg_try_advisory_lock_shared(key) - non-blocking acquire shared lock +- pg_advisory_xact_lock(key) - acquire exclusive transaction-level lock +- pg_advisory_xact_lock_shared(key) - acquire shared transaction-level lock +- pg_try_advisory_xact_lock(key) - non-blocking acquire transaction-level lock + +Keys can be either a single 64-bit integer or two 32-bit integers. +""" + +from enum import Enum +from typing import Tuple, Union, TYPE_CHECKING + +from rhosocial.activerecord.backend.expression.bases import BaseExpression, SQLQueryAndParams + +if TYPE_CHECKING: + from rhosocial.activerecord.backend.dialect import SQLDialectBase + + +class AdvisoryLockType(Enum): + """Type of advisory lock.""" + + EXCLUSIVE = "exclusive" # Only one session can hold the lock + SHARED = "shared" # Multiple sessions can hold the lock simultaneously + + +class AdvisoryLockExpression(BaseExpression): + """ + Expression for acquiring an advisory lock. + + Advisory locks are application-level locks managed by the database server. + They are not tied to specific table rows and can be used for arbitrary + coordination purposes. + + Usage: + # Acquire exclusive session-level lock + expr = AdvisoryLockExpression(dialect, key=12345) + sql, params = expr.to_sql() # SELECT pg_advisory_lock(12345) + + # Acquire shared session-level lock + expr = AdvisoryLockExpression(dialect, key=12345, shared=True) + sql, params = expr.to_sql() # SELECT pg_advisory_lock_shared(12345) + + # Transaction-level lock (auto-released on commit/rollback) + expr = AdvisoryLockExpression(dialect, key=12345, session=False) + sql, params = expr.to_sql() # SELECT pg_advisory_xact_lock(12345) + + # Two-key variant + expr = AdvisoryLockExpression(dialect, key=(123, 456)) + sql, params = expr.to_sql() # SELECT pg_advisory_lock(123, 456) + """ + + def __init__( + self, + dialect: "SQLDialectBase", + key: Union[int, Tuple[int, int]], + shared: bool = False, + session: bool = True, + ): + """ + Initialize advisory lock expression. + + Args: + dialect: The SQL dialect to use for formatting + key: Lock key - either a single 64-bit integer or tuple of two 32-bit integers + shared: If True, acquire shared lock; otherwise exclusive lock + session: If True (default), session-level lock; otherwise transaction-level + """ + super().__init__(dialect) + self.key = key + self.shared = shared + self.session = session + + def to_sql(self) -> SQLQueryAndParams: + """ + Generate SQL for acquiring advisory lock. + + Returns: + Tuple of (SQL string, parameters tuple) + """ + return self.dialect.format_advisory_lock(self) + + +class AdvisoryUnlockExpression(BaseExpression): + """ + Expression for releasing an advisory lock. + + Usage: + # Release session-level lock + expr = AdvisoryUnlockExpression(dialect, key=12345) + sql, params = expr.to_sql() # SELECT pg_advisory_unlock(12345) + + # Release shared session-level lock + expr = AdvisoryUnlockExpression(dialect, key=12345, shared=True) + # Note: pg_advisory_unlock works for both exclusive and shared locks + + # Two-key variant + expr = AdvisoryUnlockExpression(dialect, key=(123, 456)) + sql, params = expr.to_sql() # SELECT pg_advisory_unlock(123, 456) + """ + + def __init__( + self, + dialect: "SQLDialectBase", + key: Union[int, Tuple[int, int]], + shared: bool = False, + ): + """ + Initialize advisory unlock expression. + + Args: + dialect: The SQL dialect to use for formatting + key: Lock key - either a single 64-bit integer or tuple of two 32-bit integers + shared: If True, indicates this was a shared lock (informational only) + """ + super().__init__(dialect) + self.key = key + self.shared = shared + + def to_sql(self) -> SQLQueryAndParams: + """ + Generate SQL for releasing advisory lock. + + Returns: + Tuple of (SQL string, parameters tuple) + """ + return self.dialect.format_advisory_unlock(self) + + +class AdvisoryUnlockAllExpression(BaseExpression): + """ + Expression for releasing all advisory locks held by the current session. + + Usage: + expr = AdvisoryUnlockAllExpression(dialect) + sql, params = expr.to_sql() # SELECT pg_advisory_unlock_all() + """ + + def __init__(self, dialect: "SQLDialectBase"): + """Initialize advisory unlock all expression.""" + super().__init__(dialect) + + def to_sql(self) -> SQLQueryAndParams: + """ + Generate SQL for releasing all advisory locks. + + Returns: + Tuple of (SQL string, parameters tuple) + """ + return self.dialect.format_advisory_unlock_all(self) + + +class TryAdvisoryLockExpression(BaseExpression): + """ + Expression for non-blocking advisory lock acquisition. + + Returns a boolean indicating whether the lock was acquired. + Unlike AdvisoryLockExpression, this does not wait if the lock is unavailable. + + Usage: + # Non-blocking acquire exclusive session-level lock + expr = TryAdvisoryLockExpression(dialect, key=12345) + sql, params = expr.to_sql() # SELECT pg_try_advisory_lock(12345) + + # Non-blocking acquire shared session-level lock + expr = TryAdvisoryLockExpression(dialect, key=12345, shared=True) + sql, params = expr.to_sql() # SELECT pg_try_advisory_lock_shared(12345) + + # Transaction-level variant + expr = TryAdvisoryLockExpression(dialect, key=12345, session=False) + sql, params = expr.to_sql() # SELECT pg_try_advisory_xact_lock(12345) + """ + + def __init__( + self, + dialect: "SQLDialectBase", + key: Union[int, Tuple[int, int]], + shared: bool = False, + session: bool = True, + ): + """ + Initialize try advisory lock expression. + + Args: + dialect: The SQL dialect to use for formatting + key: Lock key - either a single 64-bit integer or tuple of two 32-bit integers + shared: If True, acquire shared lock; otherwise exclusive lock + session: If True (default), session-level lock; otherwise transaction-level + """ + super().__init__(dialect) + self.key = key + self.shared = shared + self.session = session + + def to_sql(self) -> SQLQueryAndParams: + """ + Generate SQL for non-blocking advisory lock acquisition. + + Returns: + Tuple of (SQL string, parameters tuple) + """ + return self.dialect.format_try_advisory_lock(self) \ No newline at end of file diff --git a/src/rhosocial/activerecord/backend/impl/postgres/expression/locking.py b/src/rhosocial/activerecord/backend/impl/postgres/expression/locking.py new file mode 100644 index 00000000..47e0c763 --- /dev/null +++ b/src/rhosocial/activerecord/backend/impl/postgres/expression/locking.py @@ -0,0 +1,178 @@ +# src/rhosocial/activerecord/backend/impl/postgres/expression/locking.py +""" +PostgreSQL-specific row-level locking expressions. + +PostgreSQL supports advanced row-level locking with multiple lock strengths +beyond the standard FOR UPDATE: +- FOR NO KEY UPDATE: Weaker exclusive lock that allows FOR KEY SHARE +- FOR SHARE: Shared lock that allows other readers +- FOR KEY SHARE: Weakest shared lock + +Version requirements: +- FOR UPDATE: All PostgreSQL versions +- FOR NO KEY UPDATE: PostgreSQL 9.0+ +- FOR SHARE: PostgreSQL 9.0+ +- FOR KEY SHARE: PostgreSQL 9.3+ +""" + +from enum import Enum +from typing import Any, Dict, List, Optional, Union, TYPE_CHECKING + +from rhosocial.activerecord.backend.expression.query_parts import ForUpdateClause +from rhosocial.activerecord.backend.expression import bases + +if TYPE_CHECKING: + from rhosocial.activerecord.backend.dialect import SQLDialectBase + + +class LockStrength(Enum): + """ + Enumeration of PostgreSQL row-level lock strength options. + + These options control the type of lock acquired on selected rows. + Only PostgreSQL supports lock strengths beyond FOR UPDATE. + + Lock strength hierarchy (PostgreSQL): + - UPDATE: Strongest exclusive lock, blocks all other locks + - NO_KEY_UPDATE: Weaker exclusive lock, allows KEY_SHARE + - SHARE: Shared lock, allows other SHARE and KEY_SHARE + - KEY_SHARE: Weakest shared lock, allows NO_KEY_UPDATE and SHARE + + Version requirements (PostgreSQL): + - FOR UPDATE: All versions + - FOR NO KEY UPDATE: PostgreSQL 9.0+ + - FOR SHARE: PostgreSQL 9.0+ + - FOR KEY SHARE: PostgreSQL 9.3+ + """ + + UPDATE = "FOR UPDATE" # Exclusive lock (strongest) + NO_KEY_UPDATE = "FOR NO KEY UPDATE" # Weaker exclusive lock + SHARE = "FOR SHARE" # Shared lock + KEY_SHARE = "FOR KEY SHARE" # Weakest shared lock + + +class PostgresForUpdateClause(ForUpdateClause): + """ + PostgreSQL-specific FOR UPDATE clause with lock strength support. + + Extends the standard ForUpdateClause with PostgreSQL's advanced + row-level locking capabilities. + + PostgreSQL supports additional lock strengths beyond FOR UPDATE: + - FOR NO KEY UPDATE: Weaker exclusive lock that allows FOR KEY SHARE + - FOR SHARE: Shared lock that allows other readers + - FOR KEY SHARE: Weakest shared lock + + Example Usage: + # Basic FOR UPDATE (same as parent class) + for_update = PostgresForUpdateClause(dialect) + + # FOR SHARE (PostgreSQL 9.0+) + for_update = PostgresForUpdateClause(dialect, strength=LockStrength.SHARE) + + # FOR NO KEY UPDATE (PostgreSQL 9.0+) + for_update = PostgresForUpdateClause(dialect, strength=LockStrength.NO_KEY_UPDATE) + + # FOR KEY SHARE (PostgreSQL 9.3+) + for_update = PostgresForUpdateClause(dialect, strength=LockStrength.KEY_SHARE) + + # FOR SHARE with NOWAIT + for_update = PostgresForUpdateClause( + dialect, + strength=LockStrength.SHARE, + nowait=True + ) + """ + + def __init__( + self, + dialect: "SQLDialectBase", + strength: Optional[LockStrength] = None, + of_columns: Optional[List[Union[str, "bases.BaseExpression"]]] = None, + nowait: bool = False, + skip_locked: bool = False, + dialect_options: Optional[Dict[str, Any]] = None, + ): + """ + Initialize PostgreSQL FOR UPDATE clause. + + Args: + dialect: The SQL dialect to use for formatting + strength: Lock strength (defaults to UPDATE for backward compatibility) + of_columns: Columns to apply the lock to + nowait: If True, fail immediately if rows are locked + skip_locked: If True, skip locked rows instead of waiting + dialect_options: Additional dialect-specific options + """ + super().__init__( + dialect, + of_columns=of_columns, + nowait=nowait, + skip_locked=skip_locked, + dialect_options=dialect_options, + ) + # Default to UPDATE for backward compatibility + self.strength = strength if strength is not None else LockStrength.UPDATE + + def to_sql(self) -> "bases.SQLQueryAndParams": + """ + Generate the SQL representation of the PostgreSQL FOR UPDATE clause. + + This method handles PostgreSQL-specific lock strengths (FOR SHARE, + FOR NO KEY UPDATE, FOR KEY SHARE) in addition to the standard + FOR UPDATE clause. + + Returns: + Tuple containing: + - SQL string fragment for the FOR UPDATE clause + - Tuple of parameter values for prepared statements + """ + from rhosocial.activerecord.backend.dialect.exceptions import UnsupportedFeatureError + + all_params = [] + + # Check version support for lock strength + if self.strength == LockStrength.NO_KEY_UPDATE: + if not self.dialect.supports_for_no_key_update(): + raise UnsupportedFeatureError( + self.dialect.name, "FOR NO KEY UPDATE (requires PostgreSQL 9.0+)" + ) + elif self.strength == LockStrength.SHARE: + if not self.dialect.supports_for_share(): + raise UnsupportedFeatureError( + self.dialect.name, "FOR SHARE (requires PostgreSQL 9.0+)" + ) + elif self.strength == LockStrength.KEY_SHARE: + if not self.dialect.supports_for_key_share(): + raise UnsupportedFeatureError( + self.dialect.name, "FOR KEY SHARE (requires PostgreSQL 9.3+)" + ) + + # Use the strength value directly (e.g., "FOR UPDATE", "FOR SHARE") + sql_parts = [self.strength.value] + + # Handle OF columns if specified + if self.of_columns: + of_parts = [] + for col in self.of_columns: + if isinstance(col, str): + of_parts.append(self.dialect.format_identifier(col)) + else: + # BaseExpression + col_sql, col_params = col.to_sql() + of_parts.append(col_sql) + all_params.extend(col_params) + if of_parts: + sql_parts.append(f"OF {', '.join(of_parts)}") + + # Handle NOWAIT/SKIP LOCKED options + if self.nowait: + sql_parts.append("NOWAIT") + elif self.skip_locked: + if not self.dialect.supports_for_update_skip_locked(): + raise UnsupportedFeatureError( + self.dialect.name, "SKIP LOCKED (requires PostgreSQL 9.5+)" + ) + sql_parts.append("SKIP LOCKED") + + return " ".join(sql_parts), tuple(all_params) \ No newline at end of file diff --git a/src/rhosocial/activerecord/backend/impl/postgres/mixins/dml/advisory_lock.py b/src/rhosocial/activerecord/backend/impl/postgres/mixins/dml/advisory_lock.py index ad493681..34b7c2de 100644 --- a/src/rhosocial/activerecord/backend/impl/postgres/mixins/dml/advisory_lock.py +++ b/src/rhosocial/activerecord/backend/impl/postgres/mixins/dml/advisory_lock.py @@ -9,7 +9,7 @@ from typing import Tuple, TYPE_CHECKING if TYPE_CHECKING: - from rhosocial.activerecord.backend.expression.advisory import ( + from rhosocial.activerecord.backend.impl.postgres.expression.advisory import ( AdvisoryLockExpression, AdvisoryUnlockExpression, AdvisoryUnlockAllExpression, diff --git a/src/rhosocial/activerecord/backend/impl/postgres/protocols/dml/advisory_lock.py b/src/rhosocial/activerecord/backend/impl/postgres/protocols/dml/advisory_lock.py index 95d1b2d9..ff6beb94 100644 --- a/src/rhosocial/activerecord/backend/impl/postgres/protocols/dml/advisory_lock.py +++ b/src/rhosocial/activerecord/backend/impl/postgres/protocols/dml/advisory_lock.py @@ -10,7 +10,7 @@ from typing import Protocol, Tuple, Union, TYPE_CHECKING if TYPE_CHECKING: - from rhosocial.activerecord.backend.expression.advisory import ( + from rhosocial.activerecord.backend.impl.postgres.expression.advisory import ( AdvisoryLockExpression, AdvisoryUnlockExpression, AdvisoryUnlockAllExpression, diff --git a/tests/rhosocial/activerecord_postgres_test/feature/backend/conftest.py b/tests/rhosocial/activerecord_postgres_test/feature/backend/conftest.py index ad2852bb..8c8a3979 100644 --- a/tests/rhosocial/activerecord_postgres_test/feature/backend/conftest.py +++ b/tests/rhosocial/activerecord_postgres_test/feature/backend/conftest.py @@ -218,3 +218,10 @@ async def async_postgres_control_backend(): backend = await provider.setup_async_backend(scenario_name) yield backend await provider.async_cleanup() + + +@pytest.fixture(scope="function") +def postgres_dialect(): + """Fixture providing PostgresDialect instance for testing transaction expressions.""" + from rhosocial.activerecord.backend.impl.postgres.dialect import PostgresDialect + return PostgresDialect() diff --git a/tests/rhosocial/activerecord_test/feature/backend/postgres/test_expressions_transaction.py b/tests/rhosocial/activerecord_postgres_test/feature/backend/expression/test_expressions_transaction.py similarity index 100% rename from tests/rhosocial/activerecord_test/feature/backend/postgres/test_expressions_transaction.py rename to tests/rhosocial/activerecord_postgres_test/feature/backend/expression/test_expressions_transaction.py diff --git a/tests/rhosocial/activerecord_postgres_test/feature/backend/test_advisory_lock.py b/tests/rhosocial/activerecord_postgres_test/feature/backend/test_advisory_lock.py index d7924a7c..1f2ba0dd 100644 --- a/tests/rhosocial/activerecord_postgres_test/feature/backend/test_advisory_lock.py +++ b/tests/rhosocial/activerecord_postgres_test/feature/backend/test_advisory_lock.py @@ -11,7 +11,7 @@ - Context manager usage """ -from rhosocial.activerecord.backend.expression.advisory import ( +from rhosocial.activerecord.backend.impl.postgres.expression.advisory import ( AdvisoryLockExpression, AdvisoryUnlockExpression, AdvisoryUnlockAllExpression, diff --git a/tests/rhosocial/activerecord_postgres_test/feature/backend/test_row_level_locks.py b/tests/rhosocial/activerecord_postgres_test/feature/backend/test_row_level_locks.py index dd527e36..969db48b 100644 --- a/tests/rhosocial/activerecord_postgres_test/feature/backend/test_row_level_locks.py +++ b/tests/rhosocial/activerecord_postgres_test/feature/backend/test_row_level_locks.py @@ -18,9 +18,9 @@ import pytest import pytest_asyncio -from rhosocial.activerecord.backend.expression.query_parts import ( - ForUpdateClause, +from rhosocial.activerecord.backend.impl.postgres.expression.locking import ( LockStrength, + PostgresForUpdateClause, ) from rhosocial.activerecord.backend.dialect.exceptions import UnsupportedFeatureError @@ -62,13 +62,13 @@ def test_supports_for_key_share(self, postgres_backend_single): assert dialect.supports_for_key_share() is False -class TestForUpdateClauseFormatting: +class TestPostgresForUpdateClauseFormatting: """Test FOR UPDATE clause SQL formatting.""" def test_for_update_basic(self, postgres_backend_single): """Test basic FOR UPDATE clause.""" dialect = postgres_backend_single.dialect - clause = ForUpdateClause(dialect) + clause = PostgresForUpdateClause(dialect) sql, params = clause.to_sql() assert sql == "FOR UPDATE" assert params == () @@ -79,7 +79,7 @@ def test_for_share_basic(self, postgres_backend_single): if not dialect.supports_for_share(): pytest.skip("FOR SHARE not supported on this PostgreSQL version") - clause = ForUpdateClause(dialect, strength=LockStrength.SHARE) + clause = PostgresForUpdateClause(dialect, strength=LockStrength.SHARE) sql, params = clause.to_sql() assert sql == "FOR SHARE" assert params == () @@ -90,7 +90,7 @@ def test_for_no_key_update_basic(self, postgres_backend_single): if not dialect.supports_for_no_key_update(): pytest.skip("FOR NO KEY UPDATE not supported on this PostgreSQL version") - clause = ForUpdateClause(dialect, strength=LockStrength.NO_KEY_UPDATE) + clause = PostgresForUpdateClause(dialect, strength=LockStrength.NO_KEY_UPDATE) sql, params = clause.to_sql() assert sql == "FOR NO KEY UPDATE" assert params == () @@ -101,7 +101,7 @@ def test_for_key_share_basic(self, postgres_backend_single): if not dialect.supports_for_key_share(): pytest.skip("FOR KEY SHARE not supported on this PostgreSQL version") - clause = ForUpdateClause(dialect, strength=LockStrength.KEY_SHARE) + clause = PostgresForUpdateClause(dialect, strength=LockStrength.KEY_SHARE) sql, params = clause.to_sql() assert sql == "FOR KEY SHARE" assert params == () @@ -109,7 +109,7 @@ def test_for_key_share_basic(self, postgres_backend_single): def test_for_update_with_nowait(self, postgres_backend_single): """Test FOR UPDATE with NOWAIT.""" dialect = postgres_backend_single.dialect - clause = ForUpdateClause(dialect, nowait=True) + clause = PostgresForUpdateClause(dialect, nowait=True) sql, params = clause.to_sql() assert sql == "FOR UPDATE NOWAIT" assert params == () @@ -120,7 +120,7 @@ def test_for_share_with_nowait(self, postgres_backend_single): if not dialect.supports_for_share(): pytest.skip("FOR SHARE not supported on this PostgreSQL version") - clause = ForUpdateClause(dialect, strength=LockStrength.SHARE, nowait=True) + clause = PostgresForUpdateClause(dialect, strength=LockStrength.SHARE, nowait=True) sql, params = clause.to_sql() assert sql == "FOR SHARE NOWAIT" assert params == () @@ -131,7 +131,7 @@ def test_for_update_with_skip_locked(self, postgres_backend_single): if not dialect.supports_for_update_skip_locked(): pytest.skip("SKIP LOCKED not supported on this PostgreSQL version") - clause = ForUpdateClause(dialect, skip_locked=True) + clause = PostgresForUpdateClause(dialect, skip_locked=True) sql, params = clause.to_sql() assert sql == "FOR UPDATE SKIP LOCKED" assert params == () @@ -144,7 +144,7 @@ def test_for_share_with_skip_locked(self, postgres_backend_single): if not dialect.supports_for_update_skip_locked(): pytest.skip("SKIP LOCKED not supported on this PostgreSQL version") - clause = ForUpdateClause(dialect, strength=LockStrength.SHARE, skip_locked=True) + clause = PostgresForUpdateClause(dialect, strength=LockStrength.SHARE, skip_locked=True) sql, params = clause.to_sql() assert sql == "FOR SHARE SKIP LOCKED" assert params == () @@ -152,7 +152,7 @@ def test_for_share_with_skip_locked(self, postgres_backend_single): def test_for_update_with_of_columns(self, postgres_backend_single): """Test FOR UPDATE with OF columns.""" dialect = postgres_backend_single.dialect - clause = ForUpdateClause(dialect, of_columns=["users", "orders"]) + clause = PostgresForUpdateClause(dialect, of_columns=["users", "orders"]) sql, params = clause.to_sql() assert sql == 'FOR UPDATE OF "users", "orders"' assert params == () @@ -163,7 +163,7 @@ def test_for_share_with_of_columns(self, postgres_backend_single): if not dialect.supports_for_share(): pytest.skip("FOR SHARE not supported on this PostgreSQL version") - clause = ForUpdateClause(dialect, strength=LockStrength.SHARE, of_columns=["users"]) + clause = PostgresForUpdateClause(dialect, strength=LockStrength.SHARE, of_columns=["users"]) sql, params = clause.to_sql() assert sql == 'FOR SHARE OF "users"' assert params == () @@ -177,7 +177,7 @@ def test_unsupported_lock_strength_raises_error(self, postgres_backend_single): # Skip if this version actually supports it pytest.skip("This test requires a PostgreSQL version < 9.0") - clause = ForUpdateClause(dialect, strength=LockStrength.SHARE) + clause = PostgresForUpdateClause(dialect, strength=LockStrength.SHARE) with pytest.raises(UnsupportedFeatureError): clause.to_sql() @@ -202,7 +202,7 @@ async def test_async_supports_for_share(self, async_postgres_backend_single): async def test_async_for_update_basic(self, async_postgres_backend_single): """Test basic FOR UPDATE clause (async).""" dialect = async_postgres_backend_single.dialect - clause = ForUpdateClause(dialect) + clause = PostgresForUpdateClause(dialect) sql, params = clause.to_sql() assert sql == "FOR UPDATE" assert params == () @@ -213,7 +213,7 @@ async def test_async_for_share_basic(self, async_postgres_backend_single): if not dialect.supports_for_share(): pytest.skip("FOR SHARE not supported on this PostgreSQL version") - clause = ForUpdateClause(dialect, strength=LockStrength.SHARE) + clause = PostgresForUpdateClause(dialect, strength=LockStrength.SHARE) sql, params = clause.to_sql() assert sql == "FOR SHARE" assert params == () \ No newline at end of file diff --git a/tests/rhosocial/activerecord_test/feature/backend/postgres/conftest.py b/tests/rhosocial/activerecord_test/feature/backend/postgres/conftest.py deleted file mode 100644 index 460be036..00000000 --- a/tests/rhosocial/activerecord_test/feature/backend/postgres/conftest.py +++ /dev/null @@ -1,11 +0,0 @@ -# tests/rhosocial/activerecord_test/feature/backend/postgres/conftest.py -"""Fixtures for PostgreSQL transaction expression tests.""" -import pytest - -from rhosocial.activerecord.backend.impl.postgres.dialect import PostgresDialect - - -@pytest.fixture(scope="function") -def postgres_dialect(): - """Fixture providing PostgresDialect instance for testing transaction expressions.""" - return PostgresDialect() From dc4c380abe19a0464005f1d37595e987a2434c27 Mon Sep 17 00:00:00 2001 From: zhao vistart Date: Thu, 9 Apr 2026 14:39:09 +0800 Subject: [PATCH 3/4] ci: update dependency sources to Git branches - rhosocial-activerecord: git+https@release/v1.0.0.dev23 - rhosocial-activerecord-testsuite: git+https@release/v1.0.0.dev10 Co-Authored-By: Claude Opus 4.6 --- .github/workflows/test.yml | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 10f0dcca..d0e66429 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -65,10 +65,10 @@ jobs: pip install -r requirements.txt pip install -r requirements-dev.txt - - name: Install activerecord and testsuite from PyPI + - name: Install activerecord and testsuite from Git run: | - pip install rhosocial-activerecord==1.0.0.dev22 - pip install rhosocial-activerecord-testsuite==1.0.0.dev9 + pip install git+https://github.com/rhosocial/python-activerecord.git@release/v1.0.0.dev23 + pip install git+https://github.com/rhosocial/python-activerecord-testsuite.git@release/v1.0.0.dev10 - name: Create postgres scenario config run: | From 0255e9aa1cbd23be65b0236ad65c8c683e1e06d8 Mon Sep 17 00:00:00 2001 From: zhao vistart Date: Thu, 9 Apr 2026 14:45:46 +0800 Subject: [PATCH 4/4] chore: add changelog fragment for PR #23 Co-Authored-By: Claude Opus 4.6 --- changelog.d/{row_level_lock.added.md => 23.added.md} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename changelog.d/{row_level_lock.added.md => 23.added.md} (100%) diff --git a/changelog.d/row_level_lock.added.md b/changelog.d/23.added.md similarity index 100% rename from changelog.d/row_level_lock.added.md rename to changelog.d/23.added.md