Page MenuHomeSoftware Heritage
Differential D6231 Diff 22575 swh/provenance/provenance.py

Changeset View

Standalone View

View Options

swh/provenance/provenance.py

Show All 11 Lines
from swh.model.model import Sha1Git from swh.model.model import Sha1Git
from .interface import ( from .interface import (
ProvenanceResult, ProvenanceResult,
ProvenanceStorageInterface, ProvenanceStorageInterface,
RelationData, RelationData,
RelationType, RelationType,
RevisionData,
) )
from .model import DirectoryEntry, FileEntry, OriginEntry, RevisionEntry from .model import DirectoryEntry, FileEntry, OriginEntry, RevisionEntry
LOGGER = logging.getLogger(__name__) LOGGER = logging.getLogger(__name__)
class DatetimeCache(TypedDict): class DatetimeCache(TypedDict):
data: Dict[Sha1Git, Optional[datetime]] data: Dict[Sha1Git, Optional[datetime]]
▲ Show 20 Lines • Show All 46 Lines • ▼ Show 20 Lines def __init__(self, storage: ProvenanceStorageInterface) -> None:
self.cache = new_cache() self.cache = new_cache()
def clear_caches(self) -> None: def clear_caches(self) -> None:
self.cache = new_cache() self.cache = new_cache()
def flush(self) -> None: def flush(self) -> None:
# Revision-content layer insertions ############################################ # Revision-content layer insertions ############################################
# After relations, dates for the entities can be safely set, acknowledging that
# these entities won't need to be reprocessed in case of failure.
cnts: Dict[Sha1Git, Optional[datetime]] = {
src: None
for src, _, _ in self.cache["content_in_revision"]
| self.cache["content_in_directory"]
}
if cnts:
while not self.storage.content_add(cnts):
LOGGER.warning(
"Unable to write content entities to the storage. Retrying..."
)
dirs: Dict[Sha1Git, Optional[datetime]] = {
dst: None for _, dst, _ in self.cache["content_in_directory"]
}
if dirs:
while not self.storage.directory_add(dirs):
LOGGER.warning(
"Unable to write directory entities to the storage. Retrying..."
)
revs: Dict[Sha1Git, RevisionData] = {
dst: RevisionData(date=None, origin=None)
for _, dst, _ in self.cache["content_in_revision"]
| self.cache["directory_in_revision"]
}
if revs:
while not self.storage.revision_add(revs):
LOGGER.warning(
"Unable to write revision entities to the storage. Retrying..."
)
paths = {
path
for _, _, path in self.cache["content_in_revision"]
| self.cache["content_in_directory"]
| self.cache["directory_in_revision"]
}
if paths:
while not self.storage.location_add(paths):
LOGGER.warning(
"Unable to write locations entities to the storage. Retrying..."
)
# For this layer, relations need to be inserted first so that, in case of # For this layer, relations need to be inserted first so that, in case of
# failure, reprocessing the input does not generated an inconsistent database. # failure, reprocessing the input does not generated an inconsistent database.
if self.cache["content_in_revision"]: if self.cache["content_in_revision"]:
while not self.storage.relation_add( while not self.storage.relation_add(
RelationType.CNT_EARLY_IN_REV, RelationType.CNT_EARLY_IN_REV,
( (
RelationData(src=src, dst=dst, path=path) RelationData(src=src, dst=dst, path=path)
for src, dst, path in self.cache["content_in_revision"] for src, dst, path in self.cache["content_in_revision"]
Show All 27 Lines def flush(self) -> None:
): ):
LOGGER.warning( LOGGER.warning(
"Unable to write %s rows to the storage. Retrying...", "Unable to write %s rows to the storage. Retrying...",
RelationType.DIR_IN_REV, RelationType.DIR_IN_REV,
) )
# After relations, dates for the entities can be safely set, acknowledging that # After relations, dates for the entities can be safely set, acknowledging that
# these entities won't need to be reprocessed in case of failure. # these entities won't need to be reprocessed in case of failure.
dates = { cnts = {
sha1: date sha1: date
for sha1, date in self.cache["content"]["data"].items() for sha1, date in self.cache["content"]["data"].items()
if sha1 in self.cache["content"]["added"] and date is not None if sha1 in self.cache["content"]["added"] and date is not None
} }
if dates: if cnts:
while not self.storage.content_set_date(dates): while not self.storage.content_add(cnts):
LOGGER.warning( LOGGER.warning(
"Unable to write content dates to the storage. Retrying..." "Unable to write content dates to the storage. Retrying..."
) )
dates = { dirs = {
sha1: date sha1: date
for sha1, date in self.cache["directory"]["data"].items() for sha1, date in self.cache["directory"]["data"].items()
if sha1 in self.cache["directory"]["added"] and date is not None if sha1 in self.cache["directory"]["added"] and date is not None
} }
if dates: if dirs:
while not self.storage.directory_set_date(dates): while not self.storage.directory_add(dirs):
LOGGER.warning( LOGGER.warning(
"Unable to write directory dates to the storage. Retrying..." "Unable to write directory dates to the storage. Retrying..."
) )
dates = { revs = {
sha1: date sha1: RevisionData(date=date, origin=None)
for sha1, date in self.cache["revision"]["data"].items() for sha1, date in self.cache["revision"]["data"].items()
if sha1 in self.cache["revision"]["added"] and date is not None if sha1 in self.cache["revision"]["added"] and date is not None
} }
if dates: if revs:
while not self.storage.revision_set_date(dates): while not self.storage.revision_add(revs):
LOGGER.warning( LOGGER.warning(
"Unable to write revision dates to the storage. Retrying..." "Unable to write revision dates to the storage. Retrying..."
) )
# Origin-revision layer insertions ############################################# # Origin-revision layer insertions #############################################
# Origins urls should be inserted first so that internal ids' resolution works # Origins and revisions should be inserted first so that internal ids'
# properly. # resolution works properly.
urls = { urls = {
sha1: url sha1: url
for sha1, url in self.cache["origin"]["data"].items() for sha1, url in self.cache["origin"]["data"].items()
if sha1 in self.cache["origin"]["added"] if sha1 in self.cache["origin"]["added"]
} }
if urls: if urls:
while not self.storage.origin_set_url(urls): while not self.storage.origin_add(urls):
LOGGER.warning( LOGGER.warning(
"Unable to write origins urls to the storage. Retrying..." "Unable to write origins urls to the storage. Retrying..."
) )
revs = {
# Destinations in this relation should match origins in the next one
**{
src: RevisionData(date=None, origin=None)
for src in self.cache["revision_before_revision"]
},
**{
# This relation comes second so that non-None origins take precedence
src: RevisionData(date=None, origin=org)
for src, org in self.cache["revision_in_origin"]
},
}
if revs:
while not self.storage.revision_add(revs):
LOGGER.warning(
"Unable to write revision entities to the storage. Retrying..."
)
# Second, flat models for revisions' histories (ie. revision-before-revision). # Second, flat models for revisions' histories (ie. revision-before-revision).
data: Iterable[RelationData] = sum( data: Iterable[RelationData] = sum(
[ [
[ [
RelationData(src=prev, dst=next, path=None) RelationData(src=prev, dst=next, path=None)
for next in self.cache["revision_before_revision"][prev] for next in self.cache["revision_before_revision"][prev]
] ]
for prev in self.cache["revision_before_revision"] for prev in self.cache["revision_before_revision"]
Show All 17 Lines def flush(self) -> None:
) )
if data: if data:
while not self.storage.relation_add(RelationType.REV_IN_ORG, data): while not self.storage.relation_add(RelationType.REV_IN_ORG, data):
LOGGER.warning( LOGGER.warning(
"Unable to write %s rows to the storage. Retrying...", "Unable to write %s rows to the storage. Retrying...",
RelationType.REV_IN_ORG, RelationType.REV_IN_ORG,
) )
# Finally, preferred origins for the visited revisions are set (this step can be
# reordered if required).
origins = {
sha1: self.cache["revision_origin"]["data"][sha1]
for sha1 in self.cache["revision_origin"]["added"]
}
if origins:
while not self.storage.revision_set_origin(origins):
LOGGER.warning(
"Unable to write preferred origins to the storage. Retrying..."
)
# clear local cache ############################################################ # clear local cache ############################################################
self.clear_caches() self.clear_caches()
def content_add_to_directory( def content_add_to_directory(
self, directory: DirectoryEntry, blob: FileEntry, prefix: bytes self, directory: DirectoryEntry, blob: FileEntry, prefix: bytes
) -> None: ) -> None:
self.cache["content_in_directory"].add( self.cache["content_in_directory"].add(
(blob.id, directory.id, normalize(os.path.join(prefix, blob.name))) (blob.id, directory.id, normalize(os.path.join(prefix, blob.name)))
▲ Show 20 Lines • Show All 129 Lines • Show Last 20 Lines
About · Developer information · Legal