Page MenuHomeSoftware Heritage
Differential D7446

Install app_label on add_forge_now model classes
ClosedPublic
Actions

Authored by ardumont on Mar 29 2022, 9:28 AM.

Details

Reviewers
olasd
Group Reviewers
Reviewers
Maniphest Tasks
T4120: Fix documentation dev build (again)
Commits
rDWAPPS0dc0765eed2e: Install app_label on add_forge_now model classes
Summary

This should alleviate problems in documentation build.

Related to D7445

Diff Detail

Repository
rDWAPPS Web applications
Branch
fix-doc-build
Lint
No Linters Available
Unit
No Unit Test Coverage
Build Status
Buildable 27937
Build 43748: Phabricator diff pipeline on jenkinsJenkins console · Jenkins
Build 43747: arc lint + arc unit

Event Timeline

ardumont created this revision.Mar 29 2022, 9:28 AM
Herald added a reviewer: Reviewers. · View Herald TranscriptMar 29 2022, 9:28 AM
swh-public-ci added a comment.Mar 29 2022, 9:30 AM

Build has FAILED

Patch application report for D7446 (id=26956)

Rebasing onto 4bc8aaf1d9...

Current branch diff-target is up to date.
Changes applied before test
commit 2531af642f005897b208255221a052ac41795c3a
Author: Antoine R. Dumont (@ardumont) <ardumont@softwareheritage.org>
Date:   Tue Mar 29 09:25:24 2022 +0200

    Install app_label on add_forge_now model classes
    
    This should alleviate problems in documentation build.

Link to build: https://jenkins.softwareheritage.org/job/DWAPPS/job/tests-on-diff/1605/
See console output for more information: https://jenkins.softwareheritage.org/job/DWAPPS/job/tests-on-diff/1605/console

Harbormaster returned this revision to the author for changes because remote builds failed.Mar 29 2022, 9:30 AM
Harbormaster failed remote builds in B27872: Diff 26956!
ardumont mentioned this in D7445: roadmap 2022 documentation.Mar 29 2022, 9:38 AM
swh-public-ci added a comment.Mar 29 2022, 10:03 AM

Build has FAILED

Patch application report for D7446 (id=26956)

Rebasing onto 4bc8aaf1d9...

Current branch diff-target is up to date.
Changes applied before test
commit 2531af642f005897b208255221a052ac41795c3a
Author: Antoine R. Dumont (@ardumont) <ardumont@softwareheritage.org>
Date:   Tue Mar 29 09:25:24 2022 +0200

    Install app_label on add_forge_now model classes
    
    This should alleviate problems in documentation build.

Link to build: https://jenkins.softwareheritage.org/job/DWAPPS/job/tests-on-diff/1606/
See console output for more information: https://jenkins.softwareheritage.org/job/DWAPPS/job/tests-on-diff/1606/console

ardumont updated this revision to Diff 26968.EditedMar 29 2022, 10:47 AM

What a mess... but it works locally now...

olasd added a subscriber: olasd.Mar 29 2022, 10:56 AM

Please just set the app label in each Model's Meta class rather than use inheritance.

ardumont added a comment.Mar 29 2022, 10:58 AM

Please just set the app label in each Model's Meta class rather than use inheritance.

fine.

ardumont added a comment.EditedMar 29 2022, 11:02 AM
In D7446#194325, @ardumont wrote:

Please just set the app label in each Model's Meta class rather than use inheritance.

fine.

well, if that adaptation change (without inheritance) it's willing to work that is...

swh-public-ci added a comment.Mar 29 2022, 11:11 AM

Build was aborted

Patch application report for D7446 (id=26968)

Could not rebase; Attempt merge onto 4bc8aaf1d9...

Updating 4bc8aaf1..c5825f74
Fast-forward
 swh/web/add_forge_now/apps.py                          |  4 +++-
 .../migrations/0004_auto_20220329_0844.py              | 15 +++++++++++++++
 swh/web/add_forge_now/models.py                        | 18 ++++++++++++++++--
 swh/web/inbound_email/management/__init__.py           |  0
 swh/web/inbound_email/management/commands/__init__.py  |  0
 5 files changed, 34 insertions(+), 3 deletions(-)
 create mode 100644 swh/web/add_forge_now/migrations/0004_auto_20220329_0844.py
 create mode 100644 swh/web/inbound_email/management/__init__.py
 create mode 100644 swh/web/inbound_email/management/commands/__init__.py
Changes applied before test
commit c5825f7487c87f87c3ae965bcb38b0a8cd738bba
Author: Antoine R. Dumont (@ardumont) <ardumont@softwareheritage.org>
Date:   Tue Mar 29 09:25:24 2022 +0200

    Install app_label on add_forge_now model classes
    
    This should alleviate problems in documentation build.
    
    This triggers a necessary noop migration otherwise django tools are not happy.

commit 0c1e63d70261d917be5f4c3400772c6864e6cb56
Author: Antoine R. Dumont (@ardumont) <ardumont@softwareheritage.org>
Date:   Tue Mar 29 09:50:12 2022 +0200

    swh.web.inbound_email: Add missing __init__.py file
    
    My hypothesis is that those missing files are making the swh-docs build fail.
    That's the only difference i see between those swh.web folders and the rest.

Link to build: https://jenkins.softwareheritage.org/job/DWAPPS/job/tests-on-diff/1613/
See console output for more information: https://jenkins.softwareheritage.org/job/DWAPPS/job/tests-on-diff/1613/console

Harbormaster returned this revision to the author for changes because remote builds failed.Mar 29 2022, 11:11 AM
Harbormaster failed remote builds in B27884: Diff 26968!
ardumont added a comment.EditedMar 29 2022, 11:23 AM

Build was aborted

*sigh*
opened D7450 for this.

works locally

=========================================================================================== 704 passed, 5 skipped, 18 warnings in 148.75s (0:02:28) ===========================================================================================
ardumont mentioned this in D7448: Reference the new swh.scrubber module.Mar 29 2022, 12:15 PM
ardumont added a comment.Mar 29 2022, 12:29 PM

Please just set the app label in each Model's Meta class rather than use inheritance.

fine.

well, if that adaptation change (without inheritance) it's willing to work that is...

Doing as you suggest plainly refuse to work at all (it was also my initial attempt btw...).

Modifying the model implies a migration even if it's a noop.
Without it, tests fails:

$ pytest -vv -x -s ./swh/web/tests/test_migrations.py
...
FAILED swh/web/tests/test_migrations.py::test_migrations_09_add_visit_status_to_sor_model - django.db.utils.ProgrammingError: relation "add_forge_now_request" already exists
ERROR swh/web/tests/test_migrations.py::test_migrations_09_add_visit_status_to_sor_model - django.db.utils.ProgrammingError: relation "add_forge_now_request" already exists

Triggering the migration creation and then for some reason, django makemigration only creates a
migration script which destroys both table Request and RequestHistory (T.T).

$ django-admin makemigrations --settings swh.web.settings.development
Migrations for 'add_forge_now':
  swh/web/add_forge_now/migrations/0004_auto_20220329_1026.py
    - Remove field request from requesthistory
    - Delete model Request
    - Delete model RequestHistory

(which is fine for the previous failing tests on migration since it's not tested but it's awfully wrong with the actual add-forge-now tests)

$ pytest -vv -s ./swh/web/tests/api/views/test_add_forge_now.py
...

FAILED swh/web/tests/api/views/test_add_forge_now.py::test_add_forge_request_create_success_post[add_forge_data0] - AssertionError: Traceback (most recent call last):
FAILED swh/web/tests/api/views/test_add_forge_now.py::test_add_forge_request_create_success_post[add_forge_data1] - AssertionError: Traceback (most recent call last):
FAILED swh/web/tests/api/views/test_add_forge_now.py::test_add_forge_request_create_success_post[add_forge_data2] - AssertionError: Traceback (most recent call last):
FAILED swh/web/tests/api/views/test_add_forge_now.py::test_add_forge_request_create_success_post[add_forge_data3] - AssertionError: Traceback (most recent call last):
FAILED swh/web/tests/api/views/test_add_forge_now.py::test_add_forge_request_create_success_form_encoded - AssertionError: Traceback (most recent call last):
FAILED swh/web/tests/api/views/test_add_forge_now.py::test_add_forge_request_create_duplicate - AssertionError: Traceback (most recent call last):
FAILED swh/web/tests/api/views/test_add_forge_now.py::test_add_forge_request_update_non_existent - AssertionError: Traceback (most recent call last):
FAILED swh/web/tests/api/views/test_add_forge_now.py::test_add_forge_request_update_empty - AssertionError: Traceback (most recent call last):
FAILED swh/web/tests/api/views/test_add_forge_now.py::test_add_forge_request_update_missing_field - AssertionError: Traceback (most recent call last):
FAILED swh/web/tests/api/views/test_add_forge_now.py::test_add_forge_request_update - AssertionError: Traceback (most recent call last):
FAILED swh/web/tests/api/views/test_add_forge_now.py::test_add_forge_request_update_invalid_new_status - AssertionError: Traceback (most recent call last):
FAILED swh/web/tests/api/views/test_add_forge_now.py::test_add_forge_request_update_status_concurrent - AssertionError: Traceback (most recent call last):
FAILED swh/web/tests/api/views/test_add_forge_now.py::test_add_forge_request_list_anonymous - AssertionError: Traceback (most recent call last):
FAILED swh/web/tests/api/views/test_add_forge_now.py::test_add_forge_request_list_moderator - AssertionError: Traceback (most recent call last):
FAILED swh/web/tests/api/views/test_add_forge_now.py::test_add_forge_request_list_pagination - AssertionError: Traceback (most recent call last):
FAILED swh/web/tests/api/views/test_add_forge_now.py::test_add_forge_request_list_submitter_filtering - AssertionError: Traceback (most recent call last):
FAILED swh/web/tests/api/views/test_add_forge_now.py::test_add_forge_request_get - AssertionError: Traceback (most recent call last):
FAILED swh/web/tests/api/views/test_add_forge_now.py::test_add_forge_request_get_moderator - AssertionError: Traceback (most recent call last):
FAILED swh/web/tests/api/views/test_add_forge_now.py::test_add_forge_request_get_invalid - AssertionError: Traceback (most recent call last):
================================================================================================== 19 failed, 4 passed, 12 warnings in 8.77s ==================================================================================================

This does not make any sense to me since it's an implementation detail which should end up being the same thing...
And i'm really annoyed with this so i'll keep this as is if you don't mind.

I've opened the other diff which should be equivalent D7452 so what i just said makes more sense to you.

ardumont mentioned this in D7452: [wip] [fail] Install app_label on add_forge_now model classes.Mar 29 2022, 12:30 PM
swh-public-ci added a comment.Mar 29 2022, 12:36 PM

Build was aborted

Patch application report for D7446 (id=26968)

Could not rebase; Attempt merge onto 4bc8aaf1d9...

Updating 4bc8aaf1..c5825f74
Fast-forward
 swh/web/add_forge_now/apps.py                          |  4 +++-
 .../migrations/0004_auto_20220329_0844.py              | 15 +++++++++++++++
 swh/web/add_forge_now/models.py                        | 18 ++++++++++++++++--
 swh/web/inbound_email/management/__init__.py           |  0
 swh/web/inbound_email/management/commands/__init__.py  |  0
 5 files changed, 34 insertions(+), 3 deletions(-)
 create mode 100644 swh/web/add_forge_now/migrations/0004_auto_20220329_0844.py
 create mode 100644 swh/web/inbound_email/management/__init__.py
 create mode 100644 swh/web/inbound_email/management/commands/__init__.py
Changes applied before test
commit c5825f7487c87f87c3ae965bcb38b0a8cd738bba
Author: Antoine R. Dumont (@ardumont) <ardumont@softwareheritage.org>
Date:   Tue Mar 29 09:25:24 2022 +0200

    Install app_label on add_forge_now model classes
    
    This should alleviate problems in documentation build.
    
    This triggers a necessary noop migration otherwise django tools are not happy.

commit 0c1e63d70261d917be5f4c3400772c6864e6cb56
Author: Antoine R. Dumont (@ardumont) <ardumont@softwareheritage.org>
Date:   Tue Mar 29 09:50:12 2022 +0200

    swh.web.inbound_email: Add missing __init__.py file
    
    My hypothesis is that those missing files are making the swh-docs build fail.
    That's the only difference i see between those swh.web folders and the rest.

Link to build: https://jenkins.softwareheritage.org/job/DWAPPS/job/tests-on-diff/1617/
See console output for more information: https://jenkins.softwareheritage.org/job/DWAPPS/job/tests-on-diff/1617/console

ardumont updated this revision to Diff 27010.Mar 29 2022, 3:40 PM

Adapt according to suggestion in D7452#194454

swh-public-ci added a comment.Mar 29 2022, 4:02 PM

Build has FAILED

Patch application report for D7446 (id=27010)

Rebasing onto f8e4ad3a51...

Current branch diff-target is up to date.
Changes applied before test
commit b347b3f431406a73ec62c43ff9e8f1850f4b9cfb
Author: Antoine R. Dumont (@ardumont) <ardumont@softwareheritage.org>
Date:   Tue Mar 29 09:25:24 2022 +0200

    Install app_label on add_forge_now model classes
    
    This should alleviate problems in documentation build.
    
    This triggers a necessary noop migration otherwise django tools are not happy.

Link to build: https://jenkins.softwareheritage.org/job/DWAPPS/job/tests-on-diff/1625/
See console output for more information: https://jenkins.softwareheritage.org/job/DWAPPS/job/tests-on-diff/1625/console

Harbormaster returned this revision to the author for changes because remote builds failed.Mar 29 2022, 4:02 PM
Harbormaster failed remote builds in B27926: Diff 27010!
ardumont updated this revision to Diff 27021.Mar 29 2022, 4:44 PM

Fix also app_label name in migration tests.

swh-public-ci added a comment.Mar 29 2022, 5:06 PM

Build was aborted

Patch application report for D7446 (id=27021)

Rebasing onto f8e4ad3a51...

Current branch diff-target is up to date.
Changes applied before test
commit 6e1b49e4344842d4a6d3fadedf57aa050a3dc901
Author: Antoine R. Dumont (@ardumont) <ardumont@softwareheritage.org>
Date:   Tue Mar 29 09:25:24 2022 +0200

    Install app_label on add_forge_now model classes
    
    This should alleviate problems in documentation build.
    
    Also, it turns out the add_forge_now apps was misconfigured. This commit fixes that.

Link to build: https://jenkins.softwareheritage.org/job/DWAPPS/job/tests-on-diff/1632/
See console output for more information: https://jenkins.softwareheritage.org/job/DWAPPS/job/tests-on-diff/1632/console

Harbormaster returned this revision to the author for changes because remote builds failed.Mar 29 2022, 5:06 PM
Harbormaster failed remote builds in B27937: Diff 27021!
ardumont updated this revision to Diff 27027.Mar 29 2022, 5:09 PM

Trigger build again

swh-public-ci added a comment.Mar 29 2022, 5:31 PM

Build was aborted

Patch application report for D7446 (id=27027)

Rebasing onto f8e4ad3a51...

Current branch diff-target is up to date.
Changes applied before test
commit 6e1b49e4344842d4a6d3fadedf57aa050a3dc901
Author: Antoine R. Dumont (@ardumont) <ardumont@softwareheritage.org>
Date:   Tue Mar 29 09:25:24 2022 +0200

    Install app_label on add_forge_now model classes
    
    This should alleviate problems in documentation build.
    
    Also, it turns out the add_forge_now apps was misconfigured. This commit fixes that.

Link to build: https://jenkins.softwareheritage.org/job/DWAPPS/job/tests-on-diff/1634/
See console output for more information: https://jenkins.softwareheritage.org/job/DWAPPS/job/tests-on-diff/1634/console

Harbormaster returned this revision to the author for changes because remote builds failed.Mar 29 2022, 5:31 PM
Harbormaster failed remote builds in B27943: Diff 27027!
ardumont updated this revision to Diff 27057.Mar 29 2022, 6:00 PM

Rebase and finally unstuck the cypress tests (after some heavy breathing [1])

[1] P1321

swh-public-ci added a comment.Mar 29 2022, 6:19 PM

Build is green

Patch application report for D7446 (id=27057)

Rebasing onto f82d436fad...

Current branch diff-target is up to date.
Changes applied before test
commit 0dc0765eed2e6213eb8cd9e015b3593d887aed87
Author: Antoine R. Dumont (@ardumont) <ardumont@softwareheritage.org>
Date:   Tue Mar 29 09:25:24 2022 +0200

    Install app_label on add_forge_now model classes
    
    This should alleviate problems in documentation build.
    
    Also, it turns out the add_forge_now apps was misconfigured. This commit fixes that.

See https://jenkins.softwareheritage.org/job/DWAPPS/job/tests-on-diff/1636/ for more details.

Harbormaster completed remote builds in B27974: Diff 27057.Mar 29 2022, 6:19 PM
ardumont requested review of this revision.Mar 29 2022, 6:19 PM
olasd accepted this revision.Mar 29 2022, 6:22 PM

Thanks!

This revision is now accepted and ready to land.Mar 29 2022, 6:22 PM
Closed by commit rDWAPPS0dc0765eed2e: Install app_label on add_forge_now model classes (authored by ardumont). · Explain WhyMar 30 2022, 9:33 AM
This revision was automatically updated to reflect the committed changes.
ardumont added a commit: rDWAPPS0dc0765eed2e: Install app_label on add_forge_now model classes.
ardumont added a task: T4120: Fix documentation dev build (again).Mar 30 2022, 4:59 PM

Revision Contents
Changeset List

PathSize
swh/
web/
add_forge_now/
6 lines
4 lines
migrations/
2 lines
2 lines
2 lines
17 lines
10 lines
tests/
24 lines

Diff 27021

View Options

swh/web/add_forge_now/__init__.py

Show First 20 Lines • Show All 83 Lines • ▼ Show 20 Lines
) )
from .schema import HASH_ALGORITHMS from .schema import HASH_ALGORITHMS
# Max block size of contents to return # Max block size of contents to return
BULK_BLOCK_CONTENT_LEN_MAX = 10000 BULK_BLOCK_CONTENT_LEN_MAX = 10000
class CassandraStorage: class CassandraStorage:
def __init__(self, hosts, keyspace, objstorage, port=9042, journal_writer=None): def __init__(
self._cql_runner: CqlRunner = CqlRunner(hosts, keyspace, port) self,
hosts,
keyspace,
objstorage,
port=9042,
journal_writer=None,
check_missing=True,
):
"""
Args:
hosts: Seed Cassandra nodes, to start connecting to the cluster
keyspace: Name of the Cassandra database to use
objstorage: Passed as argument to :cls:`ObjStorage`
port: Cassandra port
journal_writer: Passed as argument to :cls:`JournalWriter`
check_missing: Whether ``*_add`` functions will check if an object already
exists in the database before sending it in an INSERT.
``True`` is the default as it is more efficient when there is a moderately
high probability the object is already known, but ``False`` can be
useful to overwrite existing objects (eg. when applying a schema update),
or when the database is known to be mostly empty.
"""
self._hosts = hosts
self._keyspace = keyspace
self._port = port
self._set_cql_runner()
self.journal_writer: JournalWriter = JournalWriter(journal_writer) self.journal_writer: JournalWriter = JournalWriter(journal_writer)
self.objstorage: ObjStorage = ObjStorage(objstorage) self.objstorage: ObjStorage = ObjStorage(objstorage)
self._check_missing = check_missing
def _set_cql_runner(self):
"""Used by tests when they need to reset the CqlRunner"""
self._cql_runner: CqlRunner = CqlRunner(self._hosts, self._keyspace, self._port)
def check_config(self, *, check_write: bool) -> bool: def check_config(self, *, check_write: bool) -> bool:
self._cql_runner.check_read() self._cql_runner.check_read()
return True return True
def _content_get_from_hash(self, algo, hash_) -> Iterable: def _content_get_from_hash(self, algo, hash_) -> Iterable:
"""From the name of a hash algorithm and a value of that hash, """From the name of a hash algorithm and a value of that hash,
Show All 10 Lines def _content_get_from_hash(self, algo, hash_) -> Iterable:
for row in res: for row in res:
# re-check the the hash (in case of murmur3 collision) # re-check the the hash (in case of murmur3 collision)
if getattr(row, algo) == hash_: if getattr(row, algo) == hash_:
yield row yield row
def _content_add(self, contents: List[Content], with_data: bool) -> Dict: def _content_add(self, contents: List[Content], with_data: bool) -> Dict:
# Filter-out content already in the database. # Filter-out content already in the database.
if self._check_missing:
contents = [ contents = [
c for c in contents if not self._cql_runner.content_get_from_pk(c.to_dict()) c
for c in contents
if not self._cql_runner.content_get_from_pk(c.to_dict())
] ]
if with_data: if with_data:
# First insert to the objstorage, if the endpoint is # First insert to the objstorage, if the endpoint is
# `content_add` (as opposed to `content_add_metadata`). # `content_add` (as opposed to `content_add_metadata`).
# Must add to the objstorage before the DB and journal. Otherwise: # Must add to the objstorage before the DB and journal. Otherwise:
# 1. in case of a crash the DB may "believe" we have the content, but # 1. in case of a crash the DB may "believe" we have the content, but
# we didn't have time to write to the objstorage before the crash # we didn't have time to write to the objstorage before the crash
Show All 12 Lines def _content_add(self, contents: List[Content], with_data: bool) -> Dict:
# Check for sha1 or sha1_git collisions. This test is not atomic # Check for sha1 or sha1_git collisions. This test is not atomic
# with the insertion, so it won't detect a collision if both # with the insertion, so it won't detect a collision if both
# contents are inserted at the same time, but it's good enough. # contents are inserted at the same time, but it's good enough.
# #
# The proper way to do it would probably be a BATCH, but this # The proper way to do it would probably be a BATCH, but this
# would be inefficient because of the number of partitions we # would be inefficient because of the number of partitions we
# need to affect (len(HASH_ALGORITHMS)+1, which is currently 5) # need to affect (len(HASH_ALGORITHMS)+1, which is currently 5)
if self._check_missing:
for algo in {"sha1", "sha1_git"}: for algo in {"sha1", "sha1_git"}:
collisions = [] collisions = []
# Get tokens of 'content' rows with the same value for # Get tokens of 'content' rows with the same value for
# sha1/sha1_git # sha1/sha1_git
rows = self._content_get_from_hash(algo, content.get_hash(algo)) rows = self._content_get_from_hash(algo, content.get_hash(algo))
for row in rows: for row in rows:
if getattr(row, algo) != content.get_hash(algo): if getattr(row, algo) != content.get_hash(algo):
# collision of token(partition key), ignore this # collision of token(partition key), ignore this
# row # row
continue continue
for other_algo in HASH_ALGORITHMS: for other_algo in HASH_ALGORITHMS:
if getattr(row, other_algo) != content.get_hash(other_algo): if getattr(row, other_algo) != content.get_hash(other_algo):
# This hash didn't match; discard the row. # This hash didn't match; discard the row.
collisions.append( collisions.append(
{k: getattr(row, k) for k in HASH_ALGORITHMS} {k: getattr(row, k) for k in HASH_ALGORITHMS}
) )
if collisions: if collisions:
collisions.append(content.hashes()) collisions.append(content.hashes())
raise HashCollision(algo, content.get_hash(algo), collisions) raise HashCollision(algo, content.get_hash(algo), collisions)
douarddaUnsubmitted
Not Done
Inline Actions

why is the whole collision detection block is now run only if check_missing is True?

douardda: why is the whole collision detection block is now run only if check_missing is True?
vlorentzAuthorUnsubmitted
Done
Inline Actions

Because, if an object already exists with the exact same hashes and it was not filtered out initially, then self._content_get_from_hash(algo, content.get_hash(algo)) will return it and it would be detected as a collision.

I could filter the results of self._content_get_from_hash, but I don't think it's worth it, as _check_missing is pretty niche anyway.

vlorentz: Because, if an object already exists with the exact same hashes and it was not filtered out…
vlorentzAuthorUnsubmitted
Done
Inline Actions

And collision detection in Cassandra isn't guaranteed anyway, because two colliding objects may be inserted at the same time and Cassandra is only eventually consistent.

vlorentz: And collision detection in Cassandra isn't guaranteed anyway, because two colliding objects may…
(token, insertion_finalizer) = self._cql_runner.content_add_prepare( (token, insertion_finalizer) = self._cql_runner.content_add_prepare(
ContentRow(**remove_keys(content.to_dict(), ("data",))) ContentRow(**remove_keys(content.to_dict(), ("data",)))
) )
# Then add to index tables # Then add to index tables
for algo in HASH_ALGORITHMS: for algo in HASH_ALGORITHMS:
self._cql_runner.content_index_add_one(algo, content, token) self._cql_runner.content_index_add_one(algo, content, token)
▲ Show 20 Lines • Show All 136 Lines • ▼ Show 20 Linesclass CassandraStorage:
def content_get_random(self) -> Sha1Git: def content_get_random(self) -> Sha1Git:
content = self._cql_runner.content_get_random() content = self._cql_runner.content_get_random()
assert content, "Could not find any content" assert content, "Could not find any content"
return content.sha1_git return content.sha1_git
def _skipped_content_add(self, contents: List[SkippedContent]) -> Dict: def _skipped_content_add(self, contents: List[SkippedContent]) -> Dict:
# Filter-out content already in the database. # Filter-out content already in the database.
if self._check_missing:
contents = [ contents = [
c c
for c in contents for c in contents
if not self._cql_runner.skipped_content_get_from_pk(c.to_dict()) if not self._cql_runner.skipped_content_get_from_pk(c.to_dict())
] ]
self.journal_writer.skipped_content_add(contents) self.journal_writer.skipped_content_add(contents)
for content in contents: for content in contents:
# Compute token of the row in the main table # Compute token of the row in the main table
(token, insertion_finalizer) = self._cql_runner.skipped_content_add_prepare( (token, insertion_finalizer) = self._cql_runner.skipped_content_add_prepare(
SkippedContentRow.from_dict({"origin": None, **content.to_dict()}) SkippedContentRow.from_dict({"origin": None, **content.to_dict()})
) )
Show All 15 Lines def skipped_content_missing(
self, contents: List[Dict[str, Any]] self, contents: List[Dict[str, Any]]
) -> Iterable[Dict[str, Any]]: ) -> Iterable[Dict[str, Any]]:
for content in contents: for content in contents:
if not self._cql_runner.skipped_content_get_from_pk(content): if not self._cql_runner.skipped_content_get_from_pk(content):
yield {algo: content[algo] for algo in DEFAULT_ALGORITHMS} yield {algo: content[algo] for algo in DEFAULT_ALGORITHMS}
def directory_add(self, directories: List[Directory]) -> Dict: def directory_add(self, directories: List[Directory]) -> Dict:
to_add = {d.id: d for d in directories}.values() to_add = {d.id: d for d in directories}.values()
if self._check_missing:
# Filter out directories that are already inserted. # Filter out directories that are already inserted.
missing = self.directory_missing([dir_.id for dir_ in to_add]) missing = self.directory_missing([dir_.id for dir_ in to_add])
directories = [dir_ for dir_ in directories if dir_.id in missing] directories = [dir_ for dir_ in directories if dir_.id in missing]
self.journal_writer.directory_add(directories) self.journal_writer.directory_add(directories)
for directory in directories: for directory in directories:
# Add directory entries to the 'directory_entry' table # Add directory entries to the 'directory_entry' table
for entry in directory.entries: for entry in directory.entries:
self._cql_runner.directory_entry_add_one( self._cql_runner.directory_entry_add_one(
DirectoryEntryRow(directory_id=directory.id, **entry.to_dict()) DirectoryEntryRow(directory_id=directory.id, **entry.to_dict())
▲ Show 20 Lines • Show All 106 Lines • ▼ Show 20 Linesclass CassandraStorage:
def directory_get_random(self) -> Sha1Git: def directory_get_random(self) -> Sha1Git:
directory = self._cql_runner.directory_get_random() directory = self._cql_runner.directory_get_random()
assert directory, "Could not find any directory" assert directory, "Could not find any directory"
return directory.id return directory.id
def revision_add(self, revisions: List[Revision]) -> Dict: def revision_add(self, revisions: List[Revision]) -> Dict:
# Filter-out revisions already in the database # Filter-out revisions already in the database
if self._check_missing:
to_add = {r.id: r for r in revisions}.values() to_add = {r.id: r for r in revisions}.values()
missing = self.revision_missing([rev.id for rev in to_add]) missing = self.revision_missing([rev.id for rev in to_add])
revisions = [rev for rev in revisions if rev.id in missing] revisions = [rev for rev in revisions if rev.id in missing]
self.journal_writer.revision_add(revisions) self.journal_writer.revision_add(revisions)
for revision in revisions: for revision in revisions:
revobject = converters.revision_to_db(revision) revobject = converters.revision_to_db(revision)
if revobject: if revobject:
# Add parents first # Add parents first
for (rank, parent) in enumerate(revision.parents): for (rank, parent) in enumerate(revision.parents):
self._cql_runner.revision_parent_add_one( self._cql_runner.revision_parent_add_one(
▲ Show 20 Lines • Show All 92 Lines • ▼ Show 20 Lines ) -> Iterable[Optional[Tuple[Sha1Git, Tuple[Sha1Git, ...]]]]:
yield from self._get_parent_revs(revisions, seen, limit, True) yield from self._get_parent_revs(revisions, seen, limit, True)
def revision_get_random(self) -> Sha1Git: def revision_get_random(self) -> Sha1Git:
revision = self._cql_runner.revision_get_random() revision = self._cql_runner.revision_get_random()
assert revision, "Could not find any revision" assert revision, "Could not find any revision"
return revision.id return revision.id
def release_add(self, releases: List[Release]) -> Dict: def release_add(self, releases: List[Release]) -> Dict:
if self._check_missing:
to_add = {r.id: r for r in releases}.values() to_add = {r.id: r for r in releases}.values()
missing = set(self.release_missing([rel.id for rel in to_add])) missing = set(self.release_missing([rel.id for rel in to_add]))
releases = [rel for rel in to_add if rel.id in missing] releases = [rel for rel in to_add if rel.id in missing]
self.journal_writer.release_add(releases) self.journal_writer.release_add(releases)
for release in releases: for release in releases:
if release: if release:
self._cql_runner.release_add_one(converters.release_to_db(release)) self._cql_runner.release_add_one(converters.release_to_db(release))
return {"release:add": len(releases)} return {"release:add": len(releases)}
Show All 10 Lines def release_get(self, releases: List[Sha1Git]) -> List[Optional[Release]]:
return [rels.get(rel_id) for rel_id in releases] return [rels.get(rel_id) for rel_id in releases]
def release_get_random(self) -> Sha1Git: def release_get_random(self) -> Sha1Git:
release = self._cql_runner.release_get_random() release = self._cql_runner.release_get_random()
assert release, "Could not find any release" assert release, "Could not find any release"
return release.id return release.id
def snapshot_add(self, snapshots: List[Snapshot]) -> Dict: def snapshot_add(self, snapshots: List[Snapshot]) -> Dict:
if self._check_missing:
to_add = {s.id: s for s in snapshots}.values() to_add = {s.id: s for s in snapshots}.values()
missing = self._cql_runner.snapshot_missing([snp.id for snp in to_add]) missing = self._cql_runner.snapshot_missing([snp.id for snp in to_add])
snapshots = [snp for snp in snapshots if snp.id in missing] snapshots = [snp for snp in snapshots if snp.id in missing]
for snapshot in snapshots: for snapshot in snapshots:
self.journal_writer.snapshot_add([snapshot]) self.journal_writer.snapshot_add([snapshot])
# Add branches # Add branches
for (branch_name, branch) in snapshot.branches.items(): for (branch_name, branch) in snapshot.branches.items():
if branch is None: if branch is None:
target_type: Optional[str] = None target_type: Optional[str] = None
▲ Show 20 Lines • Show All 252 Lines • ▼ Show 20 Linesclass CassandraStorage:
def origin_count( def origin_count(
self, url_pattern: str, regexp: bool = False, with_visit: bool = False self, url_pattern: str, regexp: bool = False, with_visit: bool = False
) -> int: ) -> int:
raise NotImplementedError( raise NotImplementedError(
"The Cassandra backend does not implement origin_count" "The Cassandra backend does not implement origin_count"
) )
def origin_add(self, origins: List[Origin]) -> Dict[str, int]: def origin_add(self, origins: List[Origin]) -> Dict[str, int]:
if self._check_missing:
to_add = {o.url: o for o in origins}.values() to_add = {o.url: o for o in origins}.values()
origins = [ori for ori in to_add if self.origin_get_one(ori.url) is None] origins = [ori for ori in to_add if self.origin_get_one(ori.url) is None]
self.journal_writer.origin_add(origins) self.journal_writer.origin_add(origins)
for origin in origins: for origin in origins:
self._cql_runner.origin_add_one( self._cql_runner.origin_add_one(
OriginRow(sha1=hash_url(origin.url), url=origin.url, next_visit_id=1) OriginRow(sha1=hash_url(origin.url), url=origin.url, next_visit_id=1)
) )
return {"origin:add": len(origins)} return {"origin:add": len(origins)}
▲ Show 20 Lines • Show All 430 Lines • ▼ Show 20 Lines ) -> Optional[MetadataAuthority]:
url=authority.url, url=authority.url,
metadata=json.loads(authority.metadata), metadata=json.loads(authority.metadata),
) )
else: else:
return None return None
# ExtID tables # ExtID tables
def extid_add(self, ids: List[ExtID]) -> Dict[str, int]: def extid_add(self, ids: List[ExtID]) -> Dict[str, int]:
if self._check_missing:
extids = [ extids = [
extid extid
for extid in ids for extid in ids
if not self._cql_runner.extid_get_from_pk( if not self._cql_runner.extid_get_from_pk(
extid_type=extid.extid_type, extid=extid.extid, target=extid.target, extid_type=extid.extid_type, extid=extid.extid, target=extid.target,
) )
] ]
self.journal_writer.extid_add(extids) self.journal_writer.extid_add(extids)
inserted = 0 inserted = 0
for extid in extids: for extid in extids:
target_type = extid.target.object_type.value target_type = extid.target.object_type.value
target = extid.target.object_id target = extid.target.object_id
extidrow = ExtIDRow( extidrow = ExtIDRow(
▲ Show 20 Lines • Show All 60 Lines • Show Last 20 Lines
View Options

swh/web/add_forge_now/apps.py

Show First 20 Lines • Show All 83 Lines • ▼ Show 20 Lines
) )
from .schema import HASH_ALGORITHMS from .schema import HASH_ALGORITHMS
# Max block size of contents to return # Max block size of contents to return
BULK_BLOCK_CONTENT_LEN_MAX = 10000 BULK_BLOCK_CONTENT_LEN_MAX = 10000
class CassandraStorage: class CassandraStorage:
def __init__(self, hosts, keyspace, objstorage, port=9042, journal_writer=None): def __init__(
self._cql_runner: CqlRunner = CqlRunner(hosts, keyspace, port) self,
hosts,
keyspace,
objstorage,
port=9042,
journal_writer=None,
check_missing=True,
):
"""
Args:
hosts: Seed Cassandra nodes, to start connecting to the cluster
keyspace: Name of the Cassandra database to use
objstorage: Passed as argument to :cls:`ObjStorage`
port: Cassandra port
journal_writer: Passed as argument to :cls:`JournalWriter`
check_missing: Whether ``*_add`` functions will check if an object already
exists in the database before sending it in an INSERT.
``True`` is the default as it is more efficient when there is a moderately
high probability the object is already known, but ``False`` can be
useful to overwrite existing objects (eg. when applying a schema update),
or when the database is known to be mostly empty.
"""
self._hosts = hosts
self._keyspace = keyspace
self._port = port
self._set_cql_runner()
self.journal_writer: JournalWriter = JournalWriter(journal_writer) self.journal_writer: JournalWriter = JournalWriter(journal_writer)
self.objstorage: ObjStorage = ObjStorage(objstorage) self.objstorage: ObjStorage = ObjStorage(objstorage)
self._check_missing = check_missing
def _set_cql_runner(self):
"""Used by tests when they need to reset the CqlRunner"""
self._cql_runner: CqlRunner = CqlRunner(self._hosts, self._keyspace, self._port)
def check_config(self, *, check_write: bool) -> bool: def check_config(self, *, check_write: bool) -> bool:
self._cql_runner.check_read() self._cql_runner.check_read()
return True return True
def _content_get_from_hash(self, algo, hash_) -> Iterable: def _content_get_from_hash(self, algo, hash_) -> Iterable:
"""From the name of a hash algorithm and a value of that hash, """From the name of a hash algorithm and a value of that hash,
Show All 10 Lines def _content_get_from_hash(self, algo, hash_) -> Iterable:
for row in res: for row in res:
# re-check the the hash (in case of murmur3 collision) # re-check the the hash (in case of murmur3 collision)
if getattr(row, algo) == hash_: if getattr(row, algo) == hash_:
yield row yield row
def _content_add(self, contents: List[Content], with_data: bool) -> Dict: def _content_add(self, contents: List[Content], with_data: bool) -> Dict:
# Filter-out content already in the database. # Filter-out content already in the database.
if self._check_missing:
contents = [ contents = [
c for c in contents if not self._cql_runner.content_get_from_pk(c.to_dict()) c
for c in contents
if not self._cql_runner.content_get_from_pk(c.to_dict())
] ]
if with_data: if with_data:
# First insert to the objstorage, if the endpoint is # First insert to the objstorage, if the endpoint is
# `content_add` (as opposed to `content_add_metadata`). # `content_add` (as opposed to `content_add_metadata`).
# Must add to the objstorage before the DB and journal. Otherwise: # Must add to the objstorage before the DB and journal. Otherwise:
# 1. in case of a crash the DB may "believe" we have the content, but # 1. in case of a crash the DB may "believe" we have the content, but
# we didn't have time to write to the objstorage before the crash # we didn't have time to write to the objstorage before the crash
Show All 12 Lines def _content_add(self, contents: List[Content], with_data: bool) -> Dict:
# Check for sha1 or sha1_git collisions. This test is not atomic # Check for sha1 or sha1_git collisions. This test is not atomic
# with the insertion, so it won't detect a collision if both # with the insertion, so it won't detect a collision if both
# contents are inserted at the same time, but it's good enough. # contents are inserted at the same time, but it's good enough.
# #
# The proper way to do it would probably be a BATCH, but this # The proper way to do it would probably be a BATCH, but this
# would be inefficient because of the number of partitions we # would be inefficient because of the number of partitions we
# need to affect (len(HASH_ALGORITHMS)+1, which is currently 5) # need to affect (len(HASH_ALGORITHMS)+1, which is currently 5)
if self._check_missing:
for algo in {"sha1", "sha1_git"}: for algo in {"sha1", "sha1_git"}:
collisions = [] collisions = []
# Get tokens of 'content' rows with the same value for # Get tokens of 'content' rows with the same value for
# sha1/sha1_git # sha1/sha1_git
rows = self._content_get_from_hash(algo, content.get_hash(algo)) rows = self._content_get_from_hash(algo, content.get_hash(algo))
for row in rows: for row in rows:
if getattr(row, algo) != content.get_hash(algo): if getattr(row, algo) != content.get_hash(algo):
# collision of token(partition key), ignore this # collision of token(partition key), ignore this
# row # row
continue continue
for other_algo in HASH_ALGORITHMS: for other_algo in HASH_ALGORITHMS:
if getattr(row, other_algo) != content.get_hash(other_algo): if getattr(row, other_algo) != content.get_hash(other_algo):
# This hash didn't match; discard the row. # This hash didn't match; discard the row.
collisions.append( collisions.append(
{k: getattr(row, k) for k in HASH_ALGORITHMS} {k: getattr(row, k) for k in HASH_ALGORITHMS}
) )
if collisions: if collisions:
collisions.append(content.hashes()) collisions.append(content.hashes())
raise HashCollision(algo, content.get_hash(algo), collisions) raise HashCollision(algo, content.get_hash(algo), collisions)
douarddaUnsubmitted
Not Done
Inline Actions

why is the whole collision detection block is now run only if check_missing is True?

douardda: why is the whole collision detection block is now run only if check_missing is True?
vlorentzAuthorUnsubmitted
Done
Inline Actions

Because, if an object already exists with the exact same hashes and it was not filtered out initially, then self._content_get_from_hash(algo, content.get_hash(algo)) will return it and it would be detected as a collision.

I could filter the results of self._content_get_from_hash, but I don't think it's worth it, as _check_missing is pretty niche anyway.

vlorentz: Because, if an object already exists with the exact same hashes and it was not filtered out…
vlorentzAuthorUnsubmitted
Done
Inline Actions

And collision detection in Cassandra isn't guaranteed anyway, because two colliding objects may be inserted at the same time and Cassandra is only eventually consistent.

vlorentz: And collision detection in Cassandra isn't guaranteed anyway, because two colliding objects may…
(token, insertion_finalizer) = self._cql_runner.content_add_prepare( (token, insertion_finalizer) = self._cql_runner.content_add_prepare(
ContentRow(**remove_keys(content.to_dict(), ("data",))) ContentRow(**remove_keys(content.to_dict(), ("data",)))
) )
# Then add to index tables # Then add to index tables
for algo in HASH_ALGORITHMS: for algo in HASH_ALGORITHMS:
self._cql_runner.content_index_add_one(algo, content, token) self._cql_runner.content_index_add_one(algo, content, token)
▲ Show 20 Lines • Show All 136 Lines • ▼ Show 20 Linesclass CassandraStorage:
def content_get_random(self) -> Sha1Git: def content_get_random(self) -> Sha1Git:
content = self._cql_runner.content_get_random() content = self._cql_runner.content_get_random()
assert content, "Could not find any content" assert content, "Could not find any content"
return content.sha1_git return content.sha1_git
def _skipped_content_add(self, contents: List[SkippedContent]) -> Dict: def _skipped_content_add(self, contents: List[SkippedContent]) -> Dict:
# Filter-out content already in the database. # Filter-out content already in the database.
if self._check_missing:
contents = [ contents = [
c c
for c in contents for c in contents
if not self._cql_runner.skipped_content_get_from_pk(c.to_dict()) if not self._cql_runner.skipped_content_get_from_pk(c.to_dict())
] ]
self.journal_writer.skipped_content_add(contents) self.journal_writer.skipped_content_add(contents)
for content in contents: for content in contents:
# Compute token of the row in the main table # Compute token of the row in the main table
(token, insertion_finalizer) = self._cql_runner.skipped_content_add_prepare( (token, insertion_finalizer) = self._cql_runner.skipped_content_add_prepare(
SkippedContentRow.from_dict({"origin": None, **content.to_dict()}) SkippedContentRow.from_dict({"origin": None, **content.to_dict()})
) )
Show All 15 Lines def skipped_content_missing(
self, contents: List[Dict[str, Any]] self, contents: List[Dict[str, Any]]
) -> Iterable[Dict[str, Any]]: ) -> Iterable[Dict[str, Any]]:
for content in contents: for content in contents:
if not self._cql_runner.skipped_content_get_from_pk(content): if not self._cql_runner.skipped_content_get_from_pk(content):
yield {algo: content[algo] for algo in DEFAULT_ALGORITHMS} yield {algo: content[algo] for algo in DEFAULT_ALGORITHMS}
def directory_add(self, directories: List[Directory]) -> Dict: def directory_add(self, directories: List[Directory]) -> Dict:
to_add = {d.id: d for d in directories}.values() to_add = {d.id: d for d in directories}.values()
if self._check_missing:
# Filter out directories that are already inserted. # Filter out directories that are already inserted.
missing = self.directory_missing([dir_.id for dir_ in to_add]) missing = self.directory_missing([dir_.id for dir_ in to_add])
directories = [dir_ for dir_ in directories if dir_.id in missing] directories = [dir_ for dir_ in directories if dir_.id in missing]
self.journal_writer.directory_add(directories) self.journal_writer.directory_add(directories)
for directory in directories: for directory in directories:
# Add directory entries to the 'directory_entry' table # Add directory entries to the 'directory_entry' table
for entry in directory.entries: for entry in directory.entries:
self._cql_runner.directory_entry_add_one( self._cql_runner.directory_entry_add_one(
DirectoryEntryRow(directory_id=directory.id, **entry.to_dict()) DirectoryEntryRow(directory_id=directory.id, **entry.to_dict())
▲ Show 20 Lines • Show All 106 Lines • ▼ Show 20 Linesclass CassandraStorage:
def directory_get_random(self) -> Sha1Git: def directory_get_random(self) -> Sha1Git:
directory = self._cql_runner.directory_get_random() directory = self._cql_runner.directory_get_random()
assert directory, "Could not find any directory" assert directory, "Could not find any directory"
return directory.id return directory.id
def revision_add(self, revisions: List[Revision]) -> Dict: def revision_add(self, revisions: List[Revision]) -> Dict:
# Filter-out revisions already in the database # Filter-out revisions already in the database
if self._check_missing:
to_add = {r.id: r for r in revisions}.values() to_add = {r.id: r for r in revisions}.values()
missing = self.revision_missing([rev.id for rev in to_add]) missing = self.revision_missing([rev.id for rev in to_add])
revisions = [rev for rev in revisions if rev.id in missing] revisions = [rev for rev in revisions if rev.id in missing]
self.journal_writer.revision_add(revisions) self.journal_writer.revision_add(revisions)
for revision in revisions: for revision in revisions:
revobject = converters.revision_to_db(revision) revobject = converters.revision_to_db(revision)
if revobject: if revobject:
# Add parents first # Add parents first
for (rank, parent) in enumerate(revision.parents): for (rank, parent) in enumerate(revision.parents):
self._cql_runner.revision_parent_add_one( self._cql_runner.revision_parent_add_one(
▲ Show 20 Lines • Show All 92 Lines • ▼ Show 20 Lines ) -> Iterable[Optional[Tuple[Sha1Git, Tuple[Sha1Git, ...]]]]:
yield from self._get_parent_revs(revisions, seen, limit, True) yield from self._get_parent_revs(revisions, seen, limit, True)
def revision_get_random(self) -> Sha1Git: def revision_get_random(self) -> Sha1Git:
revision = self._cql_runner.revision_get_random() revision = self._cql_runner.revision_get_random()
assert revision, "Could not find any revision" assert revision, "Could not find any revision"
return revision.id return revision.id
def release_add(self, releases: List[Release]) -> Dict: def release_add(self, releases: List[Release]) -> Dict:
if self._check_missing:
to_add = {r.id: r for r in releases}.values() to_add = {r.id: r for r in releases}.values()
missing = set(self.release_missing([rel.id for rel in to_add])) missing = set(self.release_missing([rel.id for rel in to_add]))
releases = [rel for rel in to_add if rel.id in missing] releases = [rel for rel in to_add if rel.id in missing]
self.journal_writer.release_add(releases) self.journal_writer.release_add(releases)
for release in releases: for release in releases:
if release: if release:
self._cql_runner.release_add_one(converters.release_to_db(release)) self._cql_runner.release_add_one(converters.release_to_db(release))
return {"release:add": len(releases)} return {"release:add": len(releases)}
Show All 10 Lines def release_get(self, releases: List[Sha1Git]) -> List[Optional[Release]]:
return [rels.get(rel_id) for rel_id in releases] return [rels.get(rel_id) for rel_id in releases]
def release_get_random(self) -> Sha1Git: def release_get_random(self) -> Sha1Git:
release = self._cql_runner.release_get_random() release = self._cql_runner.release_get_random()
assert release, "Could not find any release" assert release, "Could not find any release"
return release.id return release.id
def snapshot_add(self, snapshots: List[Snapshot]) -> Dict: def snapshot_add(self, snapshots: List[Snapshot]) -> Dict:
if self._check_missing:
to_add = {s.id: s for s in snapshots}.values() to_add = {s.id: s for s in snapshots}.values()
missing = self._cql_runner.snapshot_missing([snp.id for snp in to_add]) missing = self._cql_runner.snapshot_missing([snp.id for snp in to_add])
snapshots = [snp for snp in snapshots if snp.id in missing] snapshots = [snp for snp in snapshots if snp.id in missing]
for snapshot in snapshots: for snapshot in snapshots:
self.journal_writer.snapshot_add([snapshot]) self.journal_writer.snapshot_add([snapshot])
# Add branches # Add branches
for (branch_name, branch) in snapshot.branches.items(): for (branch_name, branch) in snapshot.branches.items():
if branch is None: if branch is None:
target_type: Optional[str] = None target_type: Optional[str] = None
▲ Show 20 Lines • Show All 252 Lines • ▼ Show 20 Linesclass CassandraStorage:
def origin_count( def origin_count(
self, url_pattern: str, regexp: bool = False, with_visit: bool = False self, url_pattern: str, regexp: bool = False, with_visit: bool = False
) -> int: ) -> int:
raise NotImplementedError( raise NotImplementedError(
"The Cassandra backend does not implement origin_count" "The Cassandra backend does not implement origin_count"
) )
def origin_add(self, origins: List[Origin]) -> Dict[str, int]: def origin_add(self, origins: List[Origin]) -> Dict[str, int]:
if self._check_missing:
to_add = {o.url: o for o in origins}.values() to_add = {o.url: o for o in origins}.values()
origins = [ori for ori in to_add if self.origin_get_one(ori.url) is None] origins = [ori for ori in to_add if self.origin_get_one(ori.url) is None]
self.journal_writer.origin_add(origins) self.journal_writer.origin_add(origins)
for origin in origins: for origin in origins:
self._cql_runner.origin_add_one( self._cql_runner.origin_add_one(
OriginRow(sha1=hash_url(origin.url), url=origin.url, next_visit_id=1) OriginRow(sha1=hash_url(origin.url), url=origin.url, next_visit_id=1)
) )
return {"origin:add": len(origins)} return {"origin:add": len(origins)}
▲ Show 20 Lines • Show All 430 Lines • ▼ Show 20 Lines ) -> Optional[MetadataAuthority]:
url=authority.url, url=authority.url,
metadata=json.loads(authority.metadata), metadata=json.loads(authority.metadata),
) )
else: else:
return None return None
# ExtID tables # ExtID tables
def extid_add(self, ids: List[ExtID]) -> Dict[str, int]: def extid_add(self, ids: List[ExtID]) -> Dict[str, int]:
if self._check_missing:
extids = [ extids = [
extid extid
for extid in ids for extid in ids
if not self._cql_runner.extid_get_from_pk( if not self._cql_runner.extid_get_from_pk(
extid_type=extid.extid_type, extid=extid.extid, target=extid.target, extid_type=extid.extid_type, extid=extid.extid, target=extid.target,
) )
] ]
self.journal_writer.extid_add(extids) self.journal_writer.extid_add(extids)
inserted = 0 inserted = 0
for extid in extids: for extid in extids:
target_type = extid.target.object_type.value target_type = extid.target.object_type.value
target = extid.target.object_id target = extid.target.object_id
extidrow = ExtIDRow( extidrow = ExtIDRow(
▲ Show 20 Lines • Show All 60 Lines • Show Last 20 Lines
View Options

swh/web/add_forge_now/migrations/0001_initial.py

Show First 20 Lines • Show All 83 Lines • ▼ Show 20 Lines
) )
from .schema import HASH_ALGORITHMS from .schema import HASH_ALGORITHMS
# Max block size of contents to return # Max block size of contents to return
BULK_BLOCK_CONTENT_LEN_MAX = 10000 BULK_BLOCK_CONTENT_LEN_MAX = 10000
class CassandraStorage: class CassandraStorage:
def __init__(self, hosts, keyspace, objstorage, port=9042, journal_writer=None): def __init__(
self._cql_runner: CqlRunner = CqlRunner(hosts, keyspace, port) self,
hosts,
keyspace,
objstorage,
port=9042,
journal_writer=None,
check_missing=True,
):
"""
Args:
hosts: Seed Cassandra nodes, to start connecting to the cluster
keyspace: Name of the Cassandra database to use
objstorage: Passed as argument to :cls:`ObjStorage`
port: Cassandra port
journal_writer: Passed as argument to :cls:`JournalWriter`
check_missing: Whether ``*_add`` functions will check if an object already
exists in the database before sending it in an INSERT.
``True`` is the default as it is more efficient when there is a moderately
high probability the object is already known, but ``False`` can be
useful to overwrite existing objects (eg. when applying a schema update),
or when the database is known to be mostly empty.
"""
self._hosts = hosts
self._keyspace = keyspace
self._port = port
self._set_cql_runner()
self.journal_writer: JournalWriter = JournalWriter(journal_writer) self.journal_writer: JournalWriter = JournalWriter(journal_writer)
self.objstorage: ObjStorage = ObjStorage(objstorage) self.objstorage: ObjStorage = ObjStorage(objstorage)
self._check_missing = check_missing
def _set_cql_runner(self):
"""Used by tests when they need to reset the CqlRunner"""
self._cql_runner: CqlRunner = CqlRunner(self._hosts, self._keyspace, self._port)
def check_config(self, *, check_write: bool) -> bool: def check_config(self, *, check_write: bool) -> bool:
self._cql_runner.check_read() self._cql_runner.check_read()
return True return True
def _content_get_from_hash(self, algo, hash_) -> Iterable: def _content_get_from_hash(self, algo, hash_) -> Iterable:
"""From the name of a hash algorithm and a value of that hash, """From the name of a hash algorithm and a value of that hash,
Show All 10 Lines def _content_get_from_hash(self, algo, hash_) -> Iterable:
for row in res: for row in res:
# re-check the the hash (in case of murmur3 collision) # re-check the the hash (in case of murmur3 collision)
if getattr(row, algo) == hash_: if getattr(row, algo) == hash_:
yield row yield row
def _content_add(self, contents: List[Content], with_data: bool) -> Dict: def _content_add(self, contents: List[Content], with_data: bool) -> Dict:
# Filter-out content already in the database. # Filter-out content already in the database.
if self._check_missing:
contents = [ contents = [
c for c in contents if not self._cql_runner.content_get_from_pk(c.to_dict()) c
for c in contents
if not self._cql_runner.content_get_from_pk(c.to_dict())
] ]
if with_data: if with_data:
# First insert to the objstorage, if the endpoint is # First insert to the objstorage, if the endpoint is
# `content_add` (as opposed to `content_add_metadata`). # `content_add` (as opposed to `content_add_metadata`).
# Must add to the objstorage before the DB and journal. Otherwise: # Must add to the objstorage before the DB and journal. Otherwise:
# 1. in case of a crash the DB may "believe" we have the content, but # 1. in case of a crash the DB may "believe" we have the content, but
# we didn't have time to write to the objstorage before the crash # we didn't have time to write to the objstorage before the crash
Show All 12 Lines def _content_add(self, contents: List[Content], with_data: bool) -> Dict:
# Check for sha1 or sha1_git collisions. This test is not atomic # Check for sha1 or sha1_git collisions. This test is not atomic
# with the insertion, so it won't detect a collision if both # with the insertion, so it won't detect a collision if both
# contents are inserted at the same time, but it's good enough. # contents are inserted at the same time, but it's good enough.
# #
# The proper way to do it would probably be a BATCH, but this # The proper way to do it would probably be a BATCH, but this
# would be inefficient because of the number of partitions we # would be inefficient because of the number of partitions we
# need to affect (len(HASH_ALGORITHMS)+1, which is currently 5) # need to affect (len(HASH_ALGORITHMS)+1, which is currently 5)
if self._check_missing:
for algo in {"sha1", "sha1_git"}: for algo in {"sha1", "sha1_git"}:
collisions = [] collisions = []
# Get tokens of 'content' rows with the same value for # Get tokens of 'content' rows with the same value for
# sha1/sha1_git # sha1/sha1_git
rows = self._content_get_from_hash(algo, content.get_hash(algo)) rows = self._content_get_from_hash(algo, content.get_hash(algo))
for row in rows: for row in rows:
if getattr(row, algo) != content.get_hash(algo): if getattr(row, algo) != content.get_hash(algo):
# collision of token(partition key), ignore this # collision of token(partition key), ignore this
# row # row
continue continue
for other_algo in HASH_ALGORITHMS: for other_algo in HASH_ALGORITHMS:
if getattr(row, other_algo) != content.get_hash(other_algo): if getattr(row, other_algo) != content.get_hash(other_algo):
# This hash didn't match; discard the row. # This hash didn't match; discard the row.
collisions.append( collisions.append(
{k: getattr(row, k) for k in HASH_ALGORITHMS} {k: getattr(row, k) for k in HASH_ALGORITHMS}
) )
if collisions: if collisions:
collisions.append(content.hashes()) collisions.append(content.hashes())
raise HashCollision(algo, content.get_hash(algo), collisions) raise HashCollision(algo, content.get_hash(algo), collisions)
douarddaUnsubmitted
Not Done
Inline Actions

why is the whole collision detection block is now run only if check_missing is True?

douardda: why is the whole collision detection block is now run only if check_missing is True?
vlorentzAuthorUnsubmitted
Done
Inline Actions

Because, if an object already exists with the exact same hashes and it was not filtered out initially, then self._content_get_from_hash(algo, content.get_hash(algo)) will return it and it would be detected as a collision.

I could filter the results of self._content_get_from_hash, but I don't think it's worth it, as _check_missing is pretty niche anyway.

vlorentz: Because, if an object already exists with the exact same hashes and it was not filtered out…
vlorentzAuthorUnsubmitted
Done
Inline Actions

And collision detection in Cassandra isn't guaranteed anyway, because two colliding objects may be inserted at the same time and Cassandra is only eventually consistent.

vlorentz: And collision detection in Cassandra isn't guaranteed anyway, because two colliding objects may…
(token, insertion_finalizer) = self._cql_runner.content_add_prepare( (token, insertion_finalizer) = self._cql_runner.content_add_prepare(
ContentRow(**remove_keys(content.to_dict(), ("data",))) ContentRow(**remove_keys(content.to_dict(), ("data",)))
) )
# Then add to index tables # Then add to index tables
for algo in HASH_ALGORITHMS: for algo in HASH_ALGORITHMS:
self._cql_runner.content_index_add_one(algo, content, token) self._cql_runner.content_index_add_one(algo, content, token)
▲ Show 20 Lines • Show All 136 Lines • ▼ Show 20 Linesclass CassandraStorage:
def content_get_random(self) -> Sha1Git: def content_get_random(self) -> Sha1Git:
content = self._cql_runner.content_get_random() content = self._cql_runner.content_get_random()
assert content, "Could not find any content" assert content, "Could not find any content"
return content.sha1_git return content.sha1_git
def _skipped_content_add(self, contents: List[SkippedContent]) -> Dict: def _skipped_content_add(self, contents: List[SkippedContent]) -> Dict:
# Filter-out content already in the database. # Filter-out content already in the database.
if self._check_missing:
contents = [ contents = [
c c
for c in contents for c in contents
if not self._cql_runner.skipped_content_get_from_pk(c.to_dict()) if not self._cql_runner.skipped_content_get_from_pk(c.to_dict())
] ]
self.journal_writer.skipped_content_add(contents) self.journal_writer.skipped_content_add(contents)
for content in contents: for content in contents:
# Compute token of the row in the main table # Compute token of the row in the main table
(token, insertion_finalizer) = self._cql_runner.skipped_content_add_prepare( (token, insertion_finalizer) = self._cql_runner.skipped_content_add_prepare(
SkippedContentRow.from_dict({"origin": None, **content.to_dict()}) SkippedContentRow.from_dict({"origin": None, **content.to_dict()})
) )
Show All 15 Lines def skipped_content_missing(
self, contents: List[Dict[str, Any]] self, contents: List[Dict[str, Any]]
) -> Iterable[Dict[str, Any]]: ) -> Iterable[Dict[str, Any]]:
for content in contents: for content in contents:
if not self._cql_runner.skipped_content_get_from_pk(content): if not self._cql_runner.skipped_content_get_from_pk(content):
yield {algo: content[algo] for algo in DEFAULT_ALGORITHMS} yield {algo: content[algo] for algo in DEFAULT_ALGORITHMS}
def directory_add(self, directories: List[Directory]) -> Dict: def directory_add(self, directories: List[Directory]) -> Dict:
to_add = {d.id: d for d in directories}.values() to_add = {d.id: d for d in directories}.values()
if self._check_missing:
# Filter out directories that are already inserted. # Filter out directories that are already inserted.
missing = self.directory_missing([dir_.id for dir_ in to_add]) missing = self.directory_missing([dir_.id for dir_ in to_add])
directories = [dir_ for dir_ in directories if dir_.id in missing] directories = [dir_ for dir_ in directories if dir_.id in missing]
self.journal_writer.directory_add(directories) self.journal_writer.directory_add(directories)
for directory in directories: for directory in directories:
# Add directory entries to the 'directory_entry' table # Add directory entries to the 'directory_entry' table
for entry in directory.entries: for entry in directory.entries:
self._cql_runner.directory_entry_add_one( self._cql_runner.directory_entry_add_one(
DirectoryEntryRow(directory_id=directory.id, **entry.to_dict()) DirectoryEntryRow(directory_id=directory.id, **entry.to_dict())
▲ Show 20 Lines • Show All 106 Lines • ▼ Show 20 Linesclass CassandraStorage:
def directory_get_random(self) -> Sha1Git: def directory_get_random(self) -> Sha1Git:
directory = self._cql_runner.directory_get_random() directory = self._cql_runner.directory_get_random()
assert directory, "Could not find any directory" assert directory, "Could not find any directory"
return directory.id return directory.id
def revision_add(self, revisions: List[Revision]) -> Dict: def revision_add(self, revisions: List[Revision]) -> Dict:
# Filter-out revisions already in the database # Filter-out revisions already in the database
if self._check_missing:
to_add = {r.id: r for r in revisions}.values() to_add = {r.id: r for r in revisions}.values()
missing = self.revision_missing([rev.id for rev in to_add]) missing = self.revision_missing([rev.id for rev in to_add])
revisions = [rev for rev in revisions if rev.id in missing] revisions = [rev for rev in revisions if rev.id in missing]
self.journal_writer.revision_add(revisions) self.journal_writer.revision_add(revisions)
for revision in revisions: for revision in revisions:
revobject = converters.revision_to_db(revision) revobject = converters.revision_to_db(revision)
if revobject: if revobject:
# Add parents first # Add parents first
for (rank, parent) in enumerate(revision.parents): for (rank, parent) in enumerate(revision.parents):
self._cql_runner.revision_parent_add_one( self._cql_runner.revision_parent_add_one(
▲ Show 20 Lines • Show All 92 Lines • ▼ Show 20 Lines ) -> Iterable[Optional[Tuple[Sha1Git, Tuple[Sha1Git, ...]]]]:
yield from self._get_parent_revs(revisions, seen, limit, True) yield from self._get_parent_revs(revisions, seen, limit, True)
def revision_get_random(self) -> Sha1Git: def revision_get_random(self) -> Sha1Git:
revision = self._cql_runner.revision_get_random() revision = self._cql_runner.revision_get_random()
assert revision, "Could not find any revision" assert revision, "Could not find any revision"
return revision.id return revision.id
def release_add(self, releases: List[Release]) -> Dict: def release_add(self, releases: List[Release]) -> Dict:
if self._check_missing:
to_add = {r.id: r for r in releases}.values() to_add = {r.id: r for r in releases}.values()
missing = set(self.release_missing([rel.id for rel in to_add])) missing = set(self.release_missing([rel.id for rel in to_add]))
releases = [rel for rel in to_add if rel.id in missing] releases = [rel for rel in to_add if rel.id in missing]
self.journal_writer.release_add(releases) self.journal_writer.release_add(releases)
for release in releases: for release in releases:
if release: if release:
self._cql_runner.release_add_one(converters.release_to_db(release)) self._cql_runner.release_add_one(converters.release_to_db(release))
return {"release:add": len(releases)} return {"release:add": len(releases)}
Show All 10 Lines def release_get(self, releases: List[Sha1Git]) -> List[Optional[Release]]:
return [rels.get(rel_id) for rel_id in releases] return [rels.get(rel_id) for rel_id in releases]
def release_get_random(self) -> Sha1Git: def release_get_random(self) -> Sha1Git:
release = self._cql_runner.release_get_random() release = self._cql_runner.release_get_random()
assert release, "Could not find any release" assert release, "Could not find any release"
return release.id return release.id
def snapshot_add(self, snapshots: List[Snapshot]) -> Dict: def snapshot_add(self, snapshots: List[Snapshot]) -> Dict:
if self._check_missing:
to_add = {s.id: s for s in snapshots}.values() to_add = {s.id: s for s in snapshots}.values()
missing = self._cql_runner.snapshot_missing([snp.id for snp in to_add]) missing = self._cql_runner.snapshot_missing([snp.id for snp in to_add])
snapshots = [snp for snp in snapshots if snp.id in missing] snapshots = [snp for snp in snapshots if snp.id in missing]
for snapshot in snapshots: for snapshot in snapshots:
self.journal_writer.snapshot_add([snapshot]) self.journal_writer.snapshot_add([snapshot])
# Add branches # Add branches
for (branch_name, branch) in snapshot.branches.items(): for (branch_name, branch) in snapshot.branches.items():
if branch is None: if branch is None:
target_type: Optional[str] = None target_type: Optional[str] = None
▲ Show 20 Lines • Show All 252 Lines • ▼ Show 20 Linesclass CassandraStorage:
def origin_count( def origin_count(
self, url_pattern: str, regexp: bool = False, with_visit: bool = False self, url_pattern: str, regexp: bool = False, with_visit: bool = False
) -> int: ) -> int:
raise NotImplementedError( raise NotImplementedError(
"The Cassandra backend does not implement origin_count" "The Cassandra backend does not implement origin_count"
) )
def origin_add(self, origins: List[Origin]) -> Dict[str, int]: def origin_add(self, origins: List[Origin]) -> Dict[str, int]:
if self._check_missing:
to_add = {o.url: o for o in origins}.values() to_add = {o.url: o for o in origins}.values()
origins = [ori for ori in to_add if self.origin_get_one(ori.url) is None] origins = [ori for ori in to_add if self.origin_get_one(ori.url) is None]
self.journal_writer.origin_add(origins) self.journal_writer.origin_add(origins)
for origin in origins: for origin in origins:
self._cql_runner.origin_add_one( self._cql_runner.origin_add_one(
OriginRow(sha1=hash_url(origin.url), url=origin.url, next_visit_id=1) OriginRow(sha1=hash_url(origin.url), url=origin.url, next_visit_id=1)
) )
return {"origin:add": len(origins)} return {"origin:add": len(origins)}
▲ Show 20 Lines • Show All 430 Lines • ▼ Show 20 Lines ) -> Optional[MetadataAuthority]:
url=authority.url, url=authority.url,
metadata=json.loads(authority.metadata), metadata=json.loads(authority.metadata),
) )
else: else:
return None return None
# ExtID tables # ExtID tables
def extid_add(self, ids: List[ExtID]) -> Dict[str, int]: def extid_add(self, ids: List[ExtID]) -> Dict[str, int]:
if self._check_missing:
extids = [ extids = [
extid extid
for extid in ids for extid in ids
if not self._cql_runner.extid_get_from_pk( if not self._cql_runner.extid_get_from_pk(
extid_type=extid.extid_type, extid=extid.extid, target=extid.target, extid_type=extid.extid_type, extid=extid.extid, target=extid.target,
) )
] ]
self.journal_writer.extid_add(extids) self.journal_writer.extid_add(extids)
inserted = 0 inserted = 0
for extid in extids: for extid in extids:
target_type = extid.target.object_type.value target_type = extid.target.object_type.value
target = extid.target.object_id target = extid.target.object_id
extidrow = ExtIDRow( extidrow = ExtIDRow(
▲ Show 20 Lines • Show All 60 Lines • Show Last 20 Lines
View Options

swh/web/add_forge_now/migrations/0002_authorized_null_comment.py

Show First 20 Lines • Show All 83 Lines • ▼ Show 20 Lines
) )
from .schema import HASH_ALGORITHMS from .schema import HASH_ALGORITHMS
# Max block size of contents to return # Max block size of contents to return
BULK_BLOCK_CONTENT_LEN_MAX = 10000 BULK_BLOCK_CONTENT_LEN_MAX = 10000
class CassandraStorage: class CassandraStorage:
def __init__(self, hosts, keyspace, objstorage, port=9042, journal_writer=None): def __init__(
self._cql_runner: CqlRunner = CqlRunner(hosts, keyspace, port) self,
hosts,
keyspace,
objstorage,
port=9042,
journal_writer=None,
check_missing=True,
):
"""
Args:
hosts: Seed Cassandra nodes, to start connecting to the cluster
keyspace: Name of the Cassandra database to use
objstorage: Passed as argument to :cls:`ObjStorage`
port: Cassandra port
journal_writer: Passed as argument to :cls:`JournalWriter`
check_missing: Whether ``*_add`` functions will check if an object already
exists in the database before sending it in an INSERT.
``True`` is the default as it is more efficient when there is a moderately
high probability the object is already known, but ``False`` can be
useful to overwrite existing objects (eg. when applying a schema update),
or when the database is known to be mostly empty.
"""
self._hosts = hosts
self._keyspace = keyspace
self._port = port
self._set_cql_runner()
self.journal_writer: JournalWriter = JournalWriter(journal_writer) self.journal_writer: JournalWriter = JournalWriter(journal_writer)
self.objstorage: ObjStorage = ObjStorage(objstorage) self.objstorage: ObjStorage = ObjStorage(objstorage)
self._check_missing = check_missing
def _set_cql_runner(self):
"""Used by tests when they need to reset the CqlRunner"""
self._cql_runner: CqlRunner = CqlRunner(self._hosts, self._keyspace, self._port)
def check_config(self, *, check_write: bool) -> bool: def check_config(self, *, check_write: bool) -> bool:
self._cql_runner.check_read() self._cql_runner.check_read()
return True return True
def _content_get_from_hash(self, algo, hash_) -> Iterable: def _content_get_from_hash(self, algo, hash_) -> Iterable:
"""From the name of a hash algorithm and a value of that hash, """From the name of a hash algorithm and a value of that hash,
Show All 10 Lines def _content_get_from_hash(self, algo, hash_) -> Iterable:
for row in res: for row in res:
# re-check the the hash (in case of murmur3 collision) # re-check the the hash (in case of murmur3 collision)
if getattr(row, algo) == hash_: if getattr(row, algo) == hash_:
yield row yield row
def _content_add(self, contents: List[Content], with_data: bool) -> Dict: def _content_add(self, contents: List[Content], with_data: bool) -> Dict:
# Filter-out content already in the database. # Filter-out content already in the database.
if self._check_missing:
contents = [ contents = [
c for c in contents if not self._cql_runner.content_get_from_pk(c.to_dict()) c
for c in contents
if not self._cql_runner.content_get_from_pk(c.to_dict())
] ]
if with_data: if with_data:
# First insert to the objstorage, if the endpoint is # First insert to the objstorage, if the endpoint is
# `content_add` (as opposed to `content_add_metadata`). # `content_add` (as opposed to `content_add_metadata`).
# Must add to the objstorage before the DB and journal. Otherwise: # Must add to the objstorage before the DB and journal. Otherwise:
# 1. in case of a crash the DB may "believe" we have the content, but # 1. in case of a crash the DB may "believe" we have the content, but
# we didn't have time to write to the objstorage before the crash # we didn't have time to write to the objstorage before the crash
Show All 12 Lines def _content_add(self, contents: List[Content], with_data: bool) -> Dict:
# Check for sha1 or sha1_git collisions. This test is not atomic # Check for sha1 or sha1_git collisions. This test is not atomic
# with the insertion, so it won't detect a collision if both # with the insertion, so it won't detect a collision if both
# contents are inserted at the same time, but it's good enough. # contents are inserted at the same time, but it's good enough.
# #
# The proper way to do it would probably be a BATCH, but this # The proper way to do it would probably be a BATCH, but this
# would be inefficient because of the number of partitions we # would be inefficient because of the number of partitions we
# need to affect (len(HASH_ALGORITHMS)+1, which is currently 5) # need to affect (len(HASH_ALGORITHMS)+1, which is currently 5)
if self._check_missing:
for algo in {"sha1", "sha1_git"}: for algo in {"sha1", "sha1_git"}:
collisions = [] collisions = []
# Get tokens of 'content' rows with the same value for # Get tokens of 'content' rows with the same value for
# sha1/sha1_git # sha1/sha1_git
rows = self._content_get_from_hash(algo, content.get_hash(algo)) rows = self._content_get_from_hash(algo, content.get_hash(algo))
for row in rows: for row in rows:
if getattr(row, algo) != content.get_hash(algo): if getattr(row, algo) != content.get_hash(algo):
# collision of token(partition key), ignore this # collision of token(partition key), ignore this
# row # row
continue continue
for other_algo in HASH_ALGORITHMS: for other_algo in HASH_ALGORITHMS:
if getattr(row, other_algo) != content.get_hash(other_algo): if getattr(row, other_algo) != content.get_hash(other_algo):
# This hash didn't match; discard the row. # This hash didn't match; discard the row.
collisions.append( collisions.append(
{k: getattr(row, k) for k in HASH_ALGORITHMS} {k: getattr(row, k) for k in HASH_ALGORITHMS}
) )
if collisions: if collisions:
collisions.append(content.hashes()) collisions.append(content.hashes())
raise HashCollision(algo, content.get_hash(algo), collisions) raise HashCollision(algo, content.get_hash(algo), collisions)
douarddaUnsubmitted
Not Done
Inline Actions

why is the whole collision detection block is now run only if check_missing is True?

douardda: why is the whole collision detection block is now run only if check_missing is True?
vlorentzAuthorUnsubmitted
Done
Inline Actions

Because, if an object already exists with the exact same hashes and it was not filtered out initially, then self._content_get_from_hash(algo, content.get_hash(algo)) will return it and it would be detected as a collision.

I could filter the results of self._content_get_from_hash, but I don't think it's worth it, as _check_missing is pretty niche anyway.

vlorentz: Because, if an object already exists with the exact same hashes and it was not filtered out…
vlorentzAuthorUnsubmitted
Done
Inline Actions

And collision detection in Cassandra isn't guaranteed anyway, because two colliding objects may be inserted at the same time and Cassandra is only eventually consistent.

vlorentz: And collision detection in Cassandra isn't guaranteed anyway, because two colliding objects may…
(token, insertion_finalizer) = self._cql_runner.content_add_prepare( (token, insertion_finalizer) = self._cql_runner.content_add_prepare(
ContentRow(**remove_keys(content.to_dict(), ("data",))) ContentRow(**remove_keys(content.to_dict(), ("data",)))
) )
# Then add to index tables # Then add to index tables
for algo in HASH_ALGORITHMS: for algo in HASH_ALGORITHMS:
self._cql_runner.content_index_add_one(algo, content, token) self._cql_runner.content_index_add_one(algo, content, token)
▲ Show 20 Lines • Show All 136 Lines • ▼ Show 20 Linesclass CassandraStorage:
def content_get_random(self) -> Sha1Git: def content_get_random(self) -> Sha1Git:
content = self._cql_runner.content_get_random() content = self._cql_runner.content_get_random()
assert content, "Could not find any content" assert content, "Could not find any content"
return content.sha1_git return content.sha1_git
def _skipped_content_add(self, contents: List[SkippedContent]) -> Dict: def _skipped_content_add(self, contents: List[SkippedContent]) -> Dict:
# Filter-out content already in the database. # Filter-out content already in the database.
if self._check_missing:
contents = [ contents = [
c c
for c in contents for c in contents
if not self._cql_runner.skipped_content_get_from_pk(c.to_dict()) if not self._cql_runner.skipped_content_get_from_pk(c.to_dict())
] ]
self.journal_writer.skipped_content_add(contents) self.journal_writer.skipped_content_add(contents)
for content in contents: for content in contents:
# Compute token of the row in the main table # Compute token of the row in the main table
(token, insertion_finalizer) = self._cql_runner.skipped_content_add_prepare( (token, insertion_finalizer) = self._cql_runner.skipped_content_add_prepare(
SkippedContentRow.from_dict({"origin": None, **content.to_dict()}) SkippedContentRow.from_dict({"origin": None, **content.to_dict()})
) )
Show All 15 Lines def skipped_content_missing(
self, contents: List[Dict[str, Any]] self, contents: List[Dict[str, Any]]
) -> Iterable[Dict[str, Any]]: ) -> Iterable[Dict[str, Any]]:
for content in contents: for content in contents:
if not self._cql_runner.skipped_content_get_from_pk(content): if not self._cql_runner.skipped_content_get_from_pk(content):
yield {algo: content[algo] for algo in DEFAULT_ALGORITHMS} yield {algo: content[algo] for algo in DEFAULT_ALGORITHMS}
def directory_add(self, directories: List[Directory]) -> Dict: def directory_add(self, directories: List[Directory]) -> Dict:
to_add = {d.id: d for d in directories}.values() to_add = {d.id: d for d in directories}.values()
if self._check_missing:
# Filter out directories that are already inserted. # Filter out directories that are already inserted.
missing = self.directory_missing([dir_.id for dir_ in to_add]) missing = self.directory_missing([dir_.id for dir_ in to_add])
directories = [dir_ for dir_ in directories if dir_.id in missing] directories = [dir_ for dir_ in directories if dir_.id in missing]
self.journal_writer.directory_add(directories) self.journal_writer.directory_add(directories)
for directory in directories: for directory in directories:
# Add directory entries to the 'directory_entry' table # Add directory entries to the 'directory_entry' table
for entry in directory.entries: for entry in directory.entries:
self._cql_runner.directory_entry_add_one( self._cql_runner.directory_entry_add_one(
DirectoryEntryRow(directory_id=directory.id, **entry.to_dict()) DirectoryEntryRow(directory_id=directory.id, **entry.to_dict())
▲ Show 20 Lines • Show All 106 Lines • ▼ Show 20 Linesclass CassandraStorage:
def directory_get_random(self) -> Sha1Git: def directory_get_random(self) -> Sha1Git:
directory = self._cql_runner.directory_get_random() directory = self._cql_runner.directory_get_random()
assert directory, "Could not find any directory" assert directory, "Could not find any directory"
return directory.id return directory.id
def revision_add(self, revisions: List[Revision]) -> Dict: def revision_add(self, revisions: List[Revision]) -> Dict:
# Filter-out revisions already in the database # Filter-out revisions already in the database
if self._check_missing:
to_add = {r.id: r for r in revisions}.values() to_add = {r.id: r for r in revisions}.values()
missing = self.revision_missing([rev.id for rev in to_add]) missing = self.revision_missing([rev.id for rev in to_add])
revisions = [rev for rev in revisions if rev.id in missing] revisions = [rev for rev in revisions if rev.id in missing]
self.journal_writer.revision_add(revisions) self.journal_writer.revision_add(revisions)
for revision in revisions: for revision in revisions:
revobject = converters.revision_to_db(revision) revobject = converters.revision_to_db(revision)
if revobject: if revobject:
# Add parents first # Add parents first
for (rank, parent) in enumerate(revision.parents): for (rank, parent) in enumerate(revision.parents):
self._cql_runner.revision_parent_add_one( self._cql_runner.revision_parent_add_one(
▲ Show 20 Lines • Show All 92 Lines • ▼ Show 20 Lines ) -> Iterable[Optional[Tuple[Sha1Git, Tuple[Sha1Git, ...]]]]:
yield from self._get_parent_revs(revisions, seen, limit, True) yield from self._get_parent_revs(revisions, seen, limit, True)
def revision_get_random(self) -> Sha1Git: def revision_get_random(self) -> Sha1Git:
revision = self._cql_runner.revision_get_random() revision = self._cql_runner.revision_get_random()
assert revision, "Could not find any revision" assert revision, "Could not find any revision"
return revision.id return revision.id
def release_add(self, releases: List[Release]) -> Dict: def release_add(self, releases: List[Release]) -> Dict:
if self._check_missing:
to_add = {r.id: r for r in releases}.values() to_add = {r.id: r for r in releases}.values()
missing = set(self.release_missing([rel.id for rel in to_add])) missing = set(self.release_missing([rel.id for rel in to_add]))
releases = [rel for rel in to_add if rel.id in missing] releases = [rel for rel in to_add if rel.id in missing]
self.journal_writer.release_add(releases) self.journal_writer.release_add(releases)
for release in releases: for release in releases:
if release: if release:
self._cql_runner.release_add_one(converters.release_to_db(release)) self._cql_runner.release_add_one(converters.release_to_db(release))
return {"release:add": len(releases)} return {"release:add": len(releases)}
Show All 10 Lines def release_get(self, releases: List[Sha1Git]) -> List[Optional[Release]]:
return [rels.get(rel_id) for rel_id in releases] return [rels.get(rel_id) for rel_id in releases]
def release_get_random(self) -> Sha1Git: def release_get_random(self) -> Sha1Git:
release = self._cql_runner.release_get_random() release = self._cql_runner.release_get_random()
assert release, "Could not find any release" assert release, "Could not find any release"
return release.id return release.id
def snapshot_add(self, snapshots: List[Snapshot]) -> Dict: def snapshot_add(self, snapshots: List[Snapshot]) -> Dict:
if self._check_missing:
to_add = {s.id: s for s in snapshots}.values() to_add = {s.id: s for s in snapshots}.values()
missing = self._cql_runner.snapshot_missing([snp.id for snp in to_add]) missing = self._cql_runner.snapshot_missing([snp.id for snp in to_add])
snapshots = [snp for snp in snapshots if snp.id in missing] snapshots = [snp for snp in snapshots if snp.id in missing]
for snapshot in snapshots: for snapshot in snapshots:
self.journal_writer.snapshot_add([snapshot]) self.journal_writer.snapshot_add([snapshot])
# Add branches # Add branches
for (branch_name, branch) in snapshot.branches.items(): for (branch_name, branch) in snapshot.branches.items():
if branch is None: if branch is None:
target_type: Optional[str] = None target_type: Optional[str] = None
▲ Show 20 Lines • Show All 252 Lines • ▼ Show 20 Linesclass CassandraStorage:
def origin_count( def origin_count(
self, url_pattern: str, regexp: bool = False, with_visit: bool = False self, url_pattern: str, regexp: bool = False, with_visit: bool = False
) -> int: ) -> int:
raise NotImplementedError( raise NotImplementedError(
"The Cassandra backend does not implement origin_count" "The Cassandra backend does not implement origin_count"
) )
def origin_add(self, origins: List[Origin]) -> Dict[str, int]: def origin_add(self, origins: List[Origin]) -> Dict[str, int]:
if self._check_missing:
to_add = {o.url: o for o in origins}.values() to_add = {o.url: o for o in origins}.values()
origins = [ori for ori in to_add if self.origin_get_one(ori.url) is None] origins = [ori for ori in to_add if self.origin_get_one(ori.url) is None]
self.journal_writer.origin_add(origins) self.journal_writer.origin_add(origins)
for origin in origins: for origin in origins:
self._cql_runner.origin_add_one( self._cql_runner.origin_add_one(
OriginRow(sha1=hash_url(origin.url), url=origin.url, next_visit_id=1) OriginRow(sha1=hash_url(origin.url), url=origin.url, next_visit_id=1)
) )
return {"origin:add": len(origins)} return {"origin:add": len(origins)}
▲ Show 20 Lines • Show All 430 Lines • ▼ Show 20 Lines ) -> Optional[MetadataAuthority]:
url=authority.url, url=authority.url,
metadata=json.loads(authority.metadata), metadata=json.loads(authority.metadata),
) )
else: else:
return None return None
# ExtID tables # ExtID tables
def extid_add(self, ids: List[ExtID]) -> Dict[str, int]: def extid_add(self, ids: List[ExtID]) -> Dict[str, int]:
if self._check_missing:
extids = [ extids = [
extid extid
for extid in ids for extid in ids
if not self._cql_runner.extid_get_from_pk( if not self._cql_runner.extid_get_from_pk(
extid_type=extid.extid_type, extid=extid.extid, target=extid.target, extid_type=extid.extid_type, extid=extid.extid, target=extid.target,
) )
] ]
self.journal_writer.extid_add(extids) self.journal_writer.extid_add(extids)
inserted = 0 inserted = 0
for extid in extids: for extid in extids:
target_type = extid.target.object_type.value target_type = extid.target.object_type.value
target = extid.target.object_id target = extid.target.object_id
extidrow = ExtIDRow( extidrow = ExtIDRow(
▲ Show 20 Lines • Show All 60 Lines • Show Last 20 Lines
View Options

swh/web/add_forge_now/migrations/0003_request_submitter_forward_username.py

Loading...
View Options

swh/web/add_forge_now/migrations/0004_rename_tables.py

Loading...
View Options

swh/web/add_forge_now/models.py

Loading...
View Options

swh/web/add_forge_now/tests/test_migration.py

Loading...
About · Developer information · Legal