add mysql/mariadb

Author: Mantisus

pyproject.toml

@@ -81,6 +81,11 @@ sql_sqlite = [
     "sqlalchemy[asyncio]>=2.0.0,<3.0.0",
     "aiosqlite>=0.21.0",
 ]
+sql_mysql = [
+    "sqlalchemy[asyncio]>=2.0.0,<3.0.0",
+    "aiomysql>=0.3.2",
+    "cryptography>=46.0.5",
+]
 redis = ["redis[hiredis] >= 7.0.0"]
 
 [project.scripts]

src/crawlee/storage_clients/_sql/_client_mixin.py

@@ -8,9 +8,10 @@
 
 from sqlalchemy import CursorResult, delete, select, text, update
 from sqlalchemy import func as sql_func
+from sqlalchemy.dialects.mysql import insert as mysql_insert
 from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlalchemy.dialects.sqlite import insert as lite_insert
-from sqlalchemy.exc import SQLAlchemyError
+from sqlalchemy.exc import OperationalError, SQLAlchemyError
 
 from crawlee._utils.crypto import crypto_random_object_id
 
@@ -227,6 +228,9 @@ def _build_insert_stmt_with_ignore(
         if dialect == 'sqlite':
             return lite_insert(table_model).values(insert_values).on_conflict_do_nothing()
 
+        if dialect == 'mysql':
+            return mysql_insert(table_model).values(insert_values).prefix_with('IGNORE')
+
         raise NotImplementedError(f'Insert with ignore not supported for dialect: {dialect}')
 
     def _build_upsert_stmt(
@@ -260,6 +264,11 @@ def _build_upsert_stmt(
             set_ = {col: getattr(lite_stmt.excluded, col) for col in update_columns}
             return lite_stmt.on_conflict_do_update(index_elements=conflict_cols, set_=set_)
 
+        if dialect == 'mysql':
+            mysql_stmt = mysql_insert(table_model).values(insert_values)
+            set_ = {col: getattr(mysql_stmt.inserted, col) for col in update_columns}
+            return mysql_stmt.on_duplicate_key_update(**set_)
+
         raise NotImplementedError(f'Upsert not supported for dialect: {dialect}')
 
     async def _purge(self, metadata_kwargs: MetadataUpdateParams) -> None:
@@ -402,11 +411,12 @@ async def _try_acquire_buffer_lock(self, session: AsyncSession) -> bool:
         Returns:
             True if lock was acquired, False if already locked by another process.
         """
+        capture_error_code = 1020  # MariaDB error code for "Record has changed since last read"
         now = datetime.now(timezone.utc)
         lock_until = now + self._BLOCK_BUFFER_TIME
         dialect = self._storage_client.get_dialect_name()
 
-        if dialect == 'postgresql':
+        if dialect in ('postgresql', 'mysql'):
             select_stmt = (
                 select(self._METADATA_TABLE)
                 .where(
@@ -417,7 +427,17 @@ async def _try_acquire_buffer_lock(self, session: AsyncSession) -> bool:
                 )
                 .with_for_update(skip_locked=True)
             )
-            result = await session.execute(select_stmt)
+
+            try:
+                result = await session.execute(select_stmt)
+            except OperationalError as e:
+                # MariaDB raises error 1020 ("Record has changed since last read") instead of
+                # silently skipping locked rows like MySQL/PostgreSQL. Treat it as lock not acquired.
+                error_code = getattr(e.orig, 'args', [None])[0]
+                if error_code == capture_error_code:
+                    return False
+                raise
+
             metadata_row = result.scalar_one_or_none()
 
             if metadata_row is None:

src/crawlee/storage_clients/_sql/_request_queue_client.py

@@ -335,33 +335,33 @@ async def add_batch_of_requests(
                         )
                     )
 
-            if insert_values:
-                if forefront:
-                    # If the request already exists in the database, we update the sequence_number by shifting request
-                    # to the left.
-                    upsert_stmt = self._build_upsert_stmt(
-                        self._ITEM_TABLE,
-                        insert_values,
-                        update_columns=['sequence_number'],
-                        conflict_cols=['request_id', 'request_queue_id'],
-                    )
-                    result = await session.execute(upsert_stmt)
-                else:
-                    # If the request already exists in the database, we ignore this request when inserting.
-                    insert_stmt_with_ignore = self._build_insert_stmt_with_ignore(self._ITEM_TABLE, insert_values)
-                    result = await session.execute(insert_stmt_with_ignore)
+            try:
+                if insert_values:
+                    if forefront:
+                        # If the request already exists in the database, we update the sequence_number
+                        # by shifting request to the left.
+                        upsert_stmt = self._build_upsert_stmt(
+                            self._ITEM_TABLE,
+                            insert_values,
+                            update_columns=['sequence_number'],
+                            conflict_cols=['request_id', 'request_queue_id'],
+                        )
+                        result = await session.execute(upsert_stmt)
+                    else:
+                        # If the request already exists in the database, we ignore this request when inserting.
+                        insert_stmt_with_ignore = self._build_insert_stmt_with_ignore(self._ITEM_TABLE, insert_values)
+                        result = await session.execute(insert_stmt_with_ignore)
 
-                result = cast('CursorResult', result) if not isinstance(result, CursorResult) else result
-                approximate_new_request += result.rowcount
+                    result = cast('CursorResult', result) if not isinstance(result, CursorResult) else result
+                    approximate_new_request += result.rowcount
 
-            await self._add_buffer_record(
-                session,
-                update_modified_at=True,
-                delta_pending_request_count=approximate_new_request,
-                delta_total_request_count=approximate_new_request,
-            )
+                await self._add_buffer_record(
+                    session,
+                    update_modified_at=True,
+                    delta_pending_request_count=approximate_new_request,
+                    delta_total_request_count=approximate_new_request,
+                )
 
-            try:
                 await session.commit()
                 processed_requests.extend(transaction_processed_requests)
             except SQLAlchemyError as e:
@@ -433,7 +433,7 @@ async def fetch_next_request(self) -> Request | None:
 
         async with self.get_session(with_simple_commit=True) as session:
             # We use the `skip_locked` database mechanism to prevent the 'interception' of requests by another client
-            if dialect == 'postgresql':
+            if dialect in ('postgresql', 'mysql'):
                 stmt = stmt.with_for_update(skip_locked=True)
                 result = await session.execute(stmt)
                 requests_db = result.scalars().all()

src/crawlee/storage_clients/_sql/_storage_client.py

@@ -3,7 +3,7 @@
 import warnings
 from logging import getLogger
 from pathlib import Path
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 from sqlalchemy.exc import IntegrityError, OperationalError
 from sqlalchemy.ext.asyncio import AsyncEngine, async_sessionmaker, create_async_engine
@@ -116,9 +116,9 @@ async def initialize(self, configuration: Configuration) -> None:
             async with engine.begin() as conn:
                 self._dialect_name = engine.dialect.name
 
-                if self._dialect_name not in ('sqlite', 'postgresql'):
+                if self._dialect_name not in ('sqlite', 'postgresql', 'mysql'):
                     raise ValueError(
-                        f'Unsupported database dialect: {self._dialect_name}. Supported: sqlite, postgresql. '
+                        f'Unsupported database dialect: {self._dialect_name}. Supported: sqlite, postgresql, mysql. '
                         'Consider using a different database.',
                     )
 
@@ -256,11 +256,23 @@ def _get_or_create_engine(self, configuration: Configuration) -> AsyncEngine:
             # Create connection string with path to default database
             connection_string = f'sqlite+aiosqlite:///{db_path}'
 
-        if 'sqlite' not in connection_string and 'postgresql' not in connection_string:
+        if (
+            ('sqlite' not in connection_string)
+            and ('postgresql' not in connection_string)
+            and ('mysql' not in connection_string)
+        ):
             raise ValueError(
-                'Unsupported database. Supported: sqlite, postgresql. Consider using a different database.'
+                'Unsupported database. Supported: sqlite, postgresql, mysql. Consider using a different database.'
             )
 
+        connect_args: dict[str, Any]
+        kwargs: dict[str, Any] = {}
+        if 'mysql' in connection_string:
+            connect_args: dict[str, Any] = {'connect_timeout': 30}
+            kwargs['isolation_level'] = 'READ COMMITTED'
+        else:
+            connect_args = {'timeout': 30}
+
         self._engine = create_async_engine(
             connection_string,
             future=True,
@@ -270,6 +282,7 @@ def _get_or_create_engine(self, configuration: Configuration) -> AsyncEngine:
             pool_recycle=600,
             pool_pre_ping=True,
             echo=False,
-            connect_args={'timeout': 30},
+            connect_args=connect_args,
+            **kwargs,
         )
         return self._engine