feat(spider-py): Add job submission support for MariaDB storage backend. - #216
Conversation
Co-authored-by: Lin Zhihao <59785146+LinZhihao-723@users.noreply.github.com>
LinZhihao-723
left a comment
LinZhihao-723
left a comment
There was a problem hiding this comment.
Reviewed mariadb storage.
One thing to confirm: The storage format in the DB is the same as C++ implementation, right?
| self._conn.commit() | ||
| return job_ids | ||
| except mariadb.Error as e: | ||
| self._conn.rollback() |
There was a problem hiding this comment.
Do we need rollback since nothing is committed?
There was a problem hiding this comment.
It is possible some insertions already been performed. We need to rollback these insertions.
Co-authored-by: Lin Zhihao <59785146+LinZhihao-723@users.noreply.github.com>
Yes. |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (4)
python/spider-py/README.md (4)
32-37: Add a pure pytest fallback command.
Consider documenting an equivalent pytest invocation for contributors who don’t use Task (e.g.,uv run pytest -m "not storage").To run all non-storage unit tests: ```shell task test:spider-py-non-storage-unit-tests +uv run pytest -m "not storage" # alternative, if Task is not installed--- `57-63`: **Note DB readiness before running tests.** Recommend waiting for container health to be “healthy” before invoking the storage tests, or use a short retry wrapper in the task. ```diff To run all storage unit tests: ```shell task test:spider-py-storage-unit-tests-This requires a running MariaDB instance as described above.
+This requires a healthy MariaDB instance as described above (wait for the container healthcheck to pass).--- `70-71`: **Avoid repeating the MariaDB requirement.** This note was already stated above; remove here to reduce redundancy. ```diff -This requires a running MariaDB instance as described above.
54-56: Add concrete MySQL client example for running the init scriptPath
tools/scripts/storage/init_db.sqlhas been verified; include the following snippet in the README:You can choose to set up the database table manually by using the SQL script `tools/scripts/storage/init_db.sql` from the project root. +For example: +```shell +mysql -h 127.0.0.1 -P 3306 -u spider -p"$MARIADB_PASSWORD" spider_storage < tools/scripts/storage/init_db.sql +```
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
💡 Knowledge Base configuration:
- MCP integration is disabled by default for public repositories
- Jira integration is disabled by default for public repositories
- Linear integration is disabled by default for public repositories
You can enable these sources in your CodeRabbit configuration.
📒 Files selected for processing (1)
python/spider-py/README.md(1 hunks)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (3)
- GitHub Check: non-storage-unit-tests (ubuntu-24.04)
- GitHub Check: non-storage-unit-tests (ubuntu-22.04)
- GitHub Check: lint
🔇 Additional comments (2)
python/spider-py/README.md (2)
28-30: Clear split between storage and non‑storage tests — looks good.
The categorisation is clear and aligns with the CI intent.
64-66: All‑tests section reads well.
No blocking issues.
There was a problem hiding this comment.
Actionable comments posted: 0
♻️ Duplicate comments (1)
python/spider-py/README.md (1)
39-53: Harden the Docker example: avoid weak creds, bind locally, pin image, add healthcheck, and fix header grammar.
The current example exposes the DB on all interfaces, uses a trivial password, allows empty root password, and floats onlatest. Recommend the following changes.-### Setup MariaDB for Storage Unit Tests +### Set up MariaDB for storage unit tests @@ ```shell -docker run \ - --detach \ - --rm \ - --name spider-storage \ - --env MARIADB_USER=spider \ - --env MARIADB_PASSWORD=password \ - --env MARIADB_DATABASE=spider-storage \ - --env MARIADB_ALLOW_EMPTY_ROOT_PASSWORD=true \ - --publish 3306:3306 mariadb:latest +docker run \ + --detach \ + --rm \ + --name spider-storage \ + -e MARIADB_USER=spider \ + -e MARIADB_PASSWORD=spider_pwd_change_me \ + -e MARIADB_DATABASE=spider_storage \ + -e MARIADB_RANDOM_ROOT_PASSWORD=1 \ + --publish 127.0.0.1:3306:3306 \ + --health-cmd='mariadb-admin ping -h localhost -u $$MARIADB_USER --password=$$MARIADB_PASSWORD || exit 1' \ + --health-interval=5s --health-timeout=3s --health-retries=20 \ + mariadb:11.4</blockquote></details> </blockquote></details> <details> <summary>🧹 Nitpick comments (2)</summary><blockquote> <details> <summary>python/spider-py/README.md (2)</summary><blockquote> `31-38`: **Offer a direct pytest invocation (nice-to-have).** Add an alternative command for contributors not using Task. You could append: ```diff To run all non-storage unit tests: ```shell task test:spider-py-non-storage-unit-tests
+Alternatively, without Task:
+
+shell +uv run pytest -m "not storage" +--- `55-59`: **Tighten wording and capitalisation; note migrations and script as alternatives.** Small phrasing polish for accuracy and style. ```diff -The database tables will be automatically created by the spawned spider scheduler. - -You can choose to set up the database table manually by using the SQL script -`tools/scripts/storage/init_db.sql` from the project root. +The database tables will be created automatically by the Spider scheduler or by migrations on first use. + +You can also initialise the schema manually using the SQL script `tools/scripts/storage/init_db.sql` from the project root.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
💡 Knowledge Base configuration:
- MCP integration is disabled by default for public repositories
- Jira integration is disabled by default for public repositories
- Linear integration is disabled by default for public repositories
You can enable these sources in your CodeRabbit configuration.
📒 Files selected for processing (1)
python/spider-py/README.md(1 hunks)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (3)
- GitHub Check: non-storage-unit-tests (ubuntu-24.04)
- GitHub Check: non-storage-unit-tests (ubuntu-22.04)
- GitHub Check: lint
🔇 Additional comments (3)
python/spider-py/README.md (3)
28-30: Clarity of test categories looks good.
The split and brief explanation are clear.
60-69: Storage test section reads well.
Clear and actionable.
78-79: OK to reiterate requirement.
The reminder about a running MariaDB instance is helpful.
There was a problem hiding this comment.
Just curious: what's the plan for the storage tests?
There was a problem hiding this comment.
You mean storage tests in GH workflows?
There was a problem hiding this comment.
Yes. Are we planning to add this to our workflows?
There was a problem hiding this comment.
There were talks about it, but no concrete plans. I don't think this is very hard to implement, but we do need to sync the test jdbc url across C++ and Python as they are different right now.
There was a problem hiding this comment.
Sure. I think we should set it up ASAP. Let's prioritize this after ur're done with compression integration.
Co-authored-by: Lin Zhihao <59785146+LinZhihao-723@users.noreply.github.com>
|
|
||
| @abstractmethod | ||
| def submit_jobs( | ||
| self, driver_id: core.DriverId, task_graphs: Sequence[core.TaskGraph] |
There was a problem hiding this comment.
iiuc, this driver_id is actually the ID of the client. Shall we name it explicitly as the client ID?
There was a problem hiding this comment.
For now I haven't added job submission inside a task via client.TaskContext because we don't need it for the clp integration. However, a fully functional TaskContext supports it, as it does not in C++ client.
Co-authored-by: Lin Zhihao <59785146+LinZhihao-723@users.noreply.github.com>
…er into python-job-submission
There was a problem hiding this comment.
Actionable comments posted: 0
♻️ Duplicate comments (5)
python/spider-py/src/spider_py/storage/mariadb_storage.py (5)
169-171: Rollback on any exception, not just mariadb.Error.Non-driver errors (e.g., type bugs in param builders) would skip rollback.
except mariadb.Error as e: self._conn.rollback() raise StorageError(str(e)) from e + except Exception as e: + self._conn.rollback() + raise StorageError(str(e)) from e
295-301: Serialise outputtypeto a stable string before insert.Persisting arbitrary Python objects risks DB driver errors and format drift. Store a canonical string.
- output_params.append( - (task_ids[graph_index][task_index].bytes, position, task_output.type) - ) + output_params.append( + (task_ids[graph_index][task_index].bytes, position, str(task_output.type)) + )
323-331: Serialise inputtypewhen inserting TaskInputData.Keep storage schema stable and driver-friendly.
- task_input.type, + str(task_input.type), task_input.value.bytes,
350-361: Serialise inputtypeand normalisevalueto a DB-safe primitive.Cast type to string; for value, unwrap simple wrappers via
.valuewhen present.- input_value_params.append( - ( - task_ids[graph_index][task_index].bytes, - position, - task_input.type, - task_input.value, - ) - ) + input_value_params.append( + ( + task_ids[graph_index][task_index].bytes, + position, + str(task_input.type), + getattr(task_input.value, "value", task_input.value), + ) + )Also update the return type hint to reflect accepted DB primitives (bytes/str):
- ) -> list[tuple[bytes, int, str, bytes]]: + ) -> list[tuple[bytes, int, str, bytes | str]]:
387-392: Serialise inputtypefor input-output refs.Avoid storing non-primitive objects in
typecolumn.- task_graph.tasks[input_output_ref.input_task_index] - .task_inputs[input_output_ref.input_position] - .type, + str( + task_graph.tasks[input_output_ref.input_task_index] + .task_inputs[input_output_ref.input_position] + .type + ),
🧹 Nitpick comments (2)
python/spider-py/src/spider_py/storage/storage.py (1)
24-30: Clarify ordering and atomicity in the docstring.Explicitly state that returned job IDs preserve the input order and that implementations should commit atomically.
""" - Submits jobs to the storage. + Submits jobs to the storage. :param driver_id: Driver id. :param task_graphs: Task graphs to submit. - :return: A list of job IDs representing the submitted jobs. + :return: A list of job IDs in the same order as `task_graphs`. :raises StorageError: If the storage operations fail. + Note: Implementations should perform this as a single atomic transaction + (all-or-nothing semantics). """python/spider-py/src/spider_py/storage/mariadb_storage.py (1)
68-72: Fix lint nits and make destructor safe; consider explicit close/context manager.Add the required blank line after the class docstring, annotate and docstring
__del__, and guard shutdown-time errors. Prefer an explicitclose()and context manager for deterministic cleanup.class MariaDBStorage(Storage): """MariaDB Storage class.""" - def __del__(self): - self._conn.close() + + def __del__(self) -> None: + """Best-effort close to avoid leaking the connection at GC time.""" + try: + self._conn.close() + except Exception: + # Interpreter shutdown or partially-initialised object; ignore. + passOptional (outside this hunk): add deterministic cleanup.
def close(self) -> None: """Close the DB connection.""" self._conn.close() def __enter__(self) -> "MariaDBStorage": return self def __exit__(self, exc_type, exc, tb) -> None: self.close()
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
💡 Knowledge Base configuration:
- MCP integration is disabled by default for public repositories
- Jira integration is disabled by default for public repositories
- Linear integration is disabled by default for public repositories
You can enable these sources in your CodeRabbit configuration.
📒 Files selected for processing (3)
python/spider-py/pyproject.toml(1 hunks)python/spider-py/src/spider_py/storage/mariadb_storage.py(1 hunks)python/spider-py/src/spider_py/storage/storage.py(1 hunks)
🚧 Files skipped from review as they are similar to previous changes (1)
- python/spider-py/pyproject.toml
🧰 Additional context used
🧬 Code graph analysis (2)
python/spider-py/src/spider_py/storage/storage.py (2)
python/spider-py/src/spider_py/storage/mariadb_storage.py (1)
submit_jobs(90-171)python/spider-py/src/spider_py/core/task_graph.py (1)
TaskGraph(15-163)
python/spider-py/src/spider_py/storage/mariadb_storage.py (4)
python/spider-py/src/spider_py/storage/jdbc_url.py (1)
JdbcParameters(8-16)python/spider-py/src/spider_py/storage/storage.py (3)
Storage(17-30)StorageError(9-14)submit_jobs(21-30)python/spider-py/src/spider_py/core/task_graph.py (1)
TaskGraph(15-163)python/spider-py/src/spider_py/core/task.py (1)
get_state_str(56-61)
🪛 Ruff (0.12.2)
python/spider-py/src/spider_py/storage/mariadb_storage.py
69-69: 1 blank line required after class docstring
Insert 1 blank line after class docstring
(D204)
70-70: Missing return type annotation for special method __del__
Add return type annotation: None
(ANN204)
70-70: Missing docstring in magic method
(D105)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (3)
- GitHub Check: non-storage-unit-tests (ubuntu-24.04)
- GitHub Check: non-storage-unit-tests (ubuntu-22.04)
- GitHub Check: lint
🔇 Additional comments (3)
python/spider-py/src/spider_py/storage/storage.py (1)
17-30: Interface looks solid; types and contract are clear.Clean ABC, precise typing, and helpful docstring. No blockers.
python/spider-py/src/spider_py/storage/mariadb_storage.py (2)
100-168: Good transactional batching and schema write order.Clear separation of insert phases, correct use of executemany, and single commit; helpers improve readability. Nice work.
192-205: No changes needed: Python ≥3.10 is declared in pyproject.toml and CI, so zip(..., strict=True) is compatible.
There was a problem hiding this comment.
Actionable comments posted: 4
♻️ Duplicate comments (2)
python/spider-py/src/spider_py/storage/mariadb_storage.py (2)
172-175: Rollback on any exception, not just mariadb.Error.Non-DB exceptions (type errors in generators, encoding errors, etc.) will currently skip rollback and leave a partial transaction.
except mariadb.Error as e: self._conn.rollback() raise StorageError(str(e)) from e + except Exception as e: + self._conn.rollback() + raise StorageError(str(e)) from e
3-12: Serialise “type” consistently before insert (helper).DB should persist a stable textual representation. Introduce a small helper and use it in all generators that write a
typecolumn.Add this helper near the imports:
def _type_str(t: object) -> str: return t if isinstance(t, str) else getattr(t, "name", str(t))
🧹 Nitpick comments (2)
python/spider-py/src/spider_py/storage/mariadb_storage.py (2)
68-72: Harden del; satisfy Ruff; avoid destructor pitfalls.
- Add the missing blank line after the class docstring.
- Guard against partially constructed instances and ignore close() errors.
- Add a short docstring and return type.
class MariaDBStorage(Storage): - """MariaDB Storage class.""" - def __del__(self): - self._conn.close() + """MariaDB Storage class.""" + + def __del__(self) -> None: + """Close the DB connection if initialised.""" + conn = getattr(self, "_conn", None) + if conn is not None: + try: + conn.close() + except Exception: + # Best-effort cleanup in GC; never raise from __del__ + pass
103-171: Guard executemany calls for empty batches
Wrap thecursor.executemanyinvocations forInsertTask,InsertInputTask,InsertOutputTaskandInsertTaskOutputinif params:checks (or assert non-empty) to prevent passing empty lists to drivers that error on empty batches.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
💡 Knowledge Base configuration:
- MCP integration is disabled by default for public repositories
- Jira integration is disabled by default for public repositories
- Linear integration is disabled by default for public repositories
You can enable these sources in your CodeRabbit configuration.
📒 Files selected for processing (1)
python/spider-py/src/spider_py/storage/mariadb_storage.py(1 hunks)
🧰 Additional context used
🧬 Code graph analysis (1)
python/spider-py/src/spider_py/storage/mariadb_storage.py (4)
python/spider-py/src/spider_py/storage/jdbc_url.py (1)
JdbcParameters(8-16)python/spider-py/src/spider_py/storage/storage.py (3)
Storage(17-30)StorageError(9-14)submit_jobs(21-30)python/spider-py/src/spider_py/core/task_graph.py (1)
TaskGraph(15-163)python/spider-py/src/spider_py/core/task.py (1)
get_state_str(56-61)
🪛 Ruff (0.12.2)
python/spider-py/src/spider_py/storage/mariadb_storage.py
69-69: 1 blank line required after class docstring
Insert 1 blank line after class docstring
(D204)
70-70: Missing return type annotation for special method __del__
Add return type annotation: None
(ANN204)
70-70: Missing docstring in magic method
(D105)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (3)
- GitHub Check: lint
- GitHub Check: non-storage-unit-tests (ubuntu-24.04)
- GitHub Check: non-storage-unit-tests (ubuntu-22.04)
🔇 Additional comments (4)
python/spider-py/src/spider_py/storage/mariadb_storage.py (4)
93-95: Early-return on empty input — good.Keeps a no-op cheap and explicit.
176-209: Task row generation looks solid.
- Good use of strict zip and explicit loops.
- State stringification via
get_state_str()is correct.
170-171: Return type: core.JobId is an alias for UUID
core.JobIdis defined asJobId = UUIDintask_graph.py, so returning rawuuid.UUIDinstances already satisfies theSequence[core.JobId]signature—no wrapping required.Likely an incorrect or invalid review comment.
13-66: Use?placeholders with MariaDB Connector/Python MariaDB’s Python driver uses qmark style (?) for parameters by default (mariadb-corporation.github.io); confirm your test suite exercisesmariadb_storage’sexecutemanycalls against a real MariaDB instance.
| @staticmethod | ||
| def _gen_task_output_insertion_params( | ||
| task_ids: Sequence[Sequence[UUID]], | ||
| task_graphs: Sequence[core.TaskGraph], | ||
| ) -> list[tuple[bytes, int, str]]: | ||
| """ | ||
| Generates parameters for inserting task outputs into the database. | ||
| :param task_ids: The task IDs. | ||
| :param task_graphs: The task graphs. Must be the same length as `task_ids`. | ||
| :return: A list of tuples containing the parameters for each task output. Each tuple | ||
| contains: | ||
| - Task ID. | ||
| - Positional index of the output. | ||
| - Type of the output. | ||
| """ | ||
| output_params = [] | ||
| for graph_index, task_graph in enumerate(task_graphs): | ||
| for task_index, task in enumerate(task_graph.tasks): | ||
| for position, task_output in enumerate(task.task_outputs): | ||
| output_params.append( | ||
| (task_ids[graph_index][task_index].bytes, position, task_output.type) | ||
| ) | ||
| return output_params | ||
|
|
There was a problem hiding this comment.
🛠️ Refactor suggestion
Use stable type strings for task outputs.
Write a string/enum-name for type, not an arbitrary Python object.
for position, task_output in enumerate(task.task_outputs):
output_params.append(
- (task_ids[graph_index][task_index].bytes, position, task_output.type)
+ (task_ids[graph_index][task_index].bytes, position, _type_str(task_output.type))
)Committable suggestion skipped: line range outside the PR's diff.
🤖 Prompt for AI Agents
In python/spider-py/src/spider_py/storage/mariadb_storage.py around lines 282 to
305, the code currently appends task_output.type (an arbitrary Python object) as
the DB "type" field; change this to a stable string representation instead
(e.g., if task_output.type is an Enum use task_output.type.name, otherwise use
str(task_output.type)); update the appended tuple to use that string so the DB
always stores a deterministic, serializable type name.
| @staticmethod | ||
| def _gen_task_input_data_insertion_params( | ||
| task_ids: Sequence[Sequence[UUID]], | ||
| task_graphs: Sequence[core.TaskGraph], | ||
| ) -> list[tuple[bytes, int, str, bytes]]: | ||
| """ | ||
| Generates parameters for inserting task input data into the database. | ||
| :param task_ids: The task IDs. | ||
| :param task_graphs: The task graphs. Must be the same length as `task_ids`. | ||
| :return: A list of tuples containing the parameters for each task input data. Each tuple | ||
| contains: | ||
| - Task ID. | ||
| - Positional index of the input. | ||
| - Type of the input. | ||
| - Input data. | ||
| """ | ||
| input_data_params = [] | ||
| for graph_index, task_graph in enumerate(task_graphs): | ||
| for task_index, task in enumerate(task_graph.tasks): | ||
| for position, task_input in enumerate(task.task_inputs): | ||
| if isinstance(task_input.value, core.TaskInputData): | ||
| input_data_params.append( | ||
| ( | ||
| task_ids[graph_index][task_index].bytes, | ||
| position, | ||
| task_input.type, | ||
| task_input.value.bytes, | ||
| ) | ||
| ) | ||
| return input_data_params | ||
|
|
There was a problem hiding this comment.
🛠️ Refactor suggestion
Use stable type strings for input-data rows.
Match C++/cross-lang storage by serialising type to a string.
input_data_params.append(
(
task_ids[graph_index][task_index].bytes,
position,
- task_input.type,
+ _type_str(task_input.type),
task_input.value.bytes,
)
)Committable suggestion skipped: line range outside the PR's diff.
🤖 Prompt for AI Agents
In python/spider-py/src/spider_py/storage/mariadb_storage.py around lines 306 to
336, the third element of each input-data param tuple currently passes the type
object directly; to match C++/cross-language storage you must serialize the type
to a stable string. Change the code that builds the tuple so the type is passed
as a string (e.g., use str(...) or the canonical name property) instead of the
raw type object, ensuring the function still returns list[tuple[bytes, int, str,
bytes]].
| @staticmethod | ||
| def _gen_task_input_value_insertion_params( | ||
| task_ids: Sequence[Sequence[UUID]], | ||
| task_graphs: Sequence[core.TaskGraph], | ||
| ) -> list[tuple[bytes, int, str, bytes]]: | ||
| """ | ||
| Generates parameters for inserting task input values into the database. | ||
| :param task_ids: The task IDs. | ||
| :param task_graphs: The task graphs. Must be the same length as `task_ids`. | ||
| :return: A list of tuples containing the parameters for each task input value. Each tuple | ||
| contains: | ||
| - Task ID. | ||
| - Positional index of the input. | ||
| - Type of the input. | ||
| - Input value. | ||
| """ | ||
| input_value_params = [] | ||
| for graph_index, task_graph in enumerate(task_graphs): | ||
| for task_index, task in enumerate(task_graph.tasks): | ||
| for position, task_input in enumerate(task.task_inputs): | ||
| if isinstance(task_input.value, core.TaskInputValue): | ||
| input_value_params.append( | ||
| ( | ||
| task_ids[graph_index][task_index].bytes, | ||
| position, | ||
| task_input.type, | ||
| task_input.value, | ||
| ) | ||
| ) | ||
| return input_value_params | ||
|
|
There was a problem hiding this comment.
🛠️ Refactor suggestion
Unwrap TaskInputValue payload and serialise type.
- Persist
typeas a string. - Insert the raw value payload, not the wrapper object. Using
getattr(..., "value", ...)keeps it robust.
- ) -> list[tuple[bytes, int, str, bytes]]:
+ ) -> list[tuple[bytes, int, str, bytes | bytearray | memoryview | str]]:
@@
input_value_params.append(
(
task_ids[graph_index][task_index].bytes,
position,
- task_input.type,
- task_input.value,
+ _type_str(task_input.type),
+ getattr(task_input.value, "value", task_input.value),
)
)Committable suggestion skipped: line range outside the PR's diff.
🤖 Prompt for AI Agents
In python/spider-py/src/spider_py/storage/mariadb_storage.py around lines 337 to
367, the insertion params currently pass the TaskInputValue wrapper and a
non-string type; change it to persist the raw payload and a stringified type:
use getattr(task_input.value, "value", task_input.value) to extract the inner
payload and convert task_input.type to str(task_input.type) before appending;
ensure the payload is converted to the bytes expected by the DB param (e.g.,
encode if a str or serialize to bytes) so the appended tuple matches (bytes,
int, str, bytes).
| @staticmethod | ||
| def _gen_task_input_output_ref_insertion_params( | ||
| task_ids: Sequence[Sequence[UUID]], | ||
| task_graphs: Sequence[core.TaskGraph], | ||
| ) -> list[tuple[bytes, int, str, bytes, int]]: | ||
| """ | ||
| Generates parameters for inserting task input output refs into the database. | ||
| :param task_ids: The task IDs. | ||
| :param task_graphs: The task graphs. Must be the same length as `task_ids`. | ||
| :return: A list of tuples containing the parameters for each task input output ref. Each | ||
| tuple contains: | ||
| - Input task ID. | ||
| - Positional index of the input. | ||
| - Type of the input. | ||
| - Output task ID. | ||
| - Positional index of the output. | ||
| """ | ||
| input_output_params = [] | ||
| for graph_index, task_graph in enumerate(task_graphs): | ||
| for input_output_ref in task_graph.task_input_output_refs: | ||
| input_output_params.append( | ||
| ( | ||
| task_ids[graph_index][input_output_ref.input_task_index].bytes, | ||
| input_output_ref.input_position, | ||
| task_graph.tasks[input_output_ref.input_task_index] | ||
| .task_inputs[input_output_ref.input_position] | ||
| .type, | ||
| task_ids[graph_index][input_output_ref.output_task_index].bytes, | ||
| input_output_ref.output_position, | ||
| ) | ||
| ) | ||
| return input_output_params |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Serialise input ref ‘type’ to a string.
Avoid persisting Python objects in type column.
(
task_ids[graph_index][input_output_ref.input_task_index].bytes,
input_output_ref.input_position,
- task_graph.tasks[input_output_ref.input_task_index]
- .task_inputs[input_output_ref.input_position]
- .type,
+ _type_str(
+ task_graph.tasks[input_output_ref.input_task_index]
+ .task_inputs[input_output_ref.input_position]
+ .type
+ ),
task_ids[graph_index][input_output_ref.output_task_index].bytes,
input_output_ref.output_position,
)Committable suggestion skipped: line range outside the PR's diff.
🤖 Prompt for AI Agents
In python/spider-py/src/spider_py/storage/mariadb_storage.py around lines 368 to
399, the code appends the input ref 'type' object directly into the DB parameter
tuple; change that to serialize the type to a string (e.g., wrap the expression
with str(...)) before appending so you persist a plain string instead of a
Python object.
There was a problem hiding this comment.
Actionable comments posted: 2
♻️ Duplicate comments (3)
python/spider-py/src/spider_py/core/task.py (1)
78-88: State string values are known inconsistent (tracked); consider freezing mapsThe "success/fail/cancel" forms differ from “succeeded/failed/cancelled”. Already tracked in issue #217; no action here. Optionally mark maps Final to prevent accidental mutation.
+from typing import Final @@ -_StateToStrMap = { +_StateToStrMap: Final[dict[TaskState, str]] = { @@ -_StrToStateMap = {value: key for key, value in _StateToStrMap.items()} +_StrToStateMap: Final[dict[str, TaskState]] = {value: key for key, value in _StateToStrMap.items()}python/spider-py/src/spider_py/storage/jdbc_url.py (1)
35-41: Do not echo credentials in exceptionsError strings include the (possibly credentialed) URL. Redact sensitive params before raising.
- msg = "Invalid JDBC URL: {}. Missing {}." + msg = "Invalid JDBC URL: {}. Missing {}." @@ - if not parsed.scheme: - raise ValueError(msg.format(url, "protocol")) + if not parsed.scheme: + raise ValueError(msg.format(_redact_url(orig_url), "protocol")) if not parsed.hostname: - raise ValueError(msg.format(url, "host")) + raise ValueError(msg.format(_redact_url(orig_url), "host")) if not parsed.path or not parsed.path.lstrip("/"): - raise ValueError(msg.format(url, "database")) + raise ValueError(msg.format(_redact_url(orig_url), "database"))Add helper above parse_jdbc_url:
+def _redact_url(url: str) -> str: + base, sep, query = url.partition("?") + if not sep: + return url + q = urllib.parse.parse_qsl(query, keep_blank_values=True) + redacted = [(k, "***") if k.lower() in {"password", "pwd", "pass"} else (k, v) for k, v in q] + return f"{base}?{urllib.parse.urlencode(redacted)}"python/spider-py/src/spider_py/storage/mariadb_storage.py (1)
178-181: Rollback on any exception, not just mariadb.ErrorType/value errors during param generation will bypass rollback and leave an open transaction.
except mariadb.Error as e: self._conn.rollback() raise StorageError(str(e)) from e + except Exception as e: + self._conn.rollback() + raise StorageError(str(e)) from e
🧹 Nitpick comments (4)
python/spider-py/src/spider_py/core/task.py (1)
56-76: Make from_str tolerant to case/whitespaceImproves robustness for inputs coming from DB or external clients.
@staticmethod def from_str(state_str: str) -> TaskState: @@ - state = _StrToStateMap.get(state_str) + key = state_str.strip().lower() + state = _StrToStateMap.get(key) if state is not None: return state msg = f"Invalid task state string: {state_str}" raise ValueError(msg)python/spider-py/src/spider_py/storage/jdbc_url.py (1)
48-55: Normalise protocol to always include “jdbc:”Downstream code can rely on a single form.
- return JdbcParameters( - protocol=protocol_prefix + parsed.scheme, + return JdbcParameters( + protocol=f"{_JdbcPrefix}{parsed.scheme}", host=parsed.hostname, port=parsed.port, database=database, user=user, password=password, )python/spider-py/src/spider_py/storage/mariadb_storage.py (2)
71-77: Avoid exceptions in del; don’t promise raising StorageError hereFinalizers shouldn’t raise; guard close and update the docstring. Consider offering an explicit close() or context manager.
- def __del__(self) -> None: - """ - Closes the connection to the MariaDB database. - :raises StorageError: If closing the connection fails. - """ - self._conn.close() + def __del__(self) -> None: + """Closes the connection to the MariaDB database.""" + try: + self._conn.close() + except Exception: + pass
140-144: Guard executemany when params could be emptySome graphs may have zero outputs; executemany([]) can error on some drivers.
- cursor.executemany( - InsertTaskOutput, - self._gen_task_output_insertion_params(task_ids, task_graphs), - ) + output_params = self._gen_task_output_insertion_params(task_ids, task_graphs) + if output_params: + cursor.executemany(InsertTaskOutput, output_params)
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
💡 Knowledge Base configuration:
- MCP integration is disabled by default for public repositories
- Jira integration is disabled by default for public repositories
- Linear integration is disabled by default for public repositories
You can enable these sources in your CodeRabbit configuration.
📒 Files selected for processing (3)
python/spider-py/src/spider_py/core/task.py(3 hunks)python/spider-py/src/spider_py/storage/jdbc_url.py(1 hunks)python/spider-py/src/spider_py/storage/mariadb_storage.py(1 hunks)
🧰 Additional context used
🧬 Code graph analysis (1)
python/spider-py/src/spider_py/storage/mariadb_storage.py (3)
python/spider-py/src/spider_py/storage/jdbc_url.py (1)
JdbcParameters(8-16)python/spider-py/src/spider_py/storage/storage.py (3)
Storage(17-30)StorageError(9-14)submit_jobs(21-30)python/spider-py/src/spider_py/core/task_graph.py (1)
TaskGraph(15-163)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (3)
- GitHub Check: non-storage-unit-tests (ubuntu-24.04)
- GitHub Check: non-storage-unit-tests (ubuntu-22.04)
- GitHub Check: lint
🔇 Additional comments (8)
python/spider-py/src/spider_py/core/task.py (2)
3-4: Good: future annotations enabledThis reduces runtime import cycles and speeds up typing.
31-31: Type surface correctly widened to include TaskInputValueMatches storage paths that handle bytes payloads.
python/spider-py/src/spider_py/storage/mariadb_storage.py (6)
95-101: Early return on empty input is goodPrevents opening a transaction unnecessarily.
109-177: Nice: single-transaction batch insert and readable helpersThe refactor into generator helpers improves clarity and aligns with prior feedback.
182-215: Task state serialisation via str(state) is correctMatches core.TaskState.str and DB enum mapping.
312-342: Input-data branch looks correctType check vs TaskInputData and storing UUID bytes is consistent.
343-373: Input-value branch looks correctStores raw bytes for values.
374-405: Input→Output ref generation is clearJoins IDs/positions correctly.
| try: | ||
| self._conn = mariadb.connect( | ||
| host=params.host, | ||
| port=params.port, | ||
| user=params.user, | ||
| password=params.password, | ||
| database=params.database, | ||
| ) |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Don’t pass None-valued connection args; add a sane timeout
Prevents type errors in drivers and avoids indefinite hangs.
- self._conn = mariadb.connect(
- host=params.host,
- port=params.port,
- user=params.user,
- password=params.password,
- database=params.database,
- )
+ conn_kwargs = {"host": params.host, "database": params.database, "connect_timeout": 5}
+ if params.port is not None:
+ conn_kwargs["port"] = params.port
+ if params.user is not None:
+ conn_kwargs["user"] = params.user
+ if params.password is not None:
+ conn_kwargs["password"] = params.password
+ self._conn = mariadb.connect(**conn_kwargs)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| try: | |
| self._conn = mariadb.connect( | |
| host=params.host, | |
| port=params.port, | |
| user=params.user, | |
| password=params.password, | |
| database=params.database, | |
| ) | |
| try: | |
| conn_kwargs = { | |
| "host": params.host, | |
| "database": params.database, | |
| "connect_timeout": 5, | |
| } | |
| if params.port is not None: | |
| conn_kwargs["port"] = params.port | |
| if params.user is not None: | |
| conn_kwargs["user"] = params.user | |
| if params.password is not None: | |
| conn_kwargs["password"] = params.password | |
| self._conn = mariadb.connect(**conn_kwargs) |
🤖 Prompt for AI Agents
In python/spider-py/src/spider_py/storage/mariadb_storage.py around lines 84-91,
the mariadb.connect call is passing parameters that may be None and lacks a
connection timeout; update it to build a kwargs dict only including non-None
values for host, port, user, password, database (so you don't pass None to the
driver) and add a sane timeout like connect_timeout=10 (or connection_timeout if
your mariadb client uses that API) before calling mariadb.connect(**kwargs).
There was a problem hiding this comment.
As discussed offline, let's add a sentence to explain how to set up the DB. We don't have the scheduler ready as it's not the integration test.
Co-authored-by: Lin Zhihao <59785146+LinZhihao-723@users.noreply.github.com>
LinZhihao-723
left a comment
LinZhihao-723
left a comment
There was a problem hiding this comment.
For the PR title, how about:
feat(spider-py): Add job submission support for MariaDB storage backend.
sitaowang1998
changed the title
2 participants
Description
This PR adds batch job submission of a list of task graphs in MariaDB storage backend.
This PR also adds unit tests for new storage functions, and add new tasks to run storage and non-storage unit tests. The GitHub workflows are updated to run only non-storage unit tests.
Checklist
breaking change.
Validation performed
Summary by CodeRabbit
New Features
Documentation
Tests
Chores