fix(api): add bulk_upsert method and unit tests for opensearch BaseOSDB class by vkuznet · Pull Request #990 · DIRACGrid/diracx

vkuznet · 2026-07-14T18:14:31Z

This PR fixes issue #401 by introducing the bulk_upsert API to BaseOSDB class. It is also complement by full set of unit tests for BaseOSDB class which were missing. The unit test introduces mock client and different classes for different use cases.

Depends on #1007

Please note: it is my first PR and I'm happy to adjust it according to requirements/guidelines of DiracX community.

…DB class

read-the-docs-community · 2026-07-16T13:04:34Z

Documentation build overview

📚 diracx | 🛠️ Build #34477584 | 📁 Comparing 7dc18e9 against latest (19cb080)

🔍 Preview build

162 files changed · + 10 added · ± 145 modified · - 7 deleted

+ Added

± Modified

- Deleted

fstagni

In the original issue (#401) it was mentioned a specific use case. Can you already verify if this fix covers that? You can add it directly in this PR.

vkuznet · 2026-08-06T14:13:25Z

In the original issue (#401) it was mentioned a specific use case. Can you already verify if this fix covers that? You can add it directly in this PR.

@fstagni regarding issue #401. I checked the main repository code and what is described in this issue is not there the issue is based on commit outside of any branch, most likely in fork outside of the repository

diracx/diracx-routers/src/diracx/routers/jobs/status.py

Lines 174 to 180 in 7c260cc

    
           # TODO: can we upsert to multiple documents? 
        
           for job_id, p_updates_ in param_updates.items(): 
        
               if p_updates_: 
        
                   await job_parameters_db.upsert( 
        
                       int(job_id), 
        
                       p_updates_, 
        
                   )

). Therefore, before adding anything to this PR with usage of bulk insert I need to know if it is in fact required by code from main repository, see utils.py from main branch

fstagni · 2026-08-07T09:29:26Z

Hi, since #410 was created, the mentioned code has been moved and refactored, and can now be found in https://github.com/DIRACGrid/diracx/blob/main/diracx-logic/src/diracx/logic/jobs/status.py#L643. The issue anyway still holds, can you have a look?

for more information, see https://pre-commit.ci

…mentation for MockOSDBMixin class

… declarations

aldbr

Thanks @vkuznet!
Just a few additional comments, sorry for the delay we take to review PRs...

aldbr · 2026-09-02T09:41:34Z

            response,
        )

+    async def bulk_upsert(


bulk_upsert would need to overridden within job_parameters_db because we are inserting a JobID and a timestamp:

diracx/diracx-db/src/diracx/db/os/job_parameters.py

Lines 37 to 43 in 202f84c

def upsert(self, vo, doc_id, document):

document = {

"JobID": doc_id,

"timestamp": int(datetime.now(tz=UTC).timestamp() * 1000),

**document,

}

return super().upsert(vo, doc_id, document)

Here it would not work I think (and it looks like it's not spotted within the tests).

I actually wonder whether upsert is useful now that we have bulk_upsert.
I would suggest we just drop upsert and replace it everywhere with bulk_upsert, what do you think?

This is fixed now in 6d2984c

But I against dropping upsert in favor of bulk_upsert for simple reason. The code, e.g. https://github.com/DIRACGrid/diracx/blob/main/diracx-logic/src/diracx/logic/jobs/status.py#L225, uses external for loop and insert each document individually while going through that loop. To use bulk_upsert would require in this place to either use generator or allocate more memory to collect all documents and then insert them in bulk. There are cases when one API is preferable vs another. Since I can't find usage of generators I think the upsert has its place in a code.

aldbr · 2026-09-02T09:46:50Z

            if new_application:
                job_data["ApplicationStatus"] = new_application

            await job_parameters_db.upsert(res["VO"], job_id, {"Status": new_status})


bulk_upsert could (should) be used here too I think

It can but it would either refactor code to use generators or use additional memory allocation to collect all documents. Either task seems beyond this PR scope.

aldbr · 2026-09-02T09:53:59Z

+            self.client,
+            actions,
+            raise_on_error=False,
+            raise_on_exception=False,


I'm just wondering what happens if there is a connection issue with the DB and no exception is raised.
I guess you would get 0 success, N errors but would get any information to know that there is an issue with the DB itself?

Also I am wondering whether it would make sense to use max_retries and initial_backoff:
https://github.com/opensearch-project/opensearch-py/blob/213b7d6b2890c19bc83ebce0a9886d7571760240/opensearchpy/_async/helpers/actions.py#L188-L192

in order to use this features it would be desired to have them configurable rather using hard-coded defaults. At the moment (based on my limited scope of the code) I don't know how configuration work and if desired this can be done through a separate issue/PR.

I'm just wondering what happens if there is a connection issue with the DB and no exception is raised. I guess you would get 0 success, N errors but would get any information to know that there is an issue with the DB itself?

this is up to upstream code. Since it is external we can't reliably tell what is current and future functionality would be.

aldbr · 2026-09-02T10:50:07Z

+    if errors:
+        for error in errors:
+            logger.error("bulk insert error %s", error)
+        raise DocumentUpsertError("Failed to perform bulk insert operation")


I assume this piece of code is generic because will be reused every time there is an error.
Wouldn't it make sense to raise the DocumentUpsertError from diracx-db itself? So that this part of the code only lives in db/os/utils and is automatically reused by the callers

this is architectural choice, your suggestion to through exception in base bulk_upsert API while mine is to pass errors to upstream and let this code decide what to do. If you can provide specific use-case when exception is better I'll be happy to move this part down the stream and through exception in a base class.

aldbr · 2026-09-02T10:58:19Z

Isn't there some duplication with https://github.com/DIRACGrid/diracx/blob/d72016500fb6cfd73350aa7b2582c5dd2cdcf2d2/diracx-db/tests/opensearch/test_search.py?

…ing it in conftest module

vkuznet added 2 commits July 14, 2026 14:02

fix(api): add bulk_upsert method and unit tests for opensearch BaseOS…

586d311

…DB class

fix(api): adjust failed unit tests

35dd659

aldbr linked an issue Jul 15, 2026 that may be closed by this pull request

Support upserts to multiple documents for opensearch #401

Open

Merge branch 'main' into fix/issue-401

fe52fc2

fstagni reviewed Jul 31, 2026

View reviewed changes

Comment thread diracx-db/src/diracx/db/os/utils.py

vkuznet self-assigned this Aug 6, 2026

aldbr requested a review from fstagni August 6, 2026 14:36

vkuznet and others added 11 commits August 10, 2026 08:37

Merge branch 'main' into fix/issue-401

14f9515

Merge branch 'main' into fix/issue-401

cfb5e53

Merge branch 'main' into fix/issue-401

f22047f

Merge branch 'main' into fix/issue-401

0775e3d

Merge branch 'main' into fix/issue-401

0886112

[pre-commit.ci] auto fixes from pre-commit.com hooks

497583a

for more information, see https://pre-commit.ci

fix(imports): sort import errors

0f15cd2

refactor: move common mock_client to diracx-testing and use it as needed

392c88b

refactor: put back mock_client into appropriate tests

0ac5dba

refactor: move mock_client into testing module; add bulk_upsert imple…

1c66216

…mentation for MockOSDBMixin class

refactor: use mock_client via common mock_osdb module and conftest.py…

48474ea

… declarations

aldbr requested changes Sep 2, 2026

View reviewed changes

DIRACGridBot marked this pull request as draft September 2, 2026 11:05

vkuznet and others added 4 commits September 9, 2026 13:42

fix: add bulk_upsert into job_parameters

6d2984c

fix: reuse DummyOSDB from diracx.testing

c2b68aa

fix: put diracx.testing.mock_osdb into pyproject.toml instead of keep…

188948d

…ing it in conftest module

Merge branch 'main' into fix/issue-401

7dc18e9

vkuznet requested a review from aldbr September 13, 2026 14:21

	def upsert(self, vo, doc_id, document):
	document = {
	"JobID": doc_id,
	"timestamp": int(datetime.now(tz=UTC).timestamp() * 1000),
	**document,
	}
	return super().upsert(vo, doc_id, document)

Uh oh!

Conversation

vkuznet commented Jul 14, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

read-the-docs-community Bot commented Jul 16, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Documentation build overview

Uh oh!

fstagni left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

vkuznet commented Aug 6, 2026

Uh oh!

fstagni commented Aug 7, 2026

Uh oh!

aldbr left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

vkuznet commented Jul 14, 2026 •

edited

Loading

read-the-docs-community Bot commented Jul 16, 2026 •

edited

Loading