Paths

Table of Contentst

-
swh/graphql/
-
graphql/
-
app.py
-
resolvers/
-
resolvers.py
-
scalars.py
-
schema/
-
schema.graphql
-
tests/
-
functional/
-
test_release_node.py
-
test_revision.py
-
unit/resolvers/
-
resolvers/
-
test_scalars.py

Differential D8425

Return more information with revision and release dates
ClosedPublic
Actions

Authored by jayeshv on Sep 8 2022, 3:40 PM.

Details

Reviewers

anlambert

Group Reviewers

Reviewers

Maniphest Tasks

T4365: GraphQL: Use the right date time serialization

Commits

rDGQL5446d4f87f85: Return more information with revision and release dates

Summary

Return more information with revision and release dates

Revision and release dates could have different values for offset.
Add a new schema type to handle these two dates and return available
offset as a binary string.

Related to T4365

Diff Detail

Repository

rDGQL GraphQL API

Branch

timezone-D8425

Lint

No Linters Available

Unit

No Unit Test Coverage

Build Status

Buildable 33071
Build 51837: Phabricator diff pipeline on jenkins	Jenkins console · Jenkins
Build 51836: arc lint + arc unit

Event Timeline

jayeshv created this revision.Sep 8 2022, 3:40 PM

Herald added a reviewer: Reviewers. · View Herald TranscriptSep 8 2022, 3:40 PM

remove unused doc

Build is green

Patch application report for D8425 (id=30376)

Rebasing onto 467e0b7e28...

Current branch diff-target is up to date.

Changes applied before test

commit 124ea6de94268cbf71354b6e89c82f715d7dde3c
Author: Jayesh Velayudhan <jayesh@softwareheritage.org>
Date:   Thu Sep 8 15:25:40 2022 +0200

    Return more information with revision and release dates
    
    Not all revision and release dates are convertable to datetime.
    Add a new schema type to handle these two dates and return all the
    availalble data.
    
    Related to T4365

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

Harbormaster completed remote builds in B31397: Diff 30376.Sep 8 2022, 3:42 PM

Build is green

Patch application report for D8425 (id=30377)

Rebasing onto 467e0b7e28...

Current branch diff-target is up to date.

Changes applied before test

commit 79660a4c7058d7471d8ff90ad21cb4a166fa18c7
Author: Jayesh Velayudhan <jayesh@softwareheritage.org>
Date:   Thu Sep 8 15:25:40 2022 +0200

    Return more information with revision and release dates
    
    Not all revision and release dates are convertable to datetime.
    Add a new schema type to handle these two dates and return all the
    availalble data.
    
    Related to T4365

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

Harbormaster completed remote builds in B31398: Diff 30377.Sep 8 2022, 3:43 PM

jayeshv requested review of this revision.Sep 8 2022, 3:43 PM

vlorentz added a subscriber: vlorentz.Sep 12 2022, 10:39 AM

vlorentz added inline comments.

swh/graphql/resolvers/scalars.py
31–32	use `isinstance`
36	What other types can `value` be?
swh/graphql/schema/schema.graphql
566	the doc should explain why there are both an ISO8601 date and timestamp + offset.
568	why not just `Date`?

Add types and support for None date

Build is green

Patch application report for D8425 (id=30439)

Rebasing onto 6e5d34e94d...

First, rewinding head to replay your work on top of it...
Applying: Return more information with revision and release dates

Changes applied before test

commit 09665f45133082d4a0447022bd1b0f921f180e50
Author: Jayesh Velayudhan <jayesh@softwareheritage.org>
Date:   Thu Sep 8 15:25:40 2022 +0200

    Return more information with revision and release dates
    
    Not all revision and release dates are convertable to datetime.
    Add a new schema type to handle these two dates and return all the
    availalble data.
    
    Related to T4365

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

Harbormaster completed remote builds in B31455: Diff 30439.Sep 12 2022, 11:56 AM

jayeshv retitled this revision from Return more information with revision and release dates to [WIP]: Return more information with revision and release dates.Sep 12 2022, 11:56 AM

use a timestamp instead of seconds and microseconds

jayeshv marked an inline comment as done.Sep 12 2022, 12:28 PM

jayeshv added inline comments.

swh/graphql/resolvers/resolvers.py
331	@vlorentz changed this to use a timestamp instead of seconds and microseconds. What do you think?
swh/graphql/resolvers/scalars.py
36	Can be an instance of TimestampWithTimezone or a python datetime. Any other type will return a None. I don't really know the input (value) can be None, but that is anyway supported.

Build is green

Patch application report for D8425 (id=30441)

Rebasing onto 6e5d34e94d...

First, rewinding head to replay your work on top of it...
Applying: Return more information with revision and release dates

Changes applied before test

commit 41d660696bc4268a1ba8f934ea51bec06967cbbb
Author: Jayesh Velayudhan <jayesh@softwareheritage.org>
Date:   Thu Sep 8 15:25:40 2022 +0200

    Return more information with revision and release dates
    
    Not all revision and release dates are convertable to datetime.
    Add a new schema type to handle these two dates and return all the
    availalble data.
    
    Related to T4365

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

Harbormaster completed remote builds in B31457: Diff 30441.Sep 12 2022, 12:30 PM

vlorentz added inline comments.Sep 12 2022, 1:33 PM

swh/graphql/resolvers/resolvers.py
331	Using floats loses precision for very high or low timestamps; but I guess that's fine
swh/graphql/resolvers/scalars.py
36	why not two different functions, one for each type?

jayeshv added inline comments.Sep 12 2022, 2:23 PM

swh/graphql/resolvers/scalars.py
36	This same function is used to serialize datetimes from both user generated (date field in the Date object) and db dates (date in revision object). adding two different scalars for both types is perfectly fine, but wouldn't that make a client a bit more complex? (I was trying to keep the number of scalars to a minimum)

jayeshv added inline comments.Sep 12 2022, 2:43 PM

swh/graphql/resolvers/scalars.py
36	Fix the typo This same function is used to serialize datetimes from both user generated (date field in the Date object) and db dates (date in the visit status object). adding two different scalars for both types is perfectly fine, but wouldn't that make a client a bit more complex? (I was trying to keep the number of scalars to a minimum)

test fixes

Build is green

Patch application report for D8425 (id=31584)

Rebasing onto 7522a848fa...

Current branch diff-target is up to date.

Changes applied before test

commit 85c9bf459f242c103e46946cb043d66d0648b59f
Author: Jayesh Velayudhan <jayesh@softwareheritage.org>
Date:   Thu Sep 8 15:25:40 2022 +0200

    Return more information with revision and release dates
    
    Not all revision and release dates are convertable to datetime.
    Add a new schema type to handle these two dates and return all the
    availalble data.
    
    Related to T4365

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

Harbormaster completed remote builds in B32538: Diff 31584.Oct 25 2022, 9:29 AM

bug fix

Build is green

Patch application report for D8425 (id=32118)

Rebasing onto c28b65feaf...

Current branch diff-target is up to date.

Changes applied before test

commit 5e8890d818d11314c674bb742898905ac5bdce92
Author: Jayesh Velayudhan <jayesh@softwareheritage.org>
Date:   Thu Sep 8 15:25:40 2022 +0200

    Return more information with revision and release dates
    
    Not all revision and release dates are convertable to datetime.
    Add a new schema type to handle these two dates and return all the
    availalble data.
    
    Related to T4365

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

Harbormaster completed remote builds in B33071: Diff 32118.Dec 2 2022, 3:44 PM

jayeshv retitled this revision from [WIP]: Return more information with revision and release dates to Return more information with revision and release dates.Dec 12 2022, 3:32 PM

rebase

jayeshv retitled this revision from Return more information with revision and release dates to [WIP] Return more information with revision and release dates.Dec 20 2022, 10:10 AM

Build is green

Patch application report for D8425 (id=32323)

Rebasing onto c8bc6947fa...

First, rewinding head to replay your work on top of it...
Applying: Return more information with revision and release dates

Changes applied before test

commit d389e69c86db19bed90938384e2e638120fed3b9
Author: Jayesh Velayudhan <jayesh@softwareheritage.org>
Date:   Thu Sep 8 15:25:40 2022 +0200

    Return more information with revision and release dates
    
    Not all revision and release dates are convertable to datetime.
    Add a new schema type to handle these two dates and return all the
    availalble data.
    
    Related to T4365

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

Harbormaster completed remote builds in B33314: Diff 32323.Dec 20 2022, 10:12 AM

Change offset to a binary string

jayeshv edited the summary of this revision. (Show Details)Dec 20 2022, 4:00 PM

Build is green

Patch application report for D8425 (id=32332)

Rebasing onto c8bc6947fa...

First, rewinding head to replay your work on top of it...
Applying: Return more information with revision and release dates

Changes applied before test

commit ec9873527f0afbb8e500a51173d2b2b847f536e5
Author: Jayesh Velayudhan <jayesh@softwareheritage.org>
Date:   Thu Sep 8 15:25:40 2022 +0200

    Return more information with revision and release dates
    
    Revision and release dates could have different values for offset.
    Add a new schema type to handle these two dates and return available
    offset as a binary string.
    
    Related to T4365

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

Harbormaster completed remote builds in B33323: Diff 32332.Dec 20 2022, 4:02 PM

jayeshv retitled this revision from [WIP] Return more information with revision and release dates to Return more information with revision and release dates.Dec 20 2022, 4:03 PM

Add missing test

Build is green

Patch application report for D8425 (id=32333)

Rebasing onto c8bc6947fa...

First, rewinding head to replay your work on top of it...
Applying: Return more information with revision and release dates

Changes applied before test

commit 49ee60bf79d6973b61b76d07033f4e0911148930
Author: Jayesh Velayudhan <jayesh@softwareheritage.org>
Date:   Thu Sep 8 15:25:40 2022 +0200

    Return more information with revision and release dates
    
    Revision and release dates could have different values for offset.
    Add a new schema type to handle these two dates and return available
    offset as a binary string.
    
    Related to T4365

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

Harbormaster completed remote builds in B33324: Diff 32333.Dec 20 2022, 4:12 PM

LGTM but you also need to wrap the committer_date field from Revision model.

swh/graphql/schema/schema.graphql
622	I just noticed that you forgot to wrap the commiter_date field from the revision model. You should add it as that info is important to have.

This revision now requires changes to proceed.Dec 21 2022, 11:04 AM

Add committer_date im the revision object

jayeshv marked an inline comment as done.Dec 21 2022, 11:20 AM

jayeshv added inline comments.

swh/graphql/schema/schema.graphql
622	Thanks, added that.

Build is green

Patch application report for D8425 (id=32340)

Rebasing onto c8bc6947fa...

First, rewinding head to replay your work on top of it...
Applying: Return more information with revision and release dates

Changes applied before test

commit 7995e5a4858a71a54a1ac853caabe3a357842215
Author: Jayesh Velayudhan <jayesh@softwareheritage.org>
Date:   Thu Sep 8 15:25:40 2022 +0200

    Return more information with revision and release dates
    
    Revision and release dates could have different values for offset.
    Add a new schema type to handle these two dates and return available
    offset as a binary string.
    
    Related to T4365

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

Harbormaster completed remote builds in B33327: Diff 32340.Dec 21 2022, 11:21 AM

anlambert accepted this revision.Dec 21 2022, 12:03 PM

This revision is now accepted and ready to land.Dec 21 2022, 12:03 PM

rebase

Build is green

Patch application report for D8425 (id=32341)

Rebasing onto c8bc6947fa...

Current branch diff-target is up to date.

Changes applied before test

commit 5446d4f87f85fdb3b942e50e935dc2ef6c6e8d29
Author: Jayesh Velayudhan <jayesh@softwareheritage.org>
Date:   Thu Sep 8 15:25:40 2022 +0200

    Return more information with revision and release dates
    
    Revision and release dates could have different values for offset.
    Add a new schema type to handle these two dates and return available
    offset as a binary string.
    
    Related to T4365

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

Harbormaster completed remote builds in B33328: Diff 32341.Dec 21 2022, 12:18 PM

Closed by commit rDGQL5446d4f87f85: Return more information with revision and release dates (authored by jayeshv). · Explain WhyDec 21 2022, 12:18 PM

This revision was automatically updated to reflect the committed changes.

jayeshv added a commit: rDGQL5446d4f87f85: Return more information with revision and release dates.

Revision Contents
Changeset List

Path

Size

Coverage (All)

Coverage (Touched)

swh/

graphql/

app.py

1 line

100%

resolvers/

resolvers.py

33 lines

97%

scalars.py

20 lines

100%

schema/

schema.graphql

30 lines

tests/

functional/

test_release_node.py

8 lines

100%

test_revision.py

13 lines

100%

unit/

resolvers/

test_scalars.py

35 lines

100%

Diff 32118

View Options

swh/graphql/app.py

# Copyright (C) 2021 The Software Heritage developers		# Copyright (C) 2021 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution		# See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version		# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information		# See top-level LICENSE file for more information

from __future__ import annotations		from __future__ import annotations

from datetime import datetime, timezone		from datetime import datetime, timezone
import os		import os
from types import TracebackType		from types import TracebackType
from typing import Any, Dict, Generator, Iterable, List, Optional, Set, Type, Union		from typing import Any, Dict, Generator, Iterable, List, Optional, Set, Type, Union

from bson import ObjectId		from bson import ObjectId
import mongomock		import mongomock
import pymongo		import pymongo
		from pymongo import InsertOne, UpdateOne
		import sys

from swh.core.statsd import statsd		from swh.core.statsd import statsd
from swh.model.model import Sha1Git		from swh.model.model import Sha1Git

from ..interface import (		from ..interface import (
EntityType,		EntityType,
ProvenanceResult,		ProvenanceResult,
ProvenanceStorageInterface,		ProvenanceStorageInterface,
RelationData,		RelationData,
RelationType,		RelationType,
RevisionData,		RevisionData,
)		)
		from .entity import Entity

STORAGE_DURATION_METRIC = "swh_provenance_storage_mongodb_duration_seconds"		STORAGE_DURATION_METRIC = "swh_provenance_storage_mongodb_duration_seconds"


class ProvenanceStorageMongoDb:		class ProvenanceStorageMongoDb:
def __init__(self, engine: str, **kwargs):		def __init__(self, engine: str, **kwargs):
self.engine = engine		self.engine = engine
self.dbname = kwargs.pop("dbname")		self.dbname = kwargs.pop("dbname")
Show All 10 Lines	def __exit__(
exc_tb: Optional[TracebackType],		exc_tb: Optional[TracebackType],
) -> None:		) -> None:
self.close()		self.close()

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "close"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "close"})
def close(self) -> None:		def close(self) -> None:
self.db.client.close()		self.db.client.close()

		def _generate_date_upserts(self, sha1, date, inserts={}, upsert=True):
		ts = datetime.timestamp(date) if date is not None else None
		# update only those with date either as None or later than the given one
		return UpdateOne(
		{"sha1": sha1, "$or": [{"ts": {"$gte": ts}}, {"ts": None}]},
		{
		"$set": {"ts": ts},
		"$setOnInsert": inserts,
		},
		upsert=upsert,
		)

		def _generate_date_upserts_alternate(self, sha1, date, inserts={}, upsert=True):
		# FIXME, Compare the performance and decide which one to keep
		ts = datetime.timestamp(date) if date is not None else None
		# update only those with date either as None or later than the given one
		return UpdateOne(
		{"sha1": sha1},
		[
		{
		"$set": {
		"ts": {
		"$cond": {
		"if": {"$gt": ["$ts", ts]},
		"then": ts,
		"else": None,
		}
		}
		},
		},
		],
		upsert=upsert,
		)

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_add"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_add"})
def content_add(		def content_add(
self, cnts: Union[Iterable[Sha1Git], Dict[Sha1Git, Optional[datetime]]]		self, cnts: Union[Iterable[Sha1Git], Dict[Sha1Git, Optional[datetime]]]
) -> bool:		) -> bool:
data = cnts if isinstance(cnts, dict) else dict.fromkeys(cnts)		default_inserts = {"revision": {}, "directory": {}}
existing = {		if isinstance(cnts, dict):
x["sha1"]: x		writes = [
for x in self.db.content.find(		self._generate_date_upserts(sha1, date, default_inserts)
{"sha1": {"$in": list(data)}}, {"sha1": 1, "ts": 1, "_id": 1}		for sha1, date in cnts.items()
)		]
}		else:
for sha1, date in data.items():		writes = [
ts = datetime.timestamp(date) if date is not None else None		self._generate_date_upserts(sha1, None, default_inserts)
if sha1 in existing:		for sha1 in cnts
cnt = existing[sha1]		]
if ts is not None and (cnt["ts"] is None or ts < cnt["ts"]):		statsd.gauge("provenance.content-list.size", sys.getsizeof(writes))
self.db.content.update_one(		Entity.factory(self.db, "content").bulk_write(writes)
{"_id": cnt["_id"]}, {"$set": {"ts": ts}}		return True
)
else:		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "directory_add"})
self.db.content.insert_one(		def directory_add(
		self, dirs: Union[Iterable[Sha1Git], Dict[Sha1Git, Optional[datetime]]]
		) -> bool:
		default_inserts = {"revision": {}}
		if isinstance(dirs, dict):
		writes = [
		self._generate_date_upserts(sha1, date, default_inserts)
		for sha1, date in dirs.items()
		]
		else:
		writes = [
		self._generate_date_upserts(sha1, None, default_inserts)
		for sha1 in dirs
		]
		statsd.gauge("provenance.directory-list.size", sys.getsizeof(writes))
		Entity.factory(self.db, "directory").bulk_write(writes)
		return True

		def _get_oldest_revision_from_content(self, content):
		# FIXME, returing with the assumption that content has all the
		# revisons in the array
		# Change to seperate collection content_in_revision if gets too big
		return Entity.factory(self.db, "revision").find_one(
{		{
"sha1": sha1,		"_id": {"$in": [ObjectId(obj_id) for obj_id in content["revision"]]},
"ts": ts,		"ts": content["ts"],
"revision": {},
"directory": {},
}		}
)		)
return True
		def _get_preferred_origin(self, revision):
		if revision.get("preferred"):
		return Entity.factory(self.db, "origin").find_one(
		{"sha1": revision["preferred"]}
		)
		return None

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_find_first"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_find_first"})
def content_find_first(self, id: Sha1Git) -> Optional[ProvenanceResult]:		def content_find_first(self, id: Sha1Git) -> Optional[ProvenanceResult]:
		content = Entity.factory(self.db, "content").find_one({"sha1": id})
		if not content:
		return None

		oldest_revision = self._get_oldest_revision_from_content(content)
		origin = self._get_preferred_origin(oldest_revision)

		return ProvenanceResult(
		content=id,
		revision=oldest_revision["sha1"],
		date=datetime.fromtimestamp(oldest_revision["ts"], timezone.utc),
		origin=origin["url"] if origin else None,
		path=sorted(content["revision"][str(oldest_revision["_id"])])[0]
		# FIXME, this is likely to be wrong, tests pass
		)

		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_find_first"})
		def content_find_first_old(self, id: Sha1Git) -> Optional[ProvenanceResult]:
# get all the revisions		# get all the revisions
# iterate and find the earliest		# iterate and find the earliest
content = self.db.content.find_one({"sha1": id})		content = self.db.content.find_one({"sha1": id})
if not content:		if not content:
return None		return None

occurs = []		occurs = []
for revision in self.db.revision.find(		for revision in self.db.revision.find(
{"_id": {"$in": [ObjectId(obj_id) for obj_id in content["revision"]]}}		{"_id": {"$in": [ObjectId(obj_id) for obj_id in content["revision"]]}}
):		):
if revision["preferred"] is not None:		if revision["preferred"] is not None:
origin = self.db.origin.find_one({"sha1": revision["preferred"]})		origin = self.db.origin.find_one({"sha1": revision["preferred"]})
else:		else:
origin = {"url": None}		origin = {"url": None}

for path in content["revision"][str(revision["_id"])]:		for path in content["revision"][str(revision["_id"])]:
occurs.append(		occurs.append(
ProvenanceResult(		ProvenanceResult(
content=id,		content=id,
revision=revision["sha1"],		revision=revision["sha1"],
date=datetime.fromtimestamp(revision["ts"], timezone.utc),		date=datetime.fromtimestamp(revision["ts"], timezone.utc),
origin=origin["url"],		origin=origin["url"],
path=path,		path=path,
)		)
)		)
return sorted(occurs, key=lambda x: (x.date, x.revision, x.origin, x.path))[0]		return sorted(occurs, key=lambda x: (x.date, x.revision, x.origin, x.path))[0]

		# FIXME, refactor this
@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_find_all"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_find_all"})
def content_find_all(		def content_find_all(
self, id: Sha1Git, limit: Optional[int] = None		self, id: Sha1Git, limit: Optional[int] = None
) -> Generator[ProvenanceResult, None, None]:		) -> Generator[ProvenanceResult, None, None]:
content = self.db.content.find_one({"sha1": id})		content = self.db.content.find_one({"sha1": id})
if not content:		if not content:
return None		return None

Show All 12 Lines	) -> Generator[ProvenanceResult, None, None]:
content=id,		content=id,
revision=revision["sha1"],		revision=revision["sha1"],
date=datetime.fromtimestamp(revision["ts"], timezone.utc),		date=datetime.fromtimestamp(revision["ts"], timezone.utc),
origin=origin["url"],		origin=origin["url"],
path=path,		path=path,
)		)
)		)
for directory in self.db.directory.find(		for directory in self.db.directory.find(
{"_id": {"$in": [ObjectId(obj_id) for obj_id in content["directory"]]}}		{
		"_id": {
		"$in": [ObjectId(obj_id) for obj_id in content.get("directory", {})]
		}
		}
):		):
for revision in self.db.revision.find(		for revision in self.db.revision.find(
{"_id": {"$in": [ObjectId(obj_id) for obj_id in directory["revision"]]}}		{"_id": {"$in": [ObjectId(obj_id) for obj_id in directory["revision"]]}}
):		):
if revision["preferred"] is not None:		if revision["preferred"] is not None:
origin = self.db.origin.find_one({"sha1": revision["preferred"]})		origin = self.db.origin.find_one({"sha1": revision["preferred"]})
else:		else:
origin = {"url": None}		origin = {"url": None}
Show All 16 Lines	) -> Generator[ProvenanceResult, None, None]:
path=path,		path=path,
)		)
)		)
yield from sorted(occurs, key=lambda x: (x.date, x.revision, x.origin, x.path))		yield from sorted(occurs, key=lambda x: (x.date, x.revision, x.origin, x.path))

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_get"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_get"})
def content_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, datetime]:		def content_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, datetime]:
return {		return {
x["sha1"]: datetime.fromtimestamp(x["ts"], timezone.utc)		cnt["sha1"]: datetime.fromtimestamp(cnt["ts"], timezone.utc)
for x in self.db.content.find(		for cnt in Entity.factory(self.db, "content").find(
{"sha1": {"$in": list(ids)}, "ts": {"$ne": None}},		{"sha1": {"$in": list(ids)}, "ts": {"$ne": None}},
{"sha1": 1, "ts": 1, "_id": 0},		{"sha1": 1, "ts": 1, "_id": 0},
)		)
}		}

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "directory_add"})
def directory_add(
self, dirs: Union[Iterable[Sha1Git], Dict[Sha1Git, Optional[datetime]]]
) -> bool:
data = dirs if isinstance(dirs, dict) else dict.fromkeys(dirs)
existing = {
x["sha1"]: x
for x in self.db.directory.find(
{"sha1": {"$in": list(data)}}, {"sha1": 1, "ts": 1, "_id": 1}
)
}
for sha1, date in data.items():
ts = datetime.timestamp(date) if date is not None else None
if sha1 in existing:
dir = existing[sha1]
if ts is not None and (dir["ts"] is None or ts < dir["ts"]):
self.db.directory.update_one(
{"_id": dir["_id"]}, {"$set": {"ts": ts}}
)
else:
self.db.directory.insert_one({"sha1": sha1, "ts": ts, "revision": {}})
return True

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "directory_get"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "directory_get"})
def directory_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, datetime]:		def directory_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, datetime]:
return {		return {
x["sha1"]: datetime.fromtimestamp(x["ts"], timezone.utc)		dry["sha1"]: datetime.fromtimestamp(dry["ts"], timezone.utc)
for x in self.db.directory.find(		for dry in Entity.factory(self.db, "directory").find(
{"sha1": {"$in": list(ids)}, "ts": {"$ne": None}},		{"sha1": {"$in": list(ids)}, "ts": {"$ne": None}},
{"sha1": 1, "ts": 1, "_id": 0},		{"sha1": 1, "ts": 1, "_id": 0},
)		)
}		}

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "entity_get_all"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "entity_get_all"})
def entity_get_all(self, entity: EntityType) -> Set[Sha1Git]:		def entity_get_all(self, entity: EntityType) -> Set[Sha1Git]:
		# test only method; not using entity factory or cache
return {		return {
x["sha1"]		x["sha1"]
for x in self.db.get_collection(entity.value).find(		for x in self.db.get_collection(entity.value).find(
{}, {"sha1": 1, "_id": 0}		{}, {"sha1": 1, "_id": 0}
)		)
}		}

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "location_add"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "location_add"})
def location_add(self, paths: Iterable[bytes]) -> bool:		def location_add(self, paths: Iterable[bytes]) -> bool:
# TODO: implement this methods if path are to be stored in a separate collection		# TODO: implement this methods if path are to be stored in a separate collection
return True		return True

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "location_get_all"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "location_get_all"})
def location_get_all(self) -> Set[bytes]:		def location_get_all(self) -> Set[bytes]:
		# test only method; not using entity factory or cache
contents = self.db.content.find({}, {"revision": 1, "_id": 0, "directory": 1})		contents = self.db.content.find({}, {"revision": 1, "_id": 0, "directory": 1})
paths: List[Iterable[bytes]] = []		paths: List[Iterable[bytes]] = []
for content in contents:		for content in contents:
paths.extend(value for _, value in content["revision"].items())		paths.extend(value for _, value in content["revision"].items())
paths.extend(value for _, value in content["directory"].items())		paths.extend(value for _, value in content["directory"].items())

dirs = self.db.directory.find({}, {"revision": 1, "_id": 0})		dirs = self.db.directory.find({}, {"revision": 1, "_id": 0})
for each_dir in dirs:		for each_dir in dirs:
paths.extend(value for _, value in each_dir["revision"].items())		paths.extend(value for _, value in each_dir["revision"].items())
return set(sum(paths, []))		return set(sum(paths, []))

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "open"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "open"})
def open(self) -> None:		def open(self) -> None:
if self.engine == "mongomock":		if self.engine == "mongomock":
self.db = mongomock.MongoClient(**self.conn_args).get_database(self.dbname)		self.db = mongomock.MongoClient(**self.conn_args).get_database(self.dbname)
else:		else:
# assume real MongoDB server by default		# assume real MongoDB server by default
self.db = pymongo.MongoClient(**self.conn_args).get_database(self.dbname)		self.db = pymongo.MongoClient(host="127.0.0.1", port=27017, connect=False).get_database(self.dbname)

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "origin_add"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "origin_add"})
def origin_add(self, orgs: Dict[Sha1Git, str]) -> bool:		def origin_add(self, orgs: Dict[Sha1Git, str]) -> bool:
existing = {		writes = [InsertOne({"sha1": sha1, "url": url}) for (sha1, url) in orgs.items()]
x["sha1"]: x		Entity.factory(self.db, "origin").bulk_write(writes)
for x in self.db.origin.find(
{"sha1": {"$in": list(orgs)}}, {"sha1": 1, "url": 1, "_id": 1}
)
}
for sha1, url in orgs.items():
if sha1 not in existing:
# add new origin
self.db.origin.insert_one({"sha1": sha1, "url": url})
return True		return True

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "origin_get"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "origin_get"})
def origin_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, str]:		def origin_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, str]:
return {		return {
x["sha1"]: x["url"]		ori["sha1"]: ori["url"]
for x in self.db.origin.find(		for ori in Entity.factory(self.db, "origin").find(
{"sha1": {"$in": list(ids)}}, {"sha1": 1, "url": 1, "_id": 0}		{"sha1": {"$in": list(ids)}}, {"sha1": 1, "url": 1, "_id": 0}
)		)
}		}

		def _generate_revision_upserts(self, sha1, date=None, preferred=None, upsert=True):
		ts = datetime.timestamp(date) if date is not None else None
		return UpdateOne(
		{"sha1": sha1, "$or": [{"ts": {"$gte": ts}}, {"ts": None}]},
		{
		"$set": {"ts": ts, "preferred": preferred},
		"$setOnInsert": {"origin": [], "revision": []},
		},
		upsert=upsert,
		)

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "revision_add"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "revision_add"})
def revision_add(		def revision_add(
self, revs: Union[Iterable[Sha1Git], Dict[Sha1Git, RevisionData]]		self, revs: Union[Iterable[Sha1Git], Dict[Sha1Git, RevisionData]]
) -> bool:		) -> bool:
		# if isinstance(revs, dict):
		# writes = [
		# self._generate_revision_upserts(sha1, data.date, data.origin)
		# for sha1, data in revs.items()
		# ]
		# else:
		# writes = [
		# self._generate_revision_upserts(sha1)
		# for sha1 in cnts
		# ]
		# Entity.factory(self.db, "revision").bulk_write(writes)
		# return True

data = (		data = (
revs		revs
if isinstance(revs, dict)		if isinstance(revs, dict)
else dict.fromkeys(revs, RevisionData(date=None, origin=None))		else dict.fromkeys(revs, RevisionData(date=None, origin=None))
)		)
existing = {		existing = {
x["sha1"]: x		x["sha1"]: x
for x in self.db.revision.find(		for x in self.db.revision.find(
{"sha1": {"$in": list(data)}},		{"sha1": {"$in": list(data)}},
{"sha1": 1, "ts": 1, "preferred": 1, "_id": 1},		{"sha1": 1, "ts": 1, "preferred": 1, "_id": 1},
)		)
}		}

		statsd.gauge("provenance.revision-cache.size", sys.getsizeof(existing))

for sha1, info in data.items():		for sha1, info in data.items():
ts = datetime.timestamp(info.date) if info.date is not None else None		ts = datetime.timestamp(info.date) if info.date is not None else None
preferred = info.origin		preferred = info.origin
if sha1 in existing:		if sha1 in existing:
rev = existing[sha1]		rev = existing[sha1]
if ts is None or (rev["ts"] is not None and ts >= rev["ts"]):		if ts is None or (rev["ts"] is not None and ts >= rev["ts"]):
ts = rev["ts"]		ts = rev["ts"]
if preferred is None:		if preferred is None:
Show All 13 Lines	) -> bool:
"ts": ts,		"ts": ts,
}		}
)		)
return True		return True

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "revision_get"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "revision_get"})
def revision_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, RevisionData]:		def revision_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, RevisionData]:
return {		return {
x["sha1"]: RevisionData(		rev["sha1"]: RevisionData(
date=datetime.fromtimestamp(x["ts"], timezone.utc) if x["ts"] else None,		date=datetime.fromtimestamp(rev["ts"], timezone.utc)
origin=x["preferred"],		if rev["ts"]
		else None,
		origin=rev["preferred"],
)		)
for x in self.db.revision.find(		for rev in Entity.factory(self.db, "revision").find(
{		{
"sha1": {"$in": list(ids)},		"sha1": {"$in": list(ids)},
"$or": [{"preferred": {"$ne": None}}, {"ts": {"$ne": None}}],		"$or": [{"preferred": {"$ne": None}}, {"ts": {"$ne": None}}],
},		},
{"sha1": 1, "preferred": 1, "ts": 1, "_id": 0},		{"sha1": 1, "preferred": 1, "ts": 1, "_id": 0},
)		)
}		}

Show All 27 Lines	) -> bool:

src_objs = {		src_objs = {
x["sha1"]: x		x["sha1"]: x
for x in self.db.get_collection(src_relation).find(		for x in self.db.get_collection(src_relation).find(
{"sha1": {"$in": list(denorm.keys())}}		{"sha1": {"$in": list(denorm.keys())}}
)		)
}		}

		statsd.gauge("provenance.relation-src-cache.size", sys.getsizeof(src_objs))
		statsd.gauge("provenance.relation-dst-cache.size", sys.getsizeof(dst_objs))

for sha1, dsts in denorm.items():		for sha1, dsts in denorm.items():
# update		# update
if src_relation != "revision":		if src_relation != "revision":
k = {		k = {
obj_id: list(set(paths + dsts.get(obj_id, [])))		obj_id: list(set(paths + dsts.get(obj_id, [])))
for obj_id, paths in src_objs[sha1][dst_relation].items()		for obj_id, paths in src_objs[sha1].get(dst_relation, {}).items()
}		}
self.db.get_collection(src_relation).update_one(		self.db.get_collection(src_relation).update_one(
{"_id": src_objs[sha1]["_id"]},		{"_id": src_objs[sha1]["_id"]},
{"$set": {dst_relation: dict(dsts, **k)}},		{"$set": {dst_relation: dict(dsts, **k)}},
)		)
else:		else:
self.db.get_collection(src_relation).update_one(		self.db.get_collection(src_relation).update_one(
{"_id": src_objs[sha1]["_id"]},		{"_id": src_objs[sha1]["_id"]},
▲ Show 20 Lines • Show All 53 Lines • ▼ Show 20 Lines	) -> Dict[Sha1Git, Set[RelationData]]:
else:		else:
dst_objs = {		dst_objs = {
x["sha1"]: x["_id"]		x["sha1"]: x["_id"]
for x in self.db.get_collection(dst).find(		for x in self.db.get_collection(dst).find(
{"sha1": {"$in": list(sha1s)}}, {"_id": 1, "sha1": 1}		{"sha1": {"$in": list(sha1s)}}, {"_id": 1, "sha1": 1}
)		)
}		}
src_objs = {		src_objs = {
x["sha1"]: x[dst]		x["sha1"]: x.get(dst, {})
for x in self.db.get_collection(src).find(		for x in self.db.get_collection(src).find(
{}, {"_id": 0, "sha1": 1, dst: 1}		{}, {"_id": 0, "sha1": 1, dst: 1}
)		)
}		}
result: Dict[Sha1Git, Set[RelationData]] = {}		result: Dict[Sha1Git, Set[RelationData]] = {}
if src != "revision":		if src != "revision":
for dst_sha1, dst_obj_id in dst_objs.items():		for dst_sha1, dst_obj_id in dst_objs.items():
for src_sha1, denorm in src_objs.items():		for src_sha1, denorm in src_objs.items():
Show All 13 Lines	) -> Dict[Sha1Git, Set[RelationData]]:
RelationData(dst=dst_sha1, path=None)		RelationData(dst=dst_sha1, path=None)
)		)
return result		return result

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "relation_get_all"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "relation_get_all"})
def relation_get_all(		def relation_get_all(
self, relation: RelationType		self, relation: RelationType
) -> Dict[Sha1Git, Set[RelationData]]:		) -> Dict[Sha1Git, Set[RelationData]]:
		# test only method; not using entity factory or cache
src, *_, dst = relation.value.split("_")		src, *_, dst = relation.value.split("_")
empty: Union[Dict[str, bytes], List[str]] = {} if src != "revision" else []		empty: Union[Dict[str, bytes], List[str]] = {} if src != "revision" else []
src_objs = {		src_objs = {
x["sha1"]: x[dst]		x["sha1"]: x.get(dst, {})
for x in self.db.get_collection(src).find(		for x in self.db.get_collection(src).find(
{dst: {"$ne": empty}}, {"_id": 0, "sha1": 1, dst: 1}		{dst: {"$ne": empty}}, {"_id": 0, "sha1": 1, dst: 1}
)		)
}		}
dst_ids = list(		dst_ids = list(
{ObjectId(obj_id) for _, value in src_objs.items() for obj_id in value}		{ObjectId(obj_id) for _, value in src_objs.items() for obj_id in value}
)		)
dst_objs = {		dst_objs = {
Show All 22 Lines	) -> Dict[Sha1Git, Set[RelationData]]:
if dst_obj_id == dst_obj_ref		if dst_obj_id == dst_obj_ref
}		}
for src_sha1, denorm in src_objs.items()		for src_sha1, denorm in src_objs.items()
}		}

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "with_path"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "with_path"})
def with_path(self) -> bool:		def with_path(self) -> bool:
return True		return True

		def entity_get(self, entity, filters=None):
		"""
		A method used only by tests
		Returns a list of entities after applying the filter
		"""
		# FIXME, this should be added to the interface to make
		# tests backend agnostic. Adding this in SQL could be tricky
		# FIXME, change to entity factory
		return list(Entity.factory(self.db, entity).find(filters))

		def entity_add(self, entity, data):
		"""
		A method used only by tests
		"""
		# FIXME, this should be added to the interface to make
		# tests backend agnostic. Adding this in SQL could be tricky
		# FIXME, change to entity factory
		return Entity.factory(self.db, entity).insert(data)

		def create_indexes(self, entity, fields, unique=False):
		Entity.factory(self.db, entity).create_index(fields, unique)

View Options

swh/graphql/resolvers/resolvers.py

# Copyright (C) 2021 The Software Heritage developers		# Copyright (C) 2021 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution		# See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version		# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information		# See top-level LICENSE file for more information

from __future__ import annotations		from __future__ import annotations

from datetime import datetime, timezone		from datetime import datetime, timezone
import os		import os
from types import TracebackType		from types import TracebackType
from typing import Any, Dict, Generator, Iterable, List, Optional, Set, Type, Union		from typing import Any, Dict, Generator, Iterable, List, Optional, Set, Type, Union

from bson import ObjectId		from bson import ObjectId
import mongomock		import mongomock
import pymongo		import pymongo
		from pymongo import InsertOne, UpdateOne
		import sys

from swh.core.statsd import statsd		from swh.core.statsd import statsd
from swh.model.model import Sha1Git		from swh.model.model import Sha1Git

from ..interface import (		from ..interface import (
EntityType,		EntityType,
ProvenanceResult,		ProvenanceResult,
ProvenanceStorageInterface,		ProvenanceStorageInterface,
RelationData,		RelationData,
RelationType,		RelationType,
RevisionData,		RevisionData,
)		)
		from .entity import Entity

STORAGE_DURATION_METRIC = "swh_provenance_storage_mongodb_duration_seconds"		STORAGE_DURATION_METRIC = "swh_provenance_storage_mongodb_duration_seconds"


class ProvenanceStorageMongoDb:		class ProvenanceStorageMongoDb:
def __init__(self, engine: str, **kwargs):		def __init__(self, engine: str, **kwargs):
self.engine = engine		self.engine = engine
self.dbname = kwargs.pop("dbname")		self.dbname = kwargs.pop("dbname")
Show All 10 Lines	def __exit__(
exc_tb: Optional[TracebackType],		exc_tb: Optional[TracebackType],
) -> None:		) -> None:
self.close()		self.close()

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "close"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "close"})
def close(self) -> None:		def close(self) -> None:
self.db.client.close()		self.db.client.close()

		def _generate_date_upserts(self, sha1, date, inserts={}, upsert=True):
		ts = datetime.timestamp(date) if date is not None else None
		# update only those with date either as None or later than the given one
		return UpdateOne(
		{"sha1": sha1, "$or": [{"ts": {"$gte": ts}}, {"ts": None}]},
		{
		"$set": {"ts": ts},
		"$setOnInsert": inserts,
		},
		upsert=upsert,
		)

		def _generate_date_upserts_alternate(self, sha1, date, inserts={}, upsert=True):
		# FIXME, Compare the performance and decide which one to keep
		ts = datetime.timestamp(date) if date is not None else None
		# update only those with date either as None or later than the given one
		return UpdateOne(
		{"sha1": sha1},
		[
		{
		"$set": {
		"ts": {
		"$cond": {
		"if": {"$gt": ["$ts", ts]},
		"then": ts,
		"else": None,
		}
		}
		},
		},
		],
		upsert=upsert,
		)

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_add"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_add"})
def content_add(		def content_add(
self, cnts: Union[Iterable[Sha1Git], Dict[Sha1Git, Optional[datetime]]]		self, cnts: Union[Iterable[Sha1Git], Dict[Sha1Git, Optional[datetime]]]
) -> bool:		) -> bool:
data = cnts if isinstance(cnts, dict) else dict.fromkeys(cnts)		default_inserts = {"revision": {}, "directory": {}}
existing = {		if isinstance(cnts, dict):
x["sha1"]: x		writes = [
for x in self.db.content.find(		self._generate_date_upserts(sha1, date, default_inserts)
{"sha1": {"$in": list(data)}}, {"sha1": 1, "ts": 1, "_id": 1}		for sha1, date in cnts.items()
)		]
}		else:
for sha1, date in data.items():		writes = [
ts = datetime.timestamp(date) if date is not None else None		self._generate_date_upserts(sha1, None, default_inserts)
if sha1 in existing:		for sha1 in cnts
cnt = existing[sha1]		]
if ts is not None and (cnt["ts"] is None or ts < cnt["ts"]):		statsd.gauge("provenance.content-list.size", sys.getsizeof(writes))
self.db.content.update_one(		Entity.factory(self.db, "content").bulk_write(writes)
{"_id": cnt["_id"]}, {"$set": {"ts": ts}}		return True
)
else:		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "directory_add"})
self.db.content.insert_one(		def directory_add(
		self, dirs: Union[Iterable[Sha1Git], Dict[Sha1Git, Optional[datetime]]]
		) -> bool:
		default_inserts = {"revision": {}}
		if isinstance(dirs, dict):
		writes = [
		self._generate_date_upserts(sha1, date, default_inserts)
		for sha1, date in dirs.items()
		]
		else:
		writes = [
		self._generate_date_upserts(sha1, None, default_inserts)
		for sha1 in dirs
		]
		statsd.gauge("provenance.directory-list.size", sys.getsizeof(writes))
		Entity.factory(self.db, "directory").bulk_write(writes)
		return True

		def _get_oldest_revision_from_content(self, content):
		# FIXME, returing with the assumption that content has all the
		# revisons in the array
		# Change to seperate collection content_in_revision if gets too big
		return Entity.factory(self.db, "revision").find_one(
{		{
"sha1": sha1,		"_id": {"$in": [ObjectId(obj_id) for obj_id in content["revision"]]},
"ts": ts,		"ts": content["ts"],
"revision": {},
"directory": {},
}		}
)		)
return True
		def _get_preferred_origin(self, revision):
		if revision.get("preferred"):
		return Entity.factory(self.db, "origin").find_one(
		{"sha1": revision["preferred"]}
		)
		return None

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_find_first"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_find_first"})
def content_find_first(self, id: Sha1Git) -> Optional[ProvenanceResult]:		def content_find_first(self, id: Sha1Git) -> Optional[ProvenanceResult]:
		content = Entity.factory(self.db, "content").find_one({"sha1": id})
		if not content:
		return None

		oldest_revision = self._get_oldest_revision_from_content(content)
		origin = self._get_preferred_origin(oldest_revision)

		return ProvenanceResult(
		content=id,
		revision=oldest_revision["sha1"],
		date=datetime.fromtimestamp(oldest_revision["ts"], timezone.utc),
		origin=origin["url"] if origin else None,
		path=sorted(content["revision"][str(oldest_revision["_id"])])[0]
		# FIXME, this is likely to be wrong, tests pass
		)

		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_find_first"})
		def content_find_first_old(self, id: Sha1Git) -> Optional[ProvenanceResult]:
# get all the revisions		# get all the revisions
# iterate and find the earliest		# iterate and find the earliest
content = self.db.content.find_one({"sha1": id})		content = self.db.content.find_one({"sha1": id})
if not content:		if not content:
return None		return None

occurs = []		occurs = []
for revision in self.db.revision.find(		for revision in self.db.revision.find(
{"_id": {"$in": [ObjectId(obj_id) for obj_id in content["revision"]]}}		{"_id": {"$in": [ObjectId(obj_id) for obj_id in content["revision"]]}}
):		):
if revision["preferred"] is not None:		if revision["preferred"] is not None:
origin = self.db.origin.find_one({"sha1": revision["preferred"]})		origin = self.db.origin.find_one({"sha1": revision["preferred"]})
else:		else:
origin = {"url": None}		origin = {"url": None}

for path in content["revision"][str(revision["_id"])]:		for path in content["revision"][str(revision["_id"])]:
occurs.append(		occurs.append(
ProvenanceResult(		ProvenanceResult(
content=id,		content=id,
revision=revision["sha1"],		revision=revision["sha1"],
date=datetime.fromtimestamp(revision["ts"], timezone.utc),		date=datetime.fromtimestamp(revision["ts"], timezone.utc),
origin=origin["url"],		origin=origin["url"],
path=path,		path=path,
)		)
)		)
return sorted(occurs, key=lambda x: (x.date, x.revision, x.origin, x.path))[0]		return sorted(occurs, key=lambda x: (x.date, x.revision, x.origin, x.path))[0]

		# FIXME, refactor this
@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_find_all"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_find_all"})
def content_find_all(		def content_find_all(
self, id: Sha1Git, limit: Optional[int] = None		self, id: Sha1Git, limit: Optional[int] = None
) -> Generator[ProvenanceResult, None, None]:		) -> Generator[ProvenanceResult, None, None]:
content = self.db.content.find_one({"sha1": id})		content = self.db.content.find_one({"sha1": id})
if not content:		if not content:
return None		return None

Show All 12 Lines	) -> Generator[ProvenanceResult, None, None]:
content=id,		content=id,
revision=revision["sha1"],		revision=revision["sha1"],
date=datetime.fromtimestamp(revision["ts"], timezone.utc),		date=datetime.fromtimestamp(revision["ts"], timezone.utc),
origin=origin["url"],		origin=origin["url"],
path=path,		path=path,
)		)
)		)
for directory in self.db.directory.find(		for directory in self.db.directory.find(
{"_id": {"$in": [ObjectId(obj_id) for obj_id in content["directory"]]}}		{
		"_id": {
		"$in": [ObjectId(obj_id) for obj_id in content.get("directory", {})]
		}
		}
):		):
for revision in self.db.revision.find(		for revision in self.db.revision.find(
{"_id": {"$in": [ObjectId(obj_id) for obj_id in directory["revision"]]}}		{"_id": {"$in": [ObjectId(obj_id) for obj_id in directory["revision"]]}}
):		):
if revision["preferred"] is not None:		if revision["preferred"] is not None:
origin = self.db.origin.find_one({"sha1": revision["preferred"]})		origin = self.db.origin.find_one({"sha1": revision["preferred"]})
else:		else:
origin = {"url": None}		origin = {"url": None}
Show All 16 Lines	) -> Generator[ProvenanceResult, None, None]:
path=path,		path=path,
)		)
)		)
yield from sorted(occurs, key=lambda x: (x.date, x.revision, x.origin, x.path))		yield from sorted(occurs, key=lambda x: (x.date, x.revision, x.origin, x.path))

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_get"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_get"})
def content_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, datetime]:		def content_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, datetime]:
return {		return {
x["sha1"]: datetime.fromtimestamp(x["ts"], timezone.utc)		cnt["sha1"]: datetime.fromtimestamp(cnt["ts"], timezone.utc)
for x in self.db.content.find(		for cnt in Entity.factory(self.db, "content").find(
{"sha1": {"$in": list(ids)}, "ts": {"$ne": None}},		{"sha1": {"$in": list(ids)}, "ts": {"$ne": None}},
{"sha1": 1, "ts": 1, "_id": 0},		{"sha1": 1, "ts": 1, "_id": 0},
)		)
}		}

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "directory_add"})
def directory_add(
self, dirs: Union[Iterable[Sha1Git], Dict[Sha1Git, Optional[datetime]]]
) -> bool:
data = dirs if isinstance(dirs, dict) else dict.fromkeys(dirs)
existing = {
x["sha1"]: x
for x in self.db.directory.find(
{"sha1": {"$in": list(data)}}, {"sha1": 1, "ts": 1, "_id": 1}
)
}
for sha1, date in data.items():
ts = datetime.timestamp(date) if date is not None else None
if sha1 in existing:
dir = existing[sha1]
if ts is not None and (dir["ts"] is None or ts < dir["ts"]):
self.db.directory.update_one(
{"_id": dir["_id"]}, {"$set": {"ts": ts}}
)
else:
self.db.directory.insert_one({"sha1": sha1, "ts": ts, "revision": {}})
return True

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "directory_get"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "directory_get"})
def directory_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, datetime]:		def directory_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, datetime]:
return {		return {
x["sha1"]: datetime.fromtimestamp(x["ts"], timezone.utc)		dry["sha1"]: datetime.fromtimestamp(dry["ts"], timezone.utc)
for x in self.db.directory.find(		for dry in Entity.factory(self.db, "directory").find(
{"sha1": {"$in": list(ids)}, "ts": {"$ne": None}},		{"sha1": {"$in": list(ids)}, "ts": {"$ne": None}},
{"sha1": 1, "ts": 1, "_id": 0},		{"sha1": 1, "ts": 1, "_id": 0},
)		)
}		}

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "entity_get_all"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "entity_get_all"})
def entity_get_all(self, entity: EntityType) -> Set[Sha1Git]:		def entity_get_all(self, entity: EntityType) -> Set[Sha1Git]:
		# test only method; not using entity factory or cache
return {		return {
x["sha1"]		x["sha1"]
for x in self.db.get_collection(entity.value).find(		for x in self.db.get_collection(entity.value).find(
{}, {"sha1": 1, "_id": 0}		{}, {"sha1": 1, "_id": 0}
)		)
}		}

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "location_add"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "location_add"})
def location_add(self, paths: Iterable[bytes]) -> bool:		def location_add(self, paths: Iterable[bytes]) -> bool:
# TODO: implement this methods if path are to be stored in a separate collection		# TODO: implement this methods if path are to be stored in a separate collection
return True		return True

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "location_get_all"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "location_get_all"})
def location_get_all(self) -> Set[bytes]:		def location_get_all(self) -> Set[bytes]:
		# test only method; not using entity factory or cache
contents = self.db.content.find({}, {"revision": 1, "_id": 0, "directory": 1})		contents = self.db.content.find({}, {"revision": 1, "_id": 0, "directory": 1})
paths: List[Iterable[bytes]] = []		paths: List[Iterable[bytes]] = []
for content in contents:		for content in contents:
paths.extend(value for _, value in content["revision"].items())		paths.extend(value for _, value in content["revision"].items())
paths.extend(value for _, value in content["directory"].items())		paths.extend(value for _, value in content["directory"].items())

dirs = self.db.directory.find({}, {"revision": 1, "_id": 0})		dirs = self.db.directory.find({}, {"revision": 1, "_id": 0})
for each_dir in dirs:		for each_dir in dirs:
paths.extend(value for _, value in each_dir["revision"].items())		paths.extend(value for _, value in each_dir["revision"].items())
return set(sum(paths, []))		return set(sum(paths, []))

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "open"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "open"})
def open(self) -> None:		def open(self) -> None:
if self.engine == "mongomock":		if self.engine == "mongomock":
self.db = mongomock.MongoClient(**self.conn_args).get_database(self.dbname)		self.db = mongomock.MongoClient(**self.conn_args).get_database(self.dbname)
else:		else:
# assume real MongoDB server by default		# assume real MongoDB server by default
self.db = pymongo.MongoClient(**self.conn_args).get_database(self.dbname)		self.db = pymongo.MongoClient(host="127.0.0.1", port=27017, connect=False).get_database(self.dbname)

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "origin_add"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "origin_add"})
def origin_add(self, orgs: Dict[Sha1Git, str]) -> bool:		def origin_add(self, orgs: Dict[Sha1Git, str]) -> bool:
existing = {		writes = [InsertOne({"sha1": sha1, "url": url}) for (sha1, url) in orgs.items()]
x["sha1"]: x		Entity.factory(self.db, "origin").bulk_write(writes)
for x in self.db.origin.find(
{"sha1": {"$in": list(orgs)}}, {"sha1": 1, "url": 1, "_id": 1}
)
}
for sha1, url in orgs.items():
if sha1 not in existing:
# add new origin
self.db.origin.insert_one({"sha1": sha1, "url": url})
return True		return True

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "origin_get"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "origin_get"})
def origin_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, str]:		def origin_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, str]:
return {		return {
x["sha1"]: x["url"]		ori["sha1"]: ori["url"]
for x in self.db.origin.find(		for ori in Entity.factory(self.db, "origin").find(
{"sha1": {"$in": list(ids)}}, {"sha1": 1, "url": 1, "_id": 0}		{"sha1": {"$in": list(ids)}}, {"sha1": 1, "url": 1, "_id": 0}
)		)
}		}

		def _generate_revision_upserts(self, sha1, date=None, preferred=None, upsert=True):
		ts = datetime.timestamp(date) if date is not None else None
		return UpdateOne(
		{"sha1": sha1, "$or": [{"ts": {"$gte": ts}}, {"ts": None}]},
		{
		"$set": {"ts": ts, "preferred": preferred},
		"$setOnInsert": {"origin": [], "revision": []},
		},
		upsert=upsert,
		)

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "revision_add"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "revision_add"})
def revision_add(		def revision_add(
self, revs: Union[Iterable[Sha1Git], Dict[Sha1Git, RevisionData]]		self, revs: Union[Iterable[Sha1Git], Dict[Sha1Git, RevisionData]]
) -> bool:		) -> bool:
		# if isinstance(revs, dict):
		# writes = [
		# self._generate_revision_upserts(sha1, data.date, data.origin)
		# for sha1, data in revs.items()
		# ]
		# else:
		# writes = [
		# self._generate_revision_upserts(sha1)
		# for sha1 in cnts
		# ]
		# Entity.factory(self.db, "revision").bulk_write(writes)
		# return True

data = (		data = (
revs		revs
if isinstance(revs, dict)		if isinstance(revs, dict)
else dict.fromkeys(revs, RevisionData(date=None, origin=None))		else dict.fromkeys(revs, RevisionData(date=None, origin=None))
)		)
existing = {		existing = {
x["sha1"]: x		x["sha1"]: x
for x in self.db.revision.find(		for x in self.db.revision.find(
{"sha1": {"$in": list(data)}},		{"sha1": {"$in": list(data)}},
{"sha1": 1, "ts": 1, "preferred": 1, "_id": 1},		{"sha1": 1, "ts": 1, "preferred": 1, "_id": 1},
)		)
}		}

		statsd.gauge("provenance.revision-cache.size", sys.getsizeof(existing))

for sha1, info in data.items():		for sha1, info in data.items():
ts = datetime.timestamp(info.date) if info.date is not None else None		ts = datetime.timestamp(info.date) if info.date is not None else None
preferred = info.origin		preferred = info.origin
if sha1 in existing:		if sha1 in existing:
rev = existing[sha1]		rev = existing[sha1]
if ts is None or (rev["ts"] is not None and ts >= rev["ts"]):		if ts is None or (rev["ts"] is not None and ts >= rev["ts"]):
ts = rev["ts"]		ts = rev["ts"]
if preferred is None:		if preferred is None:
Show All 13 Lines	) -> bool:
"ts": ts,		"ts": ts,
}		}
)		)
return True		return True

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "revision_get"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "revision_get"})
def revision_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, RevisionData]:		def revision_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, RevisionData]:
return {		return {
x["sha1"]: RevisionData(		rev["sha1"]: RevisionData(
date=datetime.fromtimestamp(x["ts"], timezone.utc) if x["ts"] else None,		date=datetime.fromtimestamp(rev["ts"], timezone.utc)
origin=x["preferred"],		if rev["ts"]
		else None,
		origin=rev["preferred"],
)		)
for x in self.db.revision.find(		for rev in Entity.factory(self.db, "revision").find(
{		{
"sha1": {"$in": list(ids)},		"sha1": {"$in": list(ids)},
"$or": [{"preferred": {"$ne": None}}, {"ts": {"$ne": None}}],		"$or": [{"preferred": {"$ne": None}}, {"ts": {"$ne": None}}],
},		},
{"sha1": 1, "preferred": 1, "ts": 1, "_id": 0},		{"sha1": 1, "preferred": 1, "ts": 1, "_id": 0},
)		)
}		}

Show All 27 Lines	) -> bool:

src_objs = {		src_objs = {
x["sha1"]: x		x["sha1"]: x
for x in self.db.get_collection(src_relation).find(		for x in self.db.get_collection(src_relation).find(
{"sha1": {"$in": list(denorm.keys())}}		{"sha1": {"$in": list(denorm.keys())}}
)		)
}		}

		statsd.gauge("provenance.relation-src-cache.size", sys.getsizeof(src_objs))
		statsd.gauge("provenance.relation-dst-cache.size", sys.getsizeof(dst_objs))

for sha1, dsts in denorm.items():		for sha1, dsts in denorm.items():
# update		# update
if src_relation != "revision":		if src_relation != "revision":
k = {		k = {
obj_id: list(set(paths + dsts.get(obj_id, [])))		obj_id: list(set(paths + dsts.get(obj_id, [])))
for obj_id, paths in src_objs[sha1][dst_relation].items()		for obj_id, paths in src_objs[sha1].get(dst_relation, {}).items()
}		}
self.db.get_collection(src_relation).update_one(		self.db.get_collection(src_relation).update_one(
{"_id": src_objs[sha1]["_id"]},		{"_id": src_objs[sha1]["_id"]},
{"$set": {dst_relation: dict(dsts, **k)}},		{"$set": {dst_relation: dict(dsts, **k)}},
)		)
else:		else:
self.db.get_collection(src_relation).update_one(		self.db.get_collection(src_relation).update_one(
{"_id": src_objs[sha1]["_id"]},		{"_id": src_objs[sha1]["_id"]},
▲ Show 20 Lines • Show All 53 Lines • ▼ Show 20 Lines	) -> Dict[Sha1Git, Set[RelationData]]:
else:		else:
dst_objs = {		dst_objs = {
x["sha1"]: x["_id"]		x["sha1"]: x["_id"]
for x in self.db.get_collection(dst).find(		for x in self.db.get_collection(dst).find(
{"sha1": {"$in": list(sha1s)}}, {"_id": 1, "sha1": 1}		{"sha1": {"$in": list(sha1s)}}, {"_id": 1, "sha1": 1}
)		)
}		}
src_objs = {		src_objs = {
x["sha1"]: x[dst]		x["sha1"]: x.get(dst, {})
for x in self.db.get_collection(src).find(		for x in self.db.get_collection(src).find(
{}, {"_id": 0, "sha1": 1, dst: 1}		{}, {"_id": 0, "sha1": 1, dst: 1}
)		)
}		}
result: Dict[Sha1Git, Set[RelationData]] = {}		result: Dict[Sha1Git, Set[RelationData]] = {}
if src != "revision":		if src != "revision":
for dst_sha1, dst_obj_id in dst_objs.items():		for dst_sha1, dst_obj_id in dst_objs.items():
for src_sha1, denorm in src_objs.items():		for src_sha1, denorm in src_objs.items():
Show All 13 Lines	) -> Dict[Sha1Git, Set[RelationData]]:
RelationData(dst=dst_sha1, path=None)		RelationData(dst=dst_sha1, path=None)
)		)
return result		return result

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "relation_get_all"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "relation_get_all"})
def relation_get_all(		def relation_get_all(
self, relation: RelationType		self, relation: RelationType
) -> Dict[Sha1Git, Set[RelationData]]:		) -> Dict[Sha1Git, Set[RelationData]]:
		# test only method; not using entity factory or cache
src, *_, dst = relation.value.split("_")		src, *_, dst = relation.value.split("_")
empty: Union[Dict[str, bytes], List[str]] = {} if src != "revision" else []		empty: Union[Dict[str, bytes], List[str]] = {} if src != "revision" else []
src_objs = {		src_objs = {
x["sha1"]: x[dst]		x["sha1"]: x.get(dst, {})
for x in self.db.get_collection(src).find(		for x in self.db.get_collection(src).find(
{dst: {"$ne": empty}}, {"_id": 0, "sha1": 1, dst: 1}		{dst: {"$ne": empty}}, {"_id": 0, "sha1": 1, dst: 1}
)		)
}		}
dst_ids = list(		dst_ids = list(
{ObjectId(obj_id) for _, value in src_objs.items() for obj_id in value}		{ObjectId(obj_id) for _, value in src_objs.items() for obj_id in value}
)		)
dst_objs = {		dst_objs = {
Show All 22 Lines	) -> Dict[Sha1Git, Set[RelationData]]:
if dst_obj_id == dst_obj_ref		if dst_obj_id == dst_obj_ref
}		}
for src_sha1, denorm in src_objs.items()		for src_sha1, denorm in src_objs.items()
}		}

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "with_path"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "with_path"})
def with_path(self) -> bool:		def with_path(self) -> bool:
return True		return True

		def entity_get(self, entity, filters=None):
		"""
		A method used only by tests
		Returns a list of entities after applying the filter
		"""
		# FIXME, this should be added to the interface to make
		# tests backend agnostic. Adding this in SQL could be tricky
		# FIXME, change to entity factory
		return list(Entity.factory(self.db, entity).find(filters))

		def entity_add(self, entity, data):
		"""
		A method used only by tests
		"""
		# FIXME, this should be added to the interface to make
		# tests backend agnostic. Adding this in SQL could be tricky
		# FIXME, change to entity factory
		return Entity.factory(self.db, entity).insert(data)

		def create_indexes(self, entity, fields, unique=False):
		Entity.factory(self.db, entity).create_index(fields, unique)

View Options

swh/graphql/resolvers/scalars.py

# Copyright (C) 2021 The Software Heritage developers		# Copyright (C) 2021 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution		# See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version		# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information		# See top-level LICENSE file for more information

from __future__ import annotations		from __future__ import annotations

from datetime import datetime, timezone		from datetime import datetime, timezone
import os		import os
from types import TracebackType		from types import TracebackType
from typing import Any, Dict, Generator, Iterable, List, Optional, Set, Type, Union		from typing import Any, Dict, Generator, Iterable, List, Optional, Set, Type, Union

from bson import ObjectId		from bson import ObjectId
import mongomock		import mongomock
import pymongo		import pymongo
		from pymongo import InsertOne, UpdateOne
		import sys

from swh.core.statsd import statsd		from swh.core.statsd import statsd
from swh.model.model import Sha1Git		from swh.model.model import Sha1Git

from ..interface import (		from ..interface import (
EntityType,		EntityType,
ProvenanceResult,		ProvenanceResult,
ProvenanceStorageInterface,		ProvenanceStorageInterface,
RelationData,		RelationData,
RelationType,		RelationType,
RevisionData,		RevisionData,
)		)
		from .entity import Entity

STORAGE_DURATION_METRIC = "swh_provenance_storage_mongodb_duration_seconds"		STORAGE_DURATION_METRIC = "swh_provenance_storage_mongodb_duration_seconds"


class ProvenanceStorageMongoDb:		class ProvenanceStorageMongoDb:
def __init__(self, engine: str, **kwargs):		def __init__(self, engine: str, **kwargs):
self.engine = engine		self.engine = engine
self.dbname = kwargs.pop("dbname")		self.dbname = kwargs.pop("dbname")
Show All 10 Lines	def __exit__(
exc_tb: Optional[TracebackType],		exc_tb: Optional[TracebackType],
) -> None:		) -> None:
self.close()		self.close()

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "close"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "close"})
def close(self) -> None:		def close(self) -> None:
self.db.client.close()		self.db.client.close()

		def _generate_date_upserts(self, sha1, date, inserts={}, upsert=True):
		ts = datetime.timestamp(date) if date is not None else None
		# update only those with date either as None or later than the given one
		return UpdateOne(
		{"sha1": sha1, "$or": [{"ts": {"$gte": ts}}, {"ts": None}]},
		{
		"$set": {"ts": ts},
		"$setOnInsert": inserts,
		},
		upsert=upsert,
		)

		def _generate_date_upserts_alternate(self, sha1, date, inserts={}, upsert=True):
		# FIXME, Compare the performance and decide which one to keep
		ts = datetime.timestamp(date) if date is not None else None
		# update only those with date either as None or later than the given one
		return UpdateOne(
		{"sha1": sha1},
		[
		{
		"$set": {
		"ts": {
		"$cond": {
		"if": {"$gt": ["$ts", ts]},
		"then": ts,
		"else": None,
		}
		}
		},
		},
		],
		upsert=upsert,
		)

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_add"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_add"})
def content_add(		def content_add(
self, cnts: Union[Iterable[Sha1Git], Dict[Sha1Git, Optional[datetime]]]		self, cnts: Union[Iterable[Sha1Git], Dict[Sha1Git, Optional[datetime]]]
) -> bool:		) -> bool:
data = cnts if isinstance(cnts, dict) else dict.fromkeys(cnts)		default_inserts = {"revision": {}, "directory": {}}
existing = {		if isinstance(cnts, dict):
x["sha1"]: x		writes = [
for x in self.db.content.find(		self._generate_date_upserts(sha1, date, default_inserts)
{"sha1": {"$in": list(data)}}, {"sha1": 1, "ts": 1, "_id": 1}		for sha1, date in cnts.items()
)		]
}		else:
for sha1, date in data.items():		writes = [
ts = datetime.timestamp(date) if date is not None else None		self._generate_date_upserts(sha1, None, default_inserts)
if sha1 in existing:		for sha1 in cnts
cnt = existing[sha1]		]
if ts is not None and (cnt["ts"] is None or ts < cnt["ts"]):		statsd.gauge("provenance.content-list.size", sys.getsizeof(writes))
self.db.content.update_one(		Entity.factory(self.db, "content").bulk_write(writes)
{"_id": cnt["_id"]}, {"$set": {"ts": ts}}		return True
)
else:		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "directory_add"})
self.db.content.insert_one(		def directory_add(
		self, dirs: Union[Iterable[Sha1Git], Dict[Sha1Git, Optional[datetime]]]
		) -> bool:
		default_inserts = {"revision": {}}
		if isinstance(dirs, dict):
		writes = [
		self._generate_date_upserts(sha1, date, default_inserts)
		for sha1, date in dirs.items()
		]
		else:
		writes = [
		self._generate_date_upserts(sha1, None, default_inserts)
		for sha1 in dirs
		]
		statsd.gauge("provenance.directory-list.size", sys.getsizeof(writes))
		Entity.factory(self.db, "directory").bulk_write(writes)
		return True

		def _get_oldest_revision_from_content(self, content):
		# FIXME, returing with the assumption that content has all the
		# revisons in the array
		# Change to seperate collection content_in_revision if gets too big
		return Entity.factory(self.db, "revision").find_one(
{		{
"sha1": sha1,		"_id": {"$in": [ObjectId(obj_id) for obj_id in content["revision"]]},
"ts": ts,		"ts": content["ts"],
"revision": {},
"directory": {},
}		}
)		)
return True
		def _get_preferred_origin(self, revision):
		if revision.get("preferred"):
		return Entity.factory(self.db, "origin").find_one(
		{"sha1": revision["preferred"]}
		)
		return None

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_find_first"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_find_first"})
def content_find_first(self, id: Sha1Git) -> Optional[ProvenanceResult]:		def content_find_first(self, id: Sha1Git) -> Optional[ProvenanceResult]:
		content = Entity.factory(self.db, "content").find_one({"sha1": id})
		if not content:
		return None

		oldest_revision = self._get_oldest_revision_from_content(content)
		origin = self._get_preferred_origin(oldest_revision)

		return ProvenanceResult(
		content=id,
		revision=oldest_revision["sha1"],
		date=datetime.fromtimestamp(oldest_revision["ts"], timezone.utc),
		origin=origin["url"] if origin else None,
		path=sorted(content["revision"][str(oldest_revision["_id"])])[0]
		# FIXME, this is likely to be wrong, tests pass
		)

		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_find_first"})
		def content_find_first_old(self, id: Sha1Git) -> Optional[ProvenanceResult]:
# get all the revisions		# get all the revisions
# iterate and find the earliest		# iterate and find the earliest
content = self.db.content.find_one({"sha1": id})		content = self.db.content.find_one({"sha1": id})
if not content:		if not content:
return None		return None

occurs = []		occurs = []
for revision in self.db.revision.find(		for revision in self.db.revision.find(
{"_id": {"$in": [ObjectId(obj_id) for obj_id in content["revision"]]}}		{"_id": {"$in": [ObjectId(obj_id) for obj_id in content["revision"]]}}
):		):
if revision["preferred"] is not None:		if revision["preferred"] is not None:
origin = self.db.origin.find_one({"sha1": revision["preferred"]})		origin = self.db.origin.find_one({"sha1": revision["preferred"]})
else:		else:
origin = {"url": None}		origin = {"url": None}

for path in content["revision"][str(revision["_id"])]:		for path in content["revision"][str(revision["_id"])]:
occurs.append(		occurs.append(
ProvenanceResult(		ProvenanceResult(
content=id,		content=id,
revision=revision["sha1"],		revision=revision["sha1"],
date=datetime.fromtimestamp(revision["ts"], timezone.utc),		date=datetime.fromtimestamp(revision["ts"], timezone.utc),
origin=origin["url"],		origin=origin["url"],
path=path,		path=path,
)		)
)		)
return sorted(occurs, key=lambda x: (x.date, x.revision, x.origin, x.path))[0]		return sorted(occurs, key=lambda x: (x.date, x.revision, x.origin, x.path))[0]

		# FIXME, refactor this
@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_find_all"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_find_all"})
def content_find_all(		def content_find_all(
self, id: Sha1Git, limit: Optional[int] = None		self, id: Sha1Git, limit: Optional[int] = None
) -> Generator[ProvenanceResult, None, None]:		) -> Generator[ProvenanceResult, None, None]:
content = self.db.content.find_one({"sha1": id})		content = self.db.content.find_one({"sha1": id})
if not content:		if not content:
return None		return None

Show All 12 Lines	) -> Generator[ProvenanceResult, None, None]:
content=id,		content=id,
revision=revision["sha1"],		revision=revision["sha1"],
date=datetime.fromtimestamp(revision["ts"], timezone.utc),		date=datetime.fromtimestamp(revision["ts"], timezone.utc),
origin=origin["url"],		origin=origin["url"],
path=path,		path=path,
)		)
)		)
for directory in self.db.directory.find(		for directory in self.db.directory.find(
{"_id": {"$in": [ObjectId(obj_id) for obj_id in content["directory"]]}}		{
		"_id": {
		"$in": [ObjectId(obj_id) for obj_id in content.get("directory", {})]
		}
		}
):		):
for revision in self.db.revision.find(		for revision in self.db.revision.find(
{"_id": {"$in": [ObjectId(obj_id) for obj_id in directory["revision"]]}}		{"_id": {"$in": [ObjectId(obj_id) for obj_id in directory["revision"]]}}
):		):
if revision["preferred"] is not None:		if revision["preferred"] is not None:
origin = self.db.origin.find_one({"sha1": revision["preferred"]})		origin = self.db.origin.find_one({"sha1": revision["preferred"]})
else:		else:
origin = {"url": None}		origin = {"url": None}
Show All 16 Lines	) -> Generator[ProvenanceResult, None, None]:
path=path,		path=path,
)		)
)		)
yield from sorted(occurs, key=lambda x: (x.date, x.revision, x.origin, x.path))		yield from sorted(occurs, key=lambda x: (x.date, x.revision, x.origin, x.path))

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_get"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_get"})
def content_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, datetime]:		def content_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, datetime]:
return {		return {
x["sha1"]: datetime.fromtimestamp(x["ts"], timezone.utc)		cnt["sha1"]: datetime.fromtimestamp(cnt["ts"], timezone.utc)
for x in self.db.content.find(		for cnt in Entity.factory(self.db, "content").find(
{"sha1": {"$in": list(ids)}, "ts": {"$ne": None}},		{"sha1": {"$in": list(ids)}, "ts": {"$ne": None}},
{"sha1": 1, "ts": 1, "_id": 0},		{"sha1": 1, "ts": 1, "_id": 0},
)		)
}		}

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "directory_add"})
def directory_add(
self, dirs: Union[Iterable[Sha1Git], Dict[Sha1Git, Optional[datetime]]]
) -> bool:
data = dirs if isinstance(dirs, dict) else dict.fromkeys(dirs)
existing = {
x["sha1"]: x
for x in self.db.directory.find(
{"sha1": {"$in": list(data)}}, {"sha1": 1, "ts": 1, "_id": 1}
)
}
for sha1, date in data.items():
ts = datetime.timestamp(date) if date is not None else None
if sha1 in existing:
dir = existing[sha1]
if ts is not None and (dir["ts"] is None or ts < dir["ts"]):
self.db.directory.update_one(
{"_id": dir["_id"]}, {"$set": {"ts": ts}}
)
else:
self.db.directory.insert_one({"sha1": sha1, "ts": ts, "revision": {}})
return True

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "directory_get"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "directory_get"})
def directory_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, datetime]:		def directory_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, datetime]:
return {		return {
x["sha1"]: datetime.fromtimestamp(x["ts"], timezone.utc)		dry["sha1"]: datetime.fromtimestamp(dry["ts"], timezone.utc)
for x in self.db.directory.find(		for dry in Entity.factory(self.db, "directory").find(
{"sha1": {"$in": list(ids)}, "ts": {"$ne": None}},		{"sha1": {"$in": list(ids)}, "ts": {"$ne": None}},
{"sha1": 1, "ts": 1, "_id": 0},		{"sha1": 1, "ts": 1, "_id": 0},
)		)
}		}

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "entity_get_all"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "entity_get_all"})
def entity_get_all(self, entity: EntityType) -> Set[Sha1Git]:		def entity_get_all(self, entity: EntityType) -> Set[Sha1Git]:
		# test only method; not using entity factory or cache
return {		return {
x["sha1"]		x["sha1"]
for x in self.db.get_collection(entity.value).find(		for x in self.db.get_collection(entity.value).find(
{}, {"sha1": 1, "_id": 0}		{}, {"sha1": 1, "_id": 0}
)		)
}		}

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "location_add"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "location_add"})
def location_add(self, paths: Iterable[bytes]) -> bool:		def location_add(self, paths: Iterable[bytes]) -> bool:
# TODO: implement this methods if path are to be stored in a separate collection		# TODO: implement this methods if path are to be stored in a separate collection
return True		return True

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "location_get_all"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "location_get_all"})
def location_get_all(self) -> Set[bytes]:		def location_get_all(self) -> Set[bytes]:
		# test only method; not using entity factory or cache
contents = self.db.content.find({}, {"revision": 1, "_id": 0, "directory": 1})		contents = self.db.content.find({}, {"revision": 1, "_id": 0, "directory": 1})
paths: List[Iterable[bytes]] = []		paths: List[Iterable[bytes]] = []
for content in contents:		for content in contents:
paths.extend(value for _, value in content["revision"].items())		paths.extend(value for _, value in content["revision"].items())
paths.extend(value for _, value in content["directory"].items())		paths.extend(value for _, value in content["directory"].items())

dirs = self.db.directory.find({}, {"revision": 1, "_id": 0})		dirs = self.db.directory.find({}, {"revision": 1, "_id": 0})
for each_dir in dirs:		for each_dir in dirs:
paths.extend(value for _, value in each_dir["revision"].items())		paths.extend(value for _, value in each_dir["revision"].items())
return set(sum(paths, []))		return set(sum(paths, []))

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "open"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "open"})
def open(self) -> None:		def open(self) -> None:
if self.engine == "mongomock":		if self.engine == "mongomock":
self.db = mongomock.MongoClient(**self.conn_args).get_database(self.dbname)		self.db = mongomock.MongoClient(**self.conn_args).get_database(self.dbname)
else:		else:
# assume real MongoDB server by default		# assume real MongoDB server by default
self.db = pymongo.MongoClient(**self.conn_args).get_database(self.dbname)		self.db = pymongo.MongoClient(host="127.0.0.1", port=27017, connect=False).get_database(self.dbname)

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "origin_add"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "origin_add"})
def origin_add(self, orgs: Dict[Sha1Git, str]) -> bool:		def origin_add(self, orgs: Dict[Sha1Git, str]) -> bool:
existing = {		writes = [InsertOne({"sha1": sha1, "url": url}) for (sha1, url) in orgs.items()]
x["sha1"]: x		Entity.factory(self.db, "origin").bulk_write(writes)
for x in self.db.origin.find(
{"sha1": {"$in": list(orgs)}}, {"sha1": 1, "url": 1, "_id": 1}
)
}
for sha1, url in orgs.items():
if sha1 not in existing:
# add new origin
self.db.origin.insert_one({"sha1": sha1, "url": url})
return True		return True

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "origin_get"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "origin_get"})
def origin_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, str]:		def origin_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, str]:
return {		return {
x["sha1"]: x["url"]		ori["sha1"]: ori["url"]
for x in self.db.origin.find(		for ori in Entity.factory(self.db, "origin").find(
{"sha1": {"$in": list(ids)}}, {"sha1": 1, "url": 1, "_id": 0}		{"sha1": {"$in": list(ids)}}, {"sha1": 1, "url": 1, "_id": 0}
)		)
}		}

		def _generate_revision_upserts(self, sha1, date=None, preferred=None, upsert=True):
		ts = datetime.timestamp(date) if date is not None else None
		return UpdateOne(
		{"sha1": sha1, "$or": [{"ts": {"$gte": ts}}, {"ts": None}]},
		{
		"$set": {"ts": ts, "preferred": preferred},
		"$setOnInsert": {"origin": [], "revision": []},
		},
		upsert=upsert,
		)

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "revision_add"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "revision_add"})
def revision_add(		def revision_add(
self, revs: Union[Iterable[Sha1Git], Dict[Sha1Git, RevisionData]]		self, revs: Union[Iterable[Sha1Git], Dict[Sha1Git, RevisionData]]
) -> bool:		) -> bool:
		# if isinstance(revs, dict):
		# writes = [
		# self._generate_revision_upserts(sha1, data.date, data.origin)
		# for sha1, data in revs.items()
		# ]
		# else:
		# writes = [
		# self._generate_revision_upserts(sha1)
		# for sha1 in cnts
		# ]
		# Entity.factory(self.db, "revision").bulk_write(writes)
		# return True

data = (		data = (
revs		revs
if isinstance(revs, dict)		if isinstance(revs, dict)
else dict.fromkeys(revs, RevisionData(date=None, origin=None))		else dict.fromkeys(revs, RevisionData(date=None, origin=None))
)		)
existing = {		existing = {
x["sha1"]: x		x["sha1"]: x
for x in self.db.revision.find(		for x in self.db.revision.find(
{"sha1": {"$in": list(data)}},		{"sha1": {"$in": list(data)}},
{"sha1": 1, "ts": 1, "preferred": 1, "_id": 1},		{"sha1": 1, "ts": 1, "preferred": 1, "_id": 1},
)		)
}		}

		statsd.gauge("provenance.revision-cache.size", sys.getsizeof(existing))

for sha1, info in data.items():		for sha1, info in data.items():
ts = datetime.timestamp(info.date) if info.date is not None else None		ts = datetime.timestamp(info.date) if info.date is not None else None
preferred = info.origin		preferred = info.origin
if sha1 in existing:		if sha1 in existing:
rev = existing[sha1]		rev = existing[sha1]
if ts is None or (rev["ts"] is not None and ts >= rev["ts"]):		if ts is None or (rev["ts"] is not None and ts >= rev["ts"]):
ts = rev["ts"]		ts = rev["ts"]
if preferred is None:		if preferred is None:
Show All 13 Lines	) -> bool:
"ts": ts,		"ts": ts,
}		}
)		)
return True		return True

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "revision_get"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "revision_get"})
def revision_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, RevisionData]:		def revision_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, RevisionData]:
return {		return {
x["sha1"]: RevisionData(		rev["sha1"]: RevisionData(
date=datetime.fromtimestamp(x["ts"], timezone.utc) if x["ts"] else None,		date=datetime.fromtimestamp(rev["ts"], timezone.utc)
origin=x["preferred"],		if rev["ts"]
		else None,
		origin=rev["preferred"],
)		)
for x in self.db.revision.find(		for rev in Entity.factory(self.db, "revision").find(
{		{
"sha1": {"$in": list(ids)},		"sha1": {"$in": list(ids)},
"$or": [{"preferred": {"$ne": None}}, {"ts": {"$ne": None}}],		"$or": [{"preferred": {"$ne": None}}, {"ts": {"$ne": None}}],
},		},
{"sha1": 1, "preferred": 1, "ts": 1, "_id": 0},		{"sha1": 1, "preferred": 1, "ts": 1, "_id": 0},
)		)
}		}

Show All 27 Lines	) -> bool:

src_objs = {		src_objs = {
x["sha1"]: x		x["sha1"]: x
for x in self.db.get_collection(src_relation).find(		for x in self.db.get_collection(src_relation).find(
{"sha1": {"$in": list(denorm.keys())}}		{"sha1": {"$in": list(denorm.keys())}}
)		)
}		}

		statsd.gauge("provenance.relation-src-cache.size", sys.getsizeof(src_objs))
		statsd.gauge("provenance.relation-dst-cache.size", sys.getsizeof(dst_objs))

for sha1, dsts in denorm.items():		for sha1, dsts in denorm.items():
# update		# update
if src_relation != "revision":		if src_relation != "revision":
k = {		k = {
obj_id: list(set(paths + dsts.get(obj_id, [])))		obj_id: list(set(paths + dsts.get(obj_id, [])))
for obj_id, paths in src_objs[sha1][dst_relation].items()		for obj_id, paths in src_objs[sha1].get(dst_relation, {}).items()
}		}
self.db.get_collection(src_relation).update_one(		self.db.get_collection(src_relation).update_one(
{"_id": src_objs[sha1]["_id"]},		{"_id": src_objs[sha1]["_id"]},
{"$set": {dst_relation: dict(dsts, **k)}},		{"$set": {dst_relation: dict(dsts, **k)}},
)		)
else:		else:
self.db.get_collection(src_relation).update_one(		self.db.get_collection(src_relation).update_one(
{"_id": src_objs[sha1]["_id"]},		{"_id": src_objs[sha1]["_id"]},
▲ Show 20 Lines • Show All 53 Lines • ▼ Show 20 Lines	) -> Dict[Sha1Git, Set[RelationData]]:
else:		else:
dst_objs = {		dst_objs = {
x["sha1"]: x["_id"]		x["sha1"]: x["_id"]
for x in self.db.get_collection(dst).find(		for x in self.db.get_collection(dst).find(
{"sha1": {"$in": list(sha1s)}}, {"_id": 1, "sha1": 1}		{"sha1": {"$in": list(sha1s)}}, {"_id": 1, "sha1": 1}
)		)
}		}
src_objs = {		src_objs = {
x["sha1"]: x[dst]		x["sha1"]: x.get(dst, {})
for x in self.db.get_collection(src).find(		for x in self.db.get_collection(src).find(
{}, {"_id": 0, "sha1": 1, dst: 1}		{}, {"_id": 0, "sha1": 1, dst: 1}
)		)
}		}
result: Dict[Sha1Git, Set[RelationData]] = {}		result: Dict[Sha1Git, Set[RelationData]] = {}
if src != "revision":		if src != "revision":
for dst_sha1, dst_obj_id in dst_objs.items():		for dst_sha1, dst_obj_id in dst_objs.items():
for src_sha1, denorm in src_objs.items():		for src_sha1, denorm in src_objs.items():
Show All 13 Lines	) -> Dict[Sha1Git, Set[RelationData]]:
RelationData(dst=dst_sha1, path=None)		RelationData(dst=dst_sha1, path=None)
)		)
return result		return result

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "relation_get_all"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "relation_get_all"})
def relation_get_all(		def relation_get_all(
self, relation: RelationType		self, relation: RelationType
) -> Dict[Sha1Git, Set[RelationData]]:		) -> Dict[Sha1Git, Set[RelationData]]:
		# test only method; not using entity factory or cache
src, *_, dst = relation.value.split("_")		src, *_, dst = relation.value.split("_")
empty: Union[Dict[str, bytes], List[str]] = {} if src != "revision" else []		empty: Union[Dict[str, bytes], List[str]] = {} if src != "revision" else []
src_objs = {		src_objs = {
x["sha1"]: x[dst]		x["sha1"]: x.get(dst, {})
for x in self.db.get_collection(src).find(		for x in self.db.get_collection(src).find(
{dst: {"$ne": empty}}, {"_id": 0, "sha1": 1, dst: 1}		{dst: {"$ne": empty}}, {"_id": 0, "sha1": 1, dst: 1}
)		)
}		}
dst_ids = list(		dst_ids = list(
{ObjectId(obj_id) for _, value in src_objs.items() for obj_id in value}		{ObjectId(obj_id) for _, value in src_objs.items() for obj_id in value}
)		)
dst_objs = {		dst_objs = {
Show All 22 Lines	) -> Dict[Sha1Git, Set[RelationData]]:
if dst_obj_id == dst_obj_ref		if dst_obj_id == dst_obj_ref
}		}
for src_sha1, denorm in src_objs.items()		for src_sha1, denorm in src_objs.items()
}		}

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "with_path"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "with_path"})
def with_path(self) -> bool:		def with_path(self) -> bool:
return True		return True

		def entity_get(self, entity, filters=None):
		"""
		A method used only by tests
		Returns a list of entities after applying the filter
		"""
		# FIXME, this should be added to the interface to make
		# tests backend agnostic. Adding this in SQL could be tricky
		# FIXME, change to entity factory
		return list(Entity.factory(self.db, entity).find(filters))

		def entity_add(self, entity, data):
		"""
		A method used only by tests
		"""
		# FIXME, this should be added to the interface to make
		# tests backend agnostic. Adding this in SQL could be tricky
		# FIXME, change to entity factory
		return Entity.factory(self.db, entity).insert(data)

		def create_indexes(self, entity, fields, unique=False):
		Entity.factory(self.db, entity).create_index(fields, unique)

View Options

swh/graphql/schema/schema.graphql

# Copyright (C) 2021 The Software Heritage developers		# Copyright (C) 2021 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution		# See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version		# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information		# See top-level LICENSE file for more information

from __future__ import annotations		from __future__ import annotations

from datetime import datetime, timezone		from datetime import datetime, timezone
import os		import os
from types import TracebackType		from types import TracebackType
from typing import Any, Dict, Generator, Iterable, List, Optional, Set, Type, Union		from typing import Any, Dict, Generator, Iterable, List, Optional, Set, Type, Union

from bson import ObjectId		from bson import ObjectId
import mongomock		import mongomock
import pymongo		import pymongo
		from pymongo import InsertOne, UpdateOne
		import sys

from swh.core.statsd import statsd		from swh.core.statsd import statsd
from swh.model.model import Sha1Git		from swh.model.model import Sha1Git

from ..interface import (		from ..interface import (
EntityType,		EntityType,
ProvenanceResult,		ProvenanceResult,
ProvenanceStorageInterface,		ProvenanceStorageInterface,
RelationData,		RelationData,
RelationType,		RelationType,
RevisionData,		RevisionData,
)		)
		from .entity import Entity

STORAGE_DURATION_METRIC = "swh_provenance_storage_mongodb_duration_seconds"		STORAGE_DURATION_METRIC = "swh_provenance_storage_mongodb_duration_seconds"


class ProvenanceStorageMongoDb:		class ProvenanceStorageMongoDb:
def __init__(self, engine: str, **kwargs):		def __init__(self, engine: str, **kwargs):
self.engine = engine		self.engine = engine
self.dbname = kwargs.pop("dbname")		self.dbname = kwargs.pop("dbname")
Show All 10 Lines	def __exit__(
exc_tb: Optional[TracebackType],		exc_tb: Optional[TracebackType],
) -> None:		) -> None:
self.close()		self.close()

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "close"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "close"})
def close(self) -> None:		def close(self) -> None:
self.db.client.close()		self.db.client.close()

		def _generate_date_upserts(self, sha1, date, inserts={}, upsert=True):
		ts = datetime.timestamp(date) if date is not None else None
		# update only those with date either as None or later than the given one
		return UpdateOne(
		{"sha1": sha1, "$or": [{"ts": {"$gte": ts}}, {"ts": None}]},
		{
		"$set": {"ts": ts},
		"$setOnInsert": inserts,
		},
		upsert=upsert,
		)

		def _generate_date_upserts_alternate(self, sha1, date, inserts={}, upsert=True):
		# FIXME, Compare the performance and decide which one to keep
		ts = datetime.timestamp(date) if date is not None else None
		# update only those with date either as None or later than the given one
		return UpdateOne(
		{"sha1": sha1},
		[
		{
		"$set": {
		"ts": {
		"$cond": {
		"if": {"$gt": ["$ts", ts]},
		"then": ts,
		"else": None,
		}
		}
		},
		},
		],
		upsert=upsert,
		)

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_add"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_add"})
def content_add(		def content_add(
self, cnts: Union[Iterable[Sha1Git], Dict[Sha1Git, Optional[datetime]]]		self, cnts: Union[Iterable[Sha1Git], Dict[Sha1Git, Optional[datetime]]]
) -> bool:		) -> bool:
data = cnts if isinstance(cnts, dict) else dict.fromkeys(cnts)		default_inserts = {"revision": {}, "directory": {}}
existing = {		if isinstance(cnts, dict):
x["sha1"]: x		writes = [
for x in self.db.content.find(		self._generate_date_upserts(sha1, date, default_inserts)
{"sha1": {"$in": list(data)}}, {"sha1": 1, "ts": 1, "_id": 1}		for sha1, date in cnts.items()
)		]
}		else:
for sha1, date in data.items():		writes = [
ts = datetime.timestamp(date) if date is not None else None		self._generate_date_upserts(sha1, None, default_inserts)
if sha1 in existing:		for sha1 in cnts
cnt = existing[sha1]		]
if ts is not None and (cnt["ts"] is None or ts < cnt["ts"]):		statsd.gauge("provenance.content-list.size", sys.getsizeof(writes))
self.db.content.update_one(		Entity.factory(self.db, "content").bulk_write(writes)
{"_id": cnt["_id"]}, {"$set": {"ts": ts}}		return True
)
else:		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "directory_add"})
self.db.content.insert_one(		def directory_add(
		self, dirs: Union[Iterable[Sha1Git], Dict[Sha1Git, Optional[datetime]]]
		) -> bool:
		default_inserts = {"revision": {}}
		if isinstance(dirs, dict):
		writes = [
		self._generate_date_upserts(sha1, date, default_inserts)
		for sha1, date in dirs.items()
		]
		else:
		writes = [
		self._generate_date_upserts(sha1, None, default_inserts)
		for sha1 in dirs
		]
		statsd.gauge("provenance.directory-list.size", sys.getsizeof(writes))
		Entity.factory(self.db, "directory").bulk_write(writes)
		return True

		def _get_oldest_revision_from_content(self, content):
		# FIXME, returing with the assumption that content has all the
		# revisons in the array
		# Change to seperate collection content_in_revision if gets too big
		return Entity.factory(self.db, "revision").find_one(
{		{
"sha1": sha1,		"_id": {"$in": [ObjectId(obj_id) for obj_id in content["revision"]]},
"ts": ts,		"ts": content["ts"],
"revision": {},
"directory": {},
}		}
)		)
return True
		def _get_preferred_origin(self, revision):
		if revision.get("preferred"):
		return Entity.factory(self.db, "origin").find_one(
		{"sha1": revision["preferred"]}
		)
		return None

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_find_first"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_find_first"})
def content_find_first(self, id: Sha1Git) -> Optional[ProvenanceResult]:		def content_find_first(self, id: Sha1Git) -> Optional[ProvenanceResult]:
		content = Entity.factory(self.db, "content").find_one({"sha1": id})
		if not content:
		return None

		oldest_revision = self._get_oldest_revision_from_content(content)
		origin = self._get_preferred_origin(oldest_revision)

		return ProvenanceResult(
		content=id,
		revision=oldest_revision["sha1"],
		date=datetime.fromtimestamp(oldest_revision["ts"], timezone.utc),
		origin=origin["url"] if origin else None,
		path=sorted(content["revision"][str(oldest_revision["_id"])])[0]
		# FIXME, this is likely to be wrong, tests pass
		)

		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_find_first"})
		def content_find_first_old(self, id: Sha1Git) -> Optional[ProvenanceResult]:
# get all the revisions		# get all the revisions
# iterate and find the earliest		# iterate and find the earliest
content = self.db.content.find_one({"sha1": id})		content = self.db.content.find_one({"sha1": id})
if not content:		if not content:
return None		return None

occurs = []		occurs = []
for revision in self.db.revision.find(		for revision in self.db.revision.find(
{"_id": {"$in": [ObjectId(obj_id) for obj_id in content["revision"]]}}		{"_id": {"$in": [ObjectId(obj_id) for obj_id in content["revision"]]}}
):		):
if revision["preferred"] is not None:		if revision["preferred"] is not None:
origin = self.db.origin.find_one({"sha1": revision["preferred"]})		origin = self.db.origin.find_one({"sha1": revision["preferred"]})
else:		else:
origin = {"url": None}		origin = {"url": None}

for path in content["revision"][str(revision["_id"])]:		for path in content["revision"][str(revision["_id"])]:
occurs.append(		occurs.append(
ProvenanceResult(		ProvenanceResult(
content=id,		content=id,
revision=revision["sha1"],		revision=revision["sha1"],
date=datetime.fromtimestamp(revision["ts"], timezone.utc),		date=datetime.fromtimestamp(revision["ts"], timezone.utc),
origin=origin["url"],		origin=origin["url"],
path=path,		path=path,
)		)
)		)
return sorted(occurs, key=lambda x: (x.date, x.revision, x.origin, x.path))[0]		return sorted(occurs, key=lambda x: (x.date, x.revision, x.origin, x.path))[0]

		# FIXME, refactor this
@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_find_all"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_find_all"})
def content_find_all(		def content_find_all(
self, id: Sha1Git, limit: Optional[int] = None		self, id: Sha1Git, limit: Optional[int] = None
) -> Generator[ProvenanceResult, None, None]:		) -> Generator[ProvenanceResult, None, None]:
content = self.db.content.find_one({"sha1": id})		content = self.db.content.find_one({"sha1": id})
if not content:		if not content:
return None		return None

Show All 12 Lines	) -> Generator[ProvenanceResult, None, None]:
content=id,		content=id,
revision=revision["sha1"],		revision=revision["sha1"],
date=datetime.fromtimestamp(revision["ts"], timezone.utc),		date=datetime.fromtimestamp(revision["ts"], timezone.utc),
origin=origin["url"],		origin=origin["url"],
path=path,		path=path,
)		)
)		)
for directory in self.db.directory.find(		for directory in self.db.directory.find(
{"_id": {"$in": [ObjectId(obj_id) for obj_id in content["directory"]]}}		{
		"_id": {
		"$in": [ObjectId(obj_id) for obj_id in content.get("directory", {})]
		}
		}
):		):
for revision in self.db.revision.find(		for revision in self.db.revision.find(
{"_id": {"$in": [ObjectId(obj_id) for obj_id in directory["revision"]]}}		{"_id": {"$in": [ObjectId(obj_id) for obj_id in directory["revision"]]}}
):		):
if revision["preferred"] is not None:		if revision["preferred"] is not None:
origin = self.db.origin.find_one({"sha1": revision["preferred"]})		origin = self.db.origin.find_one({"sha1": revision["preferred"]})
else:		else:
origin = {"url": None}		origin = {"url": None}
Show All 16 Lines	) -> Generator[ProvenanceResult, None, None]:
path=path,		path=path,
)		)
)		)
yield from sorted(occurs, key=lambda x: (x.date, x.revision, x.origin, x.path))		yield from sorted(occurs, key=lambda x: (x.date, x.revision, x.origin, x.path))

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_get"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_get"})
def content_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, datetime]:		def content_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, datetime]:
return {		return {
x["sha1"]: datetime.fromtimestamp(x["ts"], timezone.utc)		cnt["sha1"]: datetime.fromtimestamp(cnt["ts"], timezone.utc)
for x in self.db.content.find(		for cnt in Entity.factory(self.db, "content").find(
{"sha1": {"$in": list(ids)}, "ts": {"$ne": None}},		{"sha1": {"$in": list(ids)}, "ts": {"$ne": None}},
{"sha1": 1, "ts": 1, "_id": 0},		{"sha1": 1, "ts": 1, "_id": 0},
)		)
}		}

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "directory_add"})
def directory_add(
self, dirs: Union[Iterable[Sha1Git], Dict[Sha1Git, Optional[datetime]]]
) -> bool:
data = dirs if isinstance(dirs, dict) else dict.fromkeys(dirs)
existing = {
x["sha1"]: x
for x in self.db.directory.find(
{"sha1": {"$in": list(data)}}, {"sha1": 1, "ts": 1, "_id": 1}
)
}
for sha1, date in data.items():
ts = datetime.timestamp(date) if date is not None else None
if sha1 in existing:
dir = existing[sha1]
if ts is not None and (dir["ts"] is None or ts < dir["ts"]):
self.db.directory.update_one(
{"_id": dir["_id"]}, {"$set": {"ts": ts}}
)
else:
self.db.directory.insert_one({"sha1": sha1, "ts": ts, "revision": {}})
return True

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "directory_get"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "directory_get"})
def directory_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, datetime]:		def directory_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, datetime]:
return {		return {
x["sha1"]: datetime.fromtimestamp(x["ts"], timezone.utc)		dry["sha1"]: datetime.fromtimestamp(dry["ts"], timezone.utc)
for x in self.db.directory.find(		for dry in Entity.factory(self.db, "directory").find(
{"sha1": {"$in": list(ids)}, "ts": {"$ne": None}},		{"sha1": {"$in": list(ids)}, "ts": {"$ne": None}},
{"sha1": 1, "ts": 1, "_id": 0},		{"sha1": 1, "ts": 1, "_id": 0},
)		)
}		}

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "entity_get_all"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "entity_get_all"})
def entity_get_all(self, entity: EntityType) -> Set[Sha1Git]:		def entity_get_all(self, entity: EntityType) -> Set[Sha1Git]:
		# test only method; not using entity factory or cache
return {		return {
x["sha1"]		x["sha1"]
for x in self.db.get_collection(entity.value).find(		for x in self.db.get_collection(entity.value).find(
{}, {"sha1": 1, "_id": 0}		{}, {"sha1": 1, "_id": 0}
)		)
}		}

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "location_add"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "location_add"})
def location_add(self, paths: Iterable[bytes]) -> bool:		def location_add(self, paths: Iterable[bytes]) -> bool:
# TODO: implement this methods if path are to be stored in a separate collection		# TODO: implement this methods if path are to be stored in a separate collection
return True		return True

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "location_get_all"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "location_get_all"})
def location_get_all(self) -> Set[bytes]:		def location_get_all(self) -> Set[bytes]:
		# test only method; not using entity factory or cache
contents = self.db.content.find({}, {"revision": 1, "_id": 0, "directory": 1})		contents = self.db.content.find({}, {"revision": 1, "_id": 0, "directory": 1})
paths: List[Iterable[bytes]] = []		paths: List[Iterable[bytes]] = []
for content in contents:		for content in contents:
paths.extend(value for _, value in content["revision"].items())		paths.extend(value for _, value in content["revision"].items())
paths.extend(value for _, value in content["directory"].items())		paths.extend(value for _, value in content["directory"].items())

dirs = self.db.directory.find({}, {"revision": 1, "_id": 0})		dirs = self.db.directory.find({}, {"revision": 1, "_id": 0})
for each_dir in dirs:		for each_dir in dirs:
paths.extend(value for _, value in each_dir["revision"].items())		paths.extend(value for _, value in each_dir["revision"].items())
return set(sum(paths, []))		return set(sum(paths, []))

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "open"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "open"})
def open(self) -> None:		def open(self) -> None:
if self.engine == "mongomock":		if self.engine == "mongomock":
self.db = mongomock.MongoClient(**self.conn_args).get_database(self.dbname)		self.db = mongomock.MongoClient(**self.conn_args).get_database(self.dbname)
else:		else:
# assume real MongoDB server by default		# assume real MongoDB server by default
self.db = pymongo.MongoClient(**self.conn_args).get_database(self.dbname)		self.db = pymongo.MongoClient(host="127.0.0.1", port=27017, connect=False).get_database(self.dbname)

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "origin_add"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "origin_add"})
def origin_add(self, orgs: Dict[Sha1Git, str]) -> bool:		def origin_add(self, orgs: Dict[Sha1Git, str]) -> bool:
existing = {		writes = [InsertOne({"sha1": sha1, "url": url}) for (sha1, url) in orgs.items()]
x["sha1"]: x		Entity.factory(self.db, "origin").bulk_write(writes)
for x in self.db.origin.find(
{"sha1": {"$in": list(orgs)}}, {"sha1": 1, "url": 1, "_id": 1}
)
}
for sha1, url in orgs.items():
if sha1 not in existing:
# add new origin
self.db.origin.insert_one({"sha1": sha1, "url": url})
return True		return True

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "origin_get"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "origin_get"})
def origin_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, str]:		def origin_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, str]:
return {		return {
x["sha1"]: x["url"]		ori["sha1"]: ori["url"]
for x in self.db.origin.find(		for ori in Entity.factory(self.db, "origin").find(
{"sha1": {"$in": list(ids)}}, {"sha1": 1, "url": 1, "_id": 0}		{"sha1": {"$in": list(ids)}}, {"sha1": 1, "url": 1, "_id": 0}
)		)
}		}

		def _generate_revision_upserts(self, sha1, date=None, preferred=None, upsert=True):
		ts = datetime.timestamp(date) if date is not None else None
		return UpdateOne(
		{"sha1": sha1, "$or": [{"ts": {"$gte": ts}}, {"ts": None}]},
		{
		"$set": {"ts": ts, "preferred": preferred},
		"$setOnInsert": {"origin": [], "revision": []},
		},
		upsert=upsert,
		)

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "revision_add"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "revision_add"})
def revision_add(		def revision_add(
self, revs: Union[Iterable[Sha1Git], Dict[Sha1Git, RevisionData]]		self, revs: Union[Iterable[Sha1Git], Dict[Sha1Git, RevisionData]]
) -> bool:		) -> bool:
		# if isinstance(revs, dict):
		# writes = [
		# self._generate_revision_upserts(sha1, data.date, data.origin)
		# for sha1, data in revs.items()
		# ]
		# else:
		# writes = [
		# self._generate_revision_upserts(sha1)
		# for sha1 in cnts
		# ]
		# Entity.factory(self.db, "revision").bulk_write(writes)
		# return True

data = (		data = (
revs		revs
if isinstance(revs, dict)		if isinstance(revs, dict)
else dict.fromkeys(revs, RevisionData(date=None, origin=None))		else dict.fromkeys(revs, RevisionData(date=None, origin=None))
)		)
existing = {		existing = {
x["sha1"]: x		x["sha1"]: x
for x in self.db.revision.find(		for x in self.db.revision.find(
{"sha1": {"$in": list(data)}},		{"sha1": {"$in": list(data)}},
{"sha1": 1, "ts": 1, "preferred": 1, "_id": 1},		{"sha1": 1, "ts": 1, "preferred": 1, "_id": 1},
)		)
}		}

		statsd.gauge("provenance.revision-cache.size", sys.getsizeof(existing))

for sha1, info in data.items():		for sha1, info in data.items():
ts = datetime.timestamp(info.date) if info.date is not None else None		ts = datetime.timestamp(info.date) if info.date is not None else None
preferred = info.origin		preferred = info.origin
if sha1 in existing:		if sha1 in existing:
rev = existing[sha1]		rev = existing[sha1]
if ts is None or (rev["ts"] is not None and ts >= rev["ts"]):		if ts is None or (rev["ts"] is not None and ts >= rev["ts"]):
ts = rev["ts"]		ts = rev["ts"]
if preferred is None:		if preferred is None:
Show All 13 Lines	) -> bool:
"ts": ts,		"ts": ts,
}		}
)		)
return True		return True

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "revision_get"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "revision_get"})
def revision_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, RevisionData]:		def revision_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, RevisionData]:
return {		return {
x["sha1"]: RevisionData(		rev["sha1"]: RevisionData(
date=datetime.fromtimestamp(x["ts"], timezone.utc) if x["ts"] else None,		date=datetime.fromtimestamp(rev["ts"], timezone.utc)
origin=x["preferred"],		if rev["ts"]
		else None,
		origin=rev["preferred"],
)		)
for x in self.db.revision.find(		for rev in Entity.factory(self.db, "revision").find(
{		{
"sha1": {"$in": list(ids)},		"sha1": {"$in": list(ids)},
"$or": [{"preferred": {"$ne": None}}, {"ts": {"$ne": None}}],		"$or": [{"preferred": {"$ne": None}}, {"ts": {"$ne": None}}],
},		},
{"sha1": 1, "preferred": 1, "ts": 1, "_id": 0},		{"sha1": 1, "preferred": 1, "ts": 1, "_id": 0},
)		)
}		}

Show All 27 Lines	) -> bool:

src_objs = {		src_objs = {
x["sha1"]: x		x["sha1"]: x
for x in self.db.get_collection(src_relation).find(		for x in self.db.get_collection(src_relation).find(
{"sha1": {"$in": list(denorm.keys())}}		{"sha1": {"$in": list(denorm.keys())}}
)		)
}		}

		statsd.gauge("provenance.relation-src-cache.size", sys.getsizeof(src_objs))
		statsd.gauge("provenance.relation-dst-cache.size", sys.getsizeof(dst_objs))

for sha1, dsts in denorm.items():		for sha1, dsts in denorm.items():
# update		# update
if src_relation != "revision":		if src_relation != "revision":
k = {		k = {
obj_id: list(set(paths + dsts.get(obj_id, [])))		obj_id: list(set(paths + dsts.get(obj_id, [])))
for obj_id, paths in src_objs[sha1][dst_relation].items()		for obj_id, paths in src_objs[sha1].get(dst_relation, {}).items()
}		}
self.db.get_collection(src_relation).update_one(		self.db.get_collection(src_relation).update_one(
{"_id": src_objs[sha1]["_id"]},		{"_id": src_objs[sha1]["_id"]},
{"$set": {dst_relation: dict(dsts, **k)}},		{"$set": {dst_relation: dict(dsts, **k)}},
)		)
else:		else:
self.db.get_collection(src_relation).update_one(		self.db.get_collection(src_relation).update_one(
{"_id": src_objs[sha1]["_id"]},		{"_id": src_objs[sha1]["_id"]},
▲ Show 20 Lines • Show All 53 Lines • ▼ Show 20 Lines	) -> Dict[Sha1Git, Set[RelationData]]:
else:		else:
dst_objs = {		dst_objs = {
x["sha1"]: x["_id"]		x["sha1"]: x["_id"]
for x in self.db.get_collection(dst).find(		for x in self.db.get_collection(dst).find(
{"sha1": {"$in": list(sha1s)}}, {"_id": 1, "sha1": 1}		{"sha1": {"$in": list(sha1s)}}, {"_id": 1, "sha1": 1}
)		)
}		}
src_objs = {		src_objs = {
x["sha1"]: x[dst]		x["sha1"]: x.get(dst, {})
for x in self.db.get_collection(src).find(		for x in self.db.get_collection(src).find(
{}, {"_id": 0, "sha1": 1, dst: 1}		{}, {"_id": 0, "sha1": 1, dst: 1}
)		)
}		}
result: Dict[Sha1Git, Set[RelationData]] = {}		result: Dict[Sha1Git, Set[RelationData]] = {}
if src != "revision":		if src != "revision":
for dst_sha1, dst_obj_id in dst_objs.items():		for dst_sha1, dst_obj_id in dst_objs.items():
for src_sha1, denorm in src_objs.items():		for src_sha1, denorm in src_objs.items():
Show All 13 Lines	) -> Dict[Sha1Git, Set[RelationData]]:
RelationData(dst=dst_sha1, path=None)		RelationData(dst=dst_sha1, path=None)
)		)
return result		return result

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "relation_get_all"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "relation_get_all"})
def relation_get_all(		def relation_get_all(
self, relation: RelationType		self, relation: RelationType
) -> Dict[Sha1Git, Set[RelationData]]:		) -> Dict[Sha1Git, Set[RelationData]]:
		# test only method; not using entity factory or cache
src, *_, dst = relation.value.split("_")		src, *_, dst = relation.value.split("_")
empty: Union[Dict[str, bytes], List[str]] = {} if src != "revision" else []		empty: Union[Dict[str, bytes], List[str]] = {} if src != "revision" else []
src_objs = {		src_objs = {
x["sha1"]: x[dst]		x["sha1"]: x.get(dst, {})
for x in self.db.get_collection(src).find(		for x in self.db.get_collection(src).find(
{dst: {"$ne": empty}}, {"_id": 0, "sha1": 1, dst: 1}		{dst: {"$ne": empty}}, {"_id": 0, "sha1": 1, dst: 1}
)		)
}		}
dst_ids = list(		dst_ids = list(
{ObjectId(obj_id) for _, value in src_objs.items() for obj_id in value}		{ObjectId(obj_id) for _, value in src_objs.items() for obj_id in value}
)		)
dst_objs = {		dst_objs = {
Show All 22 Lines	) -> Dict[Sha1Git, Set[RelationData]]:
if dst_obj_id == dst_obj_ref		if dst_obj_id == dst_obj_ref
}		}
for src_sha1, denorm in src_objs.items()		for src_sha1, denorm in src_objs.items()
}		}

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "with_path"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "with_path"})
def with_path(self) -> bool:		def with_path(self) -> bool:
return True		return True

		def entity_get(self, entity, filters=None):
		"""
		A method used only by tests
		Returns a list of entities after applying the filter
		"""
		# FIXME, this should be added to the interface to make
		# tests backend agnostic. Adding this in SQL could be tricky
		# FIXME, change to entity factory
		return list(Entity.factory(self.db, entity).find(filters))

		def entity_add(self, entity, data):
		"""
		A method used only by tests
		"""
		# FIXME, this should be added to the interface to make
		# tests backend agnostic. Adding this in SQL could be tricky
		# FIXME, change to entity factory
		return Entity.factory(self.db, entity).insert(data)

		def create_indexes(self, entity, fields, unique=False):
		Entity.factory(self.db, entity).create_index(fields, unique)

View Options

swh/graphql/tests/functional/test_release_node.py

# Copyright (C) 2021 The Software Heritage developers		# Copyright (C) 2021 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution		# See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version		# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information		# See top-level LICENSE file for more information

from __future__ import annotations		from __future__ import annotations

from datetime import datetime, timezone		from datetime import datetime, timezone
import os		import os
from types import TracebackType		from types import TracebackType
from typing import Any, Dict, Generator, Iterable, List, Optional, Set, Type, Union		from typing import Any, Dict, Generator, Iterable, List, Optional, Set, Type, Union

from bson import ObjectId		from bson import ObjectId
import mongomock		import mongomock
import pymongo		import pymongo
		from pymongo import InsertOne, UpdateOne
		import sys

from swh.core.statsd import statsd		from swh.core.statsd import statsd
from swh.model.model import Sha1Git		from swh.model.model import Sha1Git

from ..interface import (		from ..interface import (
EntityType,		EntityType,
ProvenanceResult,		ProvenanceResult,
ProvenanceStorageInterface,		ProvenanceStorageInterface,
RelationData,		RelationData,
RelationType,		RelationType,
RevisionData,		RevisionData,
)		)
		from .entity import Entity

STORAGE_DURATION_METRIC = "swh_provenance_storage_mongodb_duration_seconds"		STORAGE_DURATION_METRIC = "swh_provenance_storage_mongodb_duration_seconds"


class ProvenanceStorageMongoDb:		class ProvenanceStorageMongoDb:
def __init__(self, engine: str, **kwargs):		def __init__(self, engine: str, **kwargs):
self.engine = engine		self.engine = engine
self.dbname = kwargs.pop("dbname")		self.dbname = kwargs.pop("dbname")
Show All 10 Lines	def __exit__(
exc_tb: Optional[TracebackType],		exc_tb: Optional[TracebackType],
) -> None:		) -> None:
self.close()		self.close()

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "close"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "close"})
def close(self) -> None:		def close(self) -> None:
self.db.client.close()		self.db.client.close()

		def _generate_date_upserts(self, sha1, date, inserts={}, upsert=True):
		ts = datetime.timestamp(date) if date is not None else None
		# update only those with date either as None or later than the given one
		return UpdateOne(
		{"sha1": sha1, "$or": [{"ts": {"$gte": ts}}, {"ts": None}]},
		{
		"$set": {"ts": ts},
		"$setOnInsert": inserts,
		},
		upsert=upsert,
		)

		def _generate_date_upserts_alternate(self, sha1, date, inserts={}, upsert=True):
		# FIXME, Compare the performance and decide which one to keep
		ts = datetime.timestamp(date) if date is not None else None
		# update only those with date either as None or later than the given one
		return UpdateOne(
		{"sha1": sha1},
		[
		{
		"$set": {
		"ts": {
		"$cond": {
		"if": {"$gt": ["$ts", ts]},
		"then": ts,
		"else": None,
		}
		}
		},
		},
		],
		upsert=upsert,
		)

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_add"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_add"})
def content_add(		def content_add(
self, cnts: Union[Iterable[Sha1Git], Dict[Sha1Git, Optional[datetime]]]		self, cnts: Union[Iterable[Sha1Git], Dict[Sha1Git, Optional[datetime]]]
) -> bool:		) -> bool:
data = cnts if isinstance(cnts, dict) else dict.fromkeys(cnts)		default_inserts = {"revision": {}, "directory": {}}
existing = {		if isinstance(cnts, dict):
x["sha1"]: x		writes = [
for x in self.db.content.find(		self._generate_date_upserts(sha1, date, default_inserts)
{"sha1": {"$in": list(data)}}, {"sha1": 1, "ts": 1, "_id": 1}		for sha1, date in cnts.items()
)		]
}		else:
for sha1, date in data.items():		writes = [
ts = datetime.timestamp(date) if date is not None else None		self._generate_date_upserts(sha1, None, default_inserts)
if sha1 in existing:		for sha1 in cnts
cnt = existing[sha1]		]
if ts is not None and (cnt["ts"] is None or ts < cnt["ts"]):		statsd.gauge("provenance.content-list.size", sys.getsizeof(writes))
self.db.content.update_one(		Entity.factory(self.db, "content").bulk_write(writes)
{"_id": cnt["_id"]}, {"$set": {"ts": ts}}		return True
)
else:		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "directory_add"})
self.db.content.insert_one(		def directory_add(
		self, dirs: Union[Iterable[Sha1Git], Dict[Sha1Git, Optional[datetime]]]
		) -> bool:
		default_inserts = {"revision": {}}
		if isinstance(dirs, dict):
		writes = [
		self._generate_date_upserts(sha1, date, default_inserts)
		for sha1, date in dirs.items()
		]
		else:
		writes = [
		self._generate_date_upserts(sha1, None, default_inserts)
		for sha1 in dirs
		]
		statsd.gauge("provenance.directory-list.size", sys.getsizeof(writes))
		Entity.factory(self.db, "directory").bulk_write(writes)
		return True

		def _get_oldest_revision_from_content(self, content):
		# FIXME, returing with the assumption that content has all the
		# revisons in the array
		# Change to seperate collection content_in_revision if gets too big
		return Entity.factory(self.db, "revision").find_one(
{		{
"sha1": sha1,		"_id": {"$in": [ObjectId(obj_id) for obj_id in content["revision"]]},
"ts": ts,		"ts": content["ts"],
"revision": {},
"directory": {},
}		}
)		)
return True
		def _get_preferred_origin(self, revision):
		if revision.get("preferred"):
		return Entity.factory(self.db, "origin").find_one(
		{"sha1": revision["preferred"]}
		)
		return None

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_find_first"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_find_first"})
def content_find_first(self, id: Sha1Git) -> Optional[ProvenanceResult]:		def content_find_first(self, id: Sha1Git) -> Optional[ProvenanceResult]:
		content = Entity.factory(self.db, "content").find_one({"sha1": id})
		if not content:
		return None

		oldest_revision = self._get_oldest_revision_from_content(content)
		origin = self._get_preferred_origin(oldest_revision)

		return ProvenanceResult(
		content=id,
		revision=oldest_revision["sha1"],
		date=datetime.fromtimestamp(oldest_revision["ts"], timezone.utc),
		origin=origin["url"] if origin else None,
		path=sorted(content["revision"][str(oldest_revision["_id"])])[0]
		# FIXME, this is likely to be wrong, tests pass
		)

		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_find_first"})
		def content_find_first_old(self, id: Sha1Git) -> Optional[ProvenanceResult]:
# get all the revisions		# get all the revisions
# iterate and find the earliest		# iterate and find the earliest
content = self.db.content.find_one({"sha1": id})		content = self.db.content.find_one({"sha1": id})
if not content:		if not content:
return None		return None

occurs = []		occurs = []
for revision in self.db.revision.find(		for revision in self.db.revision.find(
{"_id": {"$in": [ObjectId(obj_id) for obj_id in content["revision"]]}}		{"_id": {"$in": [ObjectId(obj_id) for obj_id in content["revision"]]}}
):		):
if revision["preferred"] is not None:		if revision["preferred"] is not None:
origin = self.db.origin.find_one({"sha1": revision["preferred"]})		origin = self.db.origin.find_one({"sha1": revision["preferred"]})
else:		else:
origin = {"url": None}		origin = {"url": None}

for path in content["revision"][str(revision["_id"])]:		for path in content["revision"][str(revision["_id"])]:
occurs.append(		occurs.append(
ProvenanceResult(		ProvenanceResult(
content=id,		content=id,
revision=revision["sha1"],		revision=revision["sha1"],
date=datetime.fromtimestamp(revision["ts"], timezone.utc),		date=datetime.fromtimestamp(revision["ts"], timezone.utc),
origin=origin["url"],		origin=origin["url"],
path=path,		path=path,
)		)
)		)
return sorted(occurs, key=lambda x: (x.date, x.revision, x.origin, x.path))[0]		return sorted(occurs, key=lambda x: (x.date, x.revision, x.origin, x.path))[0]

		# FIXME, refactor this
@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_find_all"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_find_all"})
def content_find_all(		def content_find_all(
self, id: Sha1Git, limit: Optional[int] = None		self, id: Sha1Git, limit: Optional[int] = None
) -> Generator[ProvenanceResult, None, None]:		) -> Generator[ProvenanceResult, None, None]:
content = self.db.content.find_one({"sha1": id})		content = self.db.content.find_one({"sha1": id})
if not content:		if not content:
return None		return None

Show All 12 Lines	) -> Generator[ProvenanceResult, None, None]:
content=id,		content=id,
revision=revision["sha1"],		revision=revision["sha1"],
date=datetime.fromtimestamp(revision["ts"], timezone.utc),		date=datetime.fromtimestamp(revision["ts"], timezone.utc),
origin=origin["url"],		origin=origin["url"],
path=path,		path=path,
)		)
)		)
for directory in self.db.directory.find(		for directory in self.db.directory.find(
{"_id": {"$in": [ObjectId(obj_id) for obj_id in content["directory"]]}}		{
		"_id": {
		"$in": [ObjectId(obj_id) for obj_id in content.get("directory", {})]
		}
		}
):		):
for revision in self.db.revision.find(		for revision in self.db.revision.find(
{"_id": {"$in": [ObjectId(obj_id) for obj_id in directory["revision"]]}}		{"_id": {"$in": [ObjectId(obj_id) for obj_id in directory["revision"]]}}
):		):
if revision["preferred"] is not None:		if revision["preferred"] is not None:
origin = self.db.origin.find_one({"sha1": revision["preferred"]})		origin = self.db.origin.find_one({"sha1": revision["preferred"]})
else:		else:
origin = {"url": None}		origin = {"url": None}
Show All 16 Lines	) -> Generator[ProvenanceResult, None, None]:
path=path,		path=path,
)		)
)		)
yield from sorted(occurs, key=lambda x: (x.date, x.revision, x.origin, x.path))		yield from sorted(occurs, key=lambda x: (x.date, x.revision, x.origin, x.path))

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_get"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_get"})
def content_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, datetime]:		def content_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, datetime]:
return {		return {
x["sha1"]: datetime.fromtimestamp(x["ts"], timezone.utc)		cnt["sha1"]: datetime.fromtimestamp(cnt["ts"], timezone.utc)
for x in self.db.content.find(		for cnt in Entity.factory(self.db, "content").find(
{"sha1": {"$in": list(ids)}, "ts": {"$ne": None}},		{"sha1": {"$in": list(ids)}, "ts": {"$ne": None}},
{"sha1": 1, "ts": 1, "_id": 0},		{"sha1": 1, "ts": 1, "_id": 0},
)		)
}		}

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "directory_add"})
def directory_add(
self, dirs: Union[Iterable[Sha1Git], Dict[Sha1Git, Optional[datetime]]]
) -> bool:
data = dirs if isinstance(dirs, dict) else dict.fromkeys(dirs)
existing = {
x["sha1"]: x
for x in self.db.directory.find(
{"sha1": {"$in": list(data)}}, {"sha1": 1, "ts": 1, "_id": 1}
)
}
for sha1, date in data.items():
ts = datetime.timestamp(date) if date is not None else None
if sha1 in existing:
dir = existing[sha1]
if ts is not None and (dir["ts"] is None or ts < dir["ts"]):
self.db.directory.update_one(
{"_id": dir["_id"]}, {"$set": {"ts": ts}}
)
else:
self.db.directory.insert_one({"sha1": sha1, "ts": ts, "revision": {}})
return True

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "directory_get"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "directory_get"})
def directory_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, datetime]:		def directory_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, datetime]:
return {		return {
x["sha1"]: datetime.fromtimestamp(x["ts"], timezone.utc)		dry["sha1"]: datetime.fromtimestamp(dry["ts"], timezone.utc)
for x in self.db.directory.find(		for dry in Entity.factory(self.db, "directory").find(
{"sha1": {"$in": list(ids)}, "ts": {"$ne": None}},		{"sha1": {"$in": list(ids)}, "ts": {"$ne": None}},
{"sha1": 1, "ts": 1, "_id": 0},		{"sha1": 1, "ts": 1, "_id": 0},
)		)
}		}

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "entity_get_all"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "entity_get_all"})
def entity_get_all(self, entity: EntityType) -> Set[Sha1Git]:		def entity_get_all(self, entity: EntityType) -> Set[Sha1Git]:
		# test only method; not using entity factory or cache
return {		return {
x["sha1"]		x["sha1"]
for x in self.db.get_collection(entity.value).find(		for x in self.db.get_collection(entity.value).find(
{}, {"sha1": 1, "_id": 0}		{}, {"sha1": 1, "_id": 0}
)		)
}		}

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "location_add"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "location_add"})
def location_add(self, paths: Iterable[bytes]) -> bool:		def location_add(self, paths: Iterable[bytes]) -> bool:
# TODO: implement this methods if path are to be stored in a separate collection		# TODO: implement this methods if path are to be stored in a separate collection
return True		return True

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "location_get_all"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "location_get_all"})
def location_get_all(self) -> Set[bytes]:		def location_get_all(self) -> Set[bytes]:
		# test only method; not using entity factory or cache
contents = self.db.content.find({}, {"revision": 1, "_id": 0, "directory": 1})		contents = self.db.content.find({}, {"revision": 1, "_id": 0, "directory": 1})
paths: List[Iterable[bytes]] = []		paths: List[Iterable[bytes]] = []
for content in contents:		for content in contents:
paths.extend(value for _, value in content["revision"].items())		paths.extend(value for _, value in content["revision"].items())
paths.extend(value for _, value in content["directory"].items())		paths.extend(value for _, value in content["directory"].items())

dirs = self.db.directory.find({}, {"revision": 1, "_id": 0})		dirs = self.db.directory.find({}, {"revision": 1, "_id": 0})
for each_dir in dirs:		for each_dir in dirs:
paths.extend(value for _, value in each_dir["revision"].items())		paths.extend(value for _, value in each_dir["revision"].items())
return set(sum(paths, []))		return set(sum(paths, []))

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "open"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "open"})
def open(self) -> None:		def open(self) -> None:
if self.engine == "mongomock":		if self.engine == "mongomock":
self.db = mongomock.MongoClient(**self.conn_args).get_database(self.dbname)		self.db = mongomock.MongoClient(**self.conn_args).get_database(self.dbname)
else:		else:
# assume real MongoDB server by default		# assume real MongoDB server by default
self.db = pymongo.MongoClient(**self.conn_args).get_database(self.dbname)		self.db = pymongo.MongoClient(host="127.0.0.1", port=27017, connect=False).get_database(self.dbname)

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "origin_add"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "origin_add"})
def origin_add(self, orgs: Dict[Sha1Git, str]) -> bool:		def origin_add(self, orgs: Dict[Sha1Git, str]) -> bool:
existing = {		writes = [InsertOne({"sha1": sha1, "url": url}) for (sha1, url) in orgs.items()]
x["sha1"]: x		Entity.factory(self.db, "origin").bulk_write(writes)
for x in self.db.origin.find(
{"sha1": {"$in": list(orgs)}}, {"sha1": 1, "url": 1, "_id": 1}
)
}
for sha1, url in orgs.items():
if sha1 not in existing:
# add new origin
self.db.origin.insert_one({"sha1": sha1, "url": url})
return True		return True

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "origin_get"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "origin_get"})
def origin_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, str]:		def origin_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, str]:
return {		return {
x["sha1"]: x["url"]		ori["sha1"]: ori["url"]
for x in self.db.origin.find(		for ori in Entity.factory(self.db, "origin").find(
{"sha1": {"$in": list(ids)}}, {"sha1": 1, "url": 1, "_id": 0}		{"sha1": {"$in": list(ids)}}, {"sha1": 1, "url": 1, "_id": 0}
)		)
}		}

		def _generate_revision_upserts(self, sha1, date=None, preferred=None, upsert=True):
		ts = datetime.timestamp(date) if date is not None else None
		return UpdateOne(
		{"sha1": sha1, "$or": [{"ts": {"$gte": ts}}, {"ts": None}]},
		{
		"$set": {"ts": ts, "preferred": preferred},
		"$setOnInsert": {"origin": [], "revision": []},
		},
		upsert=upsert,
		)

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "revision_add"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "revision_add"})
def revision_add(		def revision_add(
self, revs: Union[Iterable[Sha1Git], Dict[Sha1Git, RevisionData]]		self, revs: Union[Iterable[Sha1Git], Dict[Sha1Git, RevisionData]]
) -> bool:		) -> bool:
		# if isinstance(revs, dict):
		# writes = [
		# self._generate_revision_upserts(sha1, data.date, data.origin)
		# for sha1, data in revs.items()
		# ]
		# else:
		# writes = [
		# self._generate_revision_upserts(sha1)
		# for sha1 in cnts
		# ]
		# Entity.factory(self.db, "revision").bulk_write(writes)
		# return True

data = (		data = (
revs		revs
if isinstance(revs, dict)		if isinstance(revs, dict)
else dict.fromkeys(revs, RevisionData(date=None, origin=None))		else dict.fromkeys(revs, RevisionData(date=None, origin=None))
)		)
existing = {		existing = {
x["sha1"]: x		x["sha1"]: x
for x in self.db.revision.find(		for x in self.db.revision.find(
{"sha1": {"$in": list(data)}},		{"sha1": {"$in": list(data)}},
{"sha1": 1, "ts": 1, "preferred": 1, "_id": 1},		{"sha1": 1, "ts": 1, "preferred": 1, "_id": 1},
)		)
}		}

		statsd.gauge("provenance.revision-cache.size", sys.getsizeof(existing))

for sha1, info in data.items():		for sha1, info in data.items():
ts = datetime.timestamp(info.date) if info.date is not None else None		ts = datetime.timestamp(info.date) if info.date is not None else None
preferred = info.origin		preferred = info.origin
if sha1 in existing:		if sha1 in existing:
rev = existing[sha1]		rev = existing[sha1]
if ts is None or (rev["ts"] is not None and ts >= rev["ts"]):		if ts is None or (rev["ts"] is not None and ts >= rev["ts"]):
ts = rev["ts"]		ts = rev["ts"]
if preferred is None:		if preferred is None:
Show All 13 Lines	) -> bool:
"ts": ts,		"ts": ts,
}		}
)		)
return True		return True

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "revision_get"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "revision_get"})
def revision_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, RevisionData]:		def revision_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, RevisionData]:
return {		return {
x["sha1"]: RevisionData(		rev["sha1"]: RevisionData(
date=datetime.fromtimestamp(x["ts"], timezone.utc) if x["ts"] else None,		date=datetime.fromtimestamp(rev["ts"], timezone.utc)
origin=x["preferred"],		if rev["ts"]
		else None,
		origin=rev["preferred"],
)		)
for x in self.db.revision.find(		for rev in Entity.factory(self.db, "revision").find(
{		{
"sha1": {"$in": list(ids)},		"sha1": {"$in": list(ids)},
"$or": [{"preferred": {"$ne": None}}, {"ts": {"$ne": None}}],		"$or": [{"preferred": {"$ne": None}}, {"ts": {"$ne": None}}],
},		},
{"sha1": 1, "preferred": 1, "ts": 1, "_id": 0},		{"sha1": 1, "preferred": 1, "ts": 1, "_id": 0},
)		)
}		}

Show All 27 Lines	) -> bool:

src_objs = {		src_objs = {
x["sha1"]: x		x["sha1"]: x
for x in self.db.get_collection(src_relation).find(		for x in self.db.get_collection(src_relation).find(
{"sha1": {"$in": list(denorm.keys())}}		{"sha1": {"$in": list(denorm.keys())}}
)		)
}		}

		statsd.gauge("provenance.relation-src-cache.size", sys.getsizeof(src_objs))
		statsd.gauge("provenance.relation-dst-cache.size", sys.getsizeof(dst_objs))

for sha1, dsts in denorm.items():		for sha1, dsts in denorm.items():
# update		# update
if src_relation != "revision":		if src_relation != "revision":
k = {		k = {
obj_id: list(set(paths + dsts.get(obj_id, [])))		obj_id: list(set(paths + dsts.get(obj_id, [])))
for obj_id, paths in src_objs[sha1][dst_relation].items()		for obj_id, paths in src_objs[sha1].get(dst_relation, {}).items()
}		}
self.db.get_collection(src_relation).update_one(		self.db.get_collection(src_relation).update_one(
{"_id": src_objs[sha1]["_id"]},		{"_id": src_objs[sha1]["_id"]},
{"$set": {dst_relation: dict(dsts, **k)}},		{"$set": {dst_relation: dict(dsts, **k)}},
)		)
else:		else:
self.db.get_collection(src_relation).update_one(		self.db.get_collection(src_relation).update_one(
{"_id": src_objs[sha1]["_id"]},		{"_id": src_objs[sha1]["_id"]},
▲ Show 20 Lines • Show All 53 Lines • ▼ Show 20 Lines	) -> Dict[Sha1Git, Set[RelationData]]:
else:		else:
dst_objs = {		dst_objs = {
x["sha1"]: x["_id"]		x["sha1"]: x["_id"]
for x in self.db.get_collection(dst).find(		for x in self.db.get_collection(dst).find(
{"sha1": {"$in": list(sha1s)}}, {"_id": 1, "sha1": 1}		{"sha1": {"$in": list(sha1s)}}, {"_id": 1, "sha1": 1}
)		)
}		}
src_objs = {		src_objs = {
x["sha1"]: x[dst]		x["sha1"]: x.get(dst, {})
for x in self.db.get_collection(src).find(		for x in self.db.get_collection(src).find(
{}, {"_id": 0, "sha1": 1, dst: 1}		{}, {"_id": 0, "sha1": 1, dst: 1}
)		)
}		}
result: Dict[Sha1Git, Set[RelationData]] = {}		result: Dict[Sha1Git, Set[RelationData]] = {}
if src != "revision":		if src != "revision":
for dst_sha1, dst_obj_id in dst_objs.items():		for dst_sha1, dst_obj_id in dst_objs.items():
for src_sha1, denorm in src_objs.items():		for src_sha1, denorm in src_objs.items():
Show All 13 Lines	) -> Dict[Sha1Git, Set[RelationData]]:
RelationData(dst=dst_sha1, path=None)		RelationData(dst=dst_sha1, path=None)
)		)
return result		return result

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "relation_get_all"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "relation_get_all"})
def relation_get_all(		def relation_get_all(
self, relation: RelationType		self, relation: RelationType
) -> Dict[Sha1Git, Set[RelationData]]:		) -> Dict[Sha1Git, Set[RelationData]]:
		# test only method; not using entity factory or cache
src, *_, dst = relation.value.split("_")		src, *_, dst = relation.value.split("_")
empty: Union[Dict[str, bytes], List[str]] = {} if src != "revision" else []		empty: Union[Dict[str, bytes], List[str]] = {} if src != "revision" else []
src_objs = {		src_objs = {
x["sha1"]: x[dst]		x["sha1"]: x.get(dst, {})
for x in self.db.get_collection(src).find(		for x in self.db.get_collection(src).find(
{dst: {"$ne": empty}}, {"_id": 0, "sha1": 1, dst: 1}		{dst: {"$ne": empty}}, {"_id": 0, "sha1": 1, dst: 1}
)		)
}		}
dst_ids = list(		dst_ids = list(
{ObjectId(obj_id) for _, value in src_objs.items() for obj_id in value}		{ObjectId(obj_id) for _, value in src_objs.items() for obj_id in value}
)		)
dst_objs = {		dst_objs = {
Show All 22 Lines	) -> Dict[Sha1Git, Set[RelationData]]:
if dst_obj_id == dst_obj_ref		if dst_obj_id == dst_obj_ref
}		}
for src_sha1, denorm in src_objs.items()		for src_sha1, denorm in src_objs.items()
}		}

@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "with_path"})		@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "with_path"})
def with_path(self) -> bool:		def with_path(self) -> bool:
return True		return True

		def entity_get(self, entity, filters=None):
		"""
		A method used only by tests
		Returns a list of entities after applying the filter
		"""
		# FIXME, this should be added to the interface to make
		# tests backend agnostic. Adding this in SQL could be tricky
		# FIXME, change to entity factory
		return list(Entity.factory(self.db, entity).find(filters))

		def entity_add(self, entity, data):
		"""
		A method used only by tests
		"""
		# FIXME, this should be added to the interface to make
		# tests backend agnostic. Adding this in SQL could be tricky
		# FIXME, change to entity factory
		return Entity.factory(self.db, entity).insert(data)

		def create_indexes(self, entity, fields, unique=False):
		Entity.factory(self.db, entity).create_index(fields, unique)

View Options

swh/graphql/tests/functional/test_revision.py

View Options

swh/graphql/tests/unit/resolvers/test_scalars.py

Return more information with revision and release datesClosedPublicActions

Details

Diff Detail

Event Timeline

Patch application report for D8425 (id=30376)

Changes applied before test

Patch application report for D8425 (id=30377)

Changes applied before test

Patch application report for D8425 (id=30439)

Changes applied before test

Patch application report for D8425 (id=30441)

Changes applied before test

Patch application report for D8425 (id=31584)

Changes applied before test

Patch application report for D8425 (id=32118)

Changes applied before test

Patch application report for D8425 (id=32323)

Changes applied before test

Patch application report for D8425 (id=32332)

Changes applied before test

Patch application report for D8425 (id=32333)

Changes applied before test

Patch application report for D8425 (id=32340)

Changes applied before test

Patch application report for D8425 (id=32341)

Changes applied before test

Revision ContentsChangeset List

Diff 32118

swh/graphql/app.py

swh/graphql/resolvers/resolvers.py

swh/graphql/resolvers/scalars.py

swh/graphql/schema/schema.graphql

swh/graphql/tests/functional/test_release_node.py

swh/graphql/tests/functional/test_revision.py

swh/graphql/tests/unit/resolvers/test_scalars.py

Return more information with revision and release dates
ClosedPublic
Actions

Revision Contents
Changeset List