remove foreign key constraint for buffer tables

Author: Mantisus

src/crawlee/storage_clients/_sql/_client_mixin.py

@@ -282,11 +282,15 @@ async def _drop(self) -> None:
         This operation is irreversible. Uses CASCADE deletion to remove all related items.
         """
         stmt = delete(self._METADATA_TABLE).where(self._METADATA_TABLE.id == self._id)
+        # Delete the buffer records with a separate query, since table don't link via foreign key.
+        buffer_stmt = delete(self._BUFFER_TABLE).where(self._BUFFER_TABLE.storage_id == self._id)
+
         async with self.get_session(with_simple_commit=True) as session:
             if self._storage_client.get_dialect_name() == 'sqlite':
                 # foreign_keys=ON is set at the connection level. Required for cascade deletion.
                 await session.execute(text('PRAGMA foreign_keys=ON'))
             await session.execute(stmt)
+            await session.execute(buffer_stmt)
 
     @overload
     async def _get_metadata(self, metadata_model: type[DatasetMetadata]) -> DatasetMetadata: ...

src/crawlee/storage_clients/_sql/_db_models.py

@@ -87,11 +87,6 @@ class DatasetMetadataDb(StorageMetadataDb, Base):
         back_populates='dataset', cascade='all, delete-orphan', lazy='noload'
     )
 
-    # Relationship to metadata buffer updates
-    buffer: Mapped[list[DatasetMetadataBufferDb]] = relationship(
-        back_populates='dataset', cascade='all, delete-orphan', lazy='noload'
-    )
-
     id = synonym('dataset_id')
     """Alias for dataset_id to match Pydantic expectations."""
 
@@ -132,11 +127,6 @@ class RequestQueueMetadataDb(StorageMetadataDb, Base):
         back_populates='queue', cascade='all, delete-orphan', lazy='noload'
     )
 
-    # Relationship to metadata buffer updates
-    buffer: Mapped[list[RequestQueueMetadataBufferDb]] = relationship(
-        back_populates='queue', cascade='all, delete-orphan', lazy='noload'
-    )
-
     id = synonym('request_queue_id')
     """Alias for request_queue_id to match Pydantic expectations."""
 
@@ -154,11 +144,6 @@ class KeyValueStoreMetadataDb(StorageMetadataDb, Base):
         back_populates='kvs', cascade='all, delete-orphan', lazy='noload'
     )
 
-    # Relationship to metadata buffer updates
-    buffer: Mapped[list[KeyValueStoreMetadataBufferDb]] = relationship(
-        back_populates='kvs', cascade='all, delete-orphan', lazy='noload'
-    )
-
     id = synonym('key_value_store_id')
     """Alias for key_value_store_id to match Pydantic expectations."""
 
@@ -317,14 +302,10 @@ class KeyValueStoreMetadataBufferDb(MetadataBufferDb, Base):
 
     __tablename__ = 'key_value_store_metadata_buffer'
 
-    key_value_store_id: Mapped[str] = mapped_column(
-        String(20), ForeignKey('key_value_stores.key_value_store_id', ondelete='CASCADE'), nullable=False, index=True
-    )
+    # Don't use foreign key constraint to avoid DB locks on high concurrency.
+    key_value_store_id: Mapped[str] = mapped_column(String(20), nullable=False, index=True)
     """ID of the key-value store being updated."""
 
-    # Relationship back to key-value store metadata
-    kvs: Mapped[KeyValueStoreMetadataDb] = relationship(back_populates='buffer')
-
     storage_id = synonym('key_value_store_id')
     """Alias for key_value_store_id to match SqlClientMixin expectations."""
 
@@ -334,18 +315,14 @@ class DatasetMetadataBufferDb(MetadataBufferDb, Base):
 
     __tablename__ = 'dataset_metadata_buffer'
 
-    dataset_id: Mapped[str] = mapped_column(
-        String(20), ForeignKey('datasets.dataset_id', ondelete='CASCADE'), nullable=False, index=True
-    )
+    # Don't use foreign key constraint to avoid DB locks on high concurrency.
+    dataset_id: Mapped[str] = mapped_column(String(20), nullable=False, index=True)
     """ID of the dataset being updated."""
 
-    # Counter deltas - use SUM when aggregating
+    # Counter deltas - use SUM when aggregating.
     delta_item_count: Mapped[int | None] = mapped_column(Integer, nullable=True)
     """Delta for dataset item_count."""
 
-    # Relationship back to dataset metadata
-    dataset: Mapped[DatasetMetadataDb] = relationship(back_populates='buffer')
-
     storage_id = synonym('dataset_id')
     """Alias for dataset_id to match SqlClientMixin expectations."""
 
@@ -357,15 +334,14 @@ class RequestQueueMetadataBufferDb(MetadataBufferDb, Base):
 
     __table_args__ = (Index('idx_rq_client', 'request_queue_id', 'client_id'),)
 
-    request_queue_id: Mapped[str] = mapped_column(
-        String(20), ForeignKey('request_queues.request_queue_id', ondelete='CASCADE'), nullable=False, index=True
-    )
+    # Don't use foreign key constraint to avoid DB locks on high concurrency.
+    request_queue_id: Mapped[str] = mapped_column(String(20), nullable=False, index=True)
     """ID of the request queue being updated."""
 
     client_id: Mapped[str] = mapped_column(String(32), nullable=False)
     """Identifier of the client making this update."""
 
-    # Counter deltas - use SUM when aggregating
+    # Counter deltas - use SUM when aggregating.
     delta_handled_count: Mapped[int | None] = mapped_column(Integer, nullable=True)
     """Delta for handled_request_count."""
 
@@ -378,8 +354,5 @@ class RequestQueueMetadataBufferDb(MetadataBufferDb, Base):
     need_recalc: Mapped[bool | None] = mapped_column(Boolean, nullable=False, default=False)
     """Flag indicating that counters need recalculation from actual data."""
 
-    # Relationship back to request queue metadata
-    queue: Mapped[RequestQueueMetadataDb] = relationship(back_populates='buffer')
-
     storage_id = synonym('request_queue_id')
     """Alias for request_queue_id to match SqlClientMixin expectations."""

src/crawlee/storage_clients/_sql/_request_queue_client.py

@@ -444,7 +444,7 @@ async def fetch_next_request(self) -> Request | None:
 
                 update_stmt = (
                     update(self._ITEM_TABLE)
-                    .where(self._ITEM_TABLE.request_id.in_(request_ids))
+                    .where(self._ITEM_TABLE.request_queue_id == self._id, self._ITEM_TABLE.request_id.in_(request_ids))
                     .values(time_blocked_until=block_until, client_key=self.client_key)
                 )
                 await session.execute(update_stmt)