Differential D4118 Diff 14561 swh/indexer/storage/model.py

Changeset View

Standalone View

swh/indexer/storage/model.py

# Copyright (C) 2020 The Software Heritage developers		# Copyright (C) 2020 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

"""Classes used internally by the in-memory idx-storage, and will be		"""Classes used internally by the in-memory idx-storage, and will be
used for the interface of the idx-storage in the near future."""		used for the interface of the idx-storage in the near future."""

from __future__ import annotations		from __future__ import annotations

from typing import Any, Dict, List, Optional, Type, TypeVar		from typing import Any, Dict, List, Optional, Tuple, Type, TypeVar

import attr		import attr
from typing_extensions import TypedDict

from swh.model.model import Sha1Git, dictify		from swh.model.model import Sha1Git, dictify


class KeyDict(TypedDict):
id: Sha1Git
indexer_configuration_id: int


TSelf = TypeVar("TSelf")		TSelf = TypeVar("TSelf")


@attr.s		@attr.s
class BaseRow:		class BaseRow:
		UNIQUE_KEY_FIELDS: Tuple = ("id", "indexer_configuration_id")

id = attr.ib(type=Sha1Git)		id = attr.ib(type=Sha1Git)
indexer_configuration_id = attr.ib(type=Optional[int], default=None, kw_only=True)		indexer_configuration_id = attr.ib(type=Optional[int], default=None, kw_only=True)
tool = attr.ib(type=Optional[Dict], default=None, kw_only=True)		tool = attr.ib(type=Optional[Dict], default=None, kw_only=True)

def __attrs_post_init__(self):		def __attrs_post_init__(self):
if self.indexer_configuration_id is None and self.tool is None:		if self.indexer_configuration_id is None and self.tool is None:
raise TypeError("Either indexer_configuration_id or tool must be not None.")		raise TypeError("Either indexer_configuration_id or tool must be not None.")
if self.indexer_configuration_id is not None and self.tool is not None:		if self.indexer_configuration_id is not None and self.tool is not None:
Show All 16 Lines	def to_dict(self) -> Dict[str, Any]:
del d["tool"]		del d["tool"]

return d		return d

@classmethod		@classmethod
def from_dict(cls: Type[TSelf], d) -> TSelf:		def from_dict(cls: Type[TSelf], d) -> TSelf:
return cls(**d) # type: ignore		return cls(**d) # type: ignore

		def unique_key(self) -> Dict:
		if self.indexer_configuration_id is None:
		raise ValueError(
		"Can only call unique_key() on objects with indexer_configuration_id."
		)
		return {key: getattr(self, key) for key in self.UNIQUE_KEY_FIELDS}


@attr.s		@attr.s
class ContentMimetypeRow(BaseRow):		class ContentMimetypeRow(BaseRow):
mimetype = attr.ib(type=str)		mimetype = attr.ib(type=str)
encoding = attr.ib(type=str)		encoding = attr.ib(type=str)


@attr.s		@attr.s
class ContentLanguageRow(BaseRow):		class ContentLanguageRow(BaseRow):
lang = attr.ib(type=str)		lang = attr.ib(type=str)


@attr.s		@attr.s
class ContentCtagsEntry:		class ContentCtagsRow(BaseRow):
		UNIQUE_KEY_FIELDS = (
		"id",
		"indexer_configuration_id",
		"name",
		"kind",
		"line",
		"lang",
		)

name = attr.ib(type=str)		name = attr.ib(type=str)
kind = attr.ib(type=str)		kind = attr.ib(type=str)
line = attr.ib(type=int)		line = attr.ib(type=int)
lang = attr.ib(type=str)		lang = attr.ib(type=str)

def to_dict(self) -> Dict[str, Any]:
return dictify(attr.asdict(self, recurse=False))

@classmethod
def from_dict(cls, d) -> ContentCtagsEntry:
return cls(**d)


@attr.s		@attr.s
class ContentCtagsRow(BaseRow):		class ContentLicenseRow(BaseRow):
ctags = attr.ib(type=List[ContentCtagsEntry])		UNIQUE_KEY_FIELDS = ("id", "indexer_configuration_id", "license")

@classmethod		license = attr.ib(type=str)
def from_dict(cls, d) -> ContentCtagsRow:
d = d.copy()
items = d.pop("ctags")
return cls(ctags=[ContentCtagsEntry.from_dict(item) for item in items], **d,)


@attr.s
class ContentLicensesRow(BaseRow):
licenses = attr.ib(type=List[str])


@attr.s		@attr.s
class ContentMetadataRow(BaseRow):		class ContentMetadataRow(BaseRow):
metadata = attr.ib(type=Dict[str, Any])		metadata = attr.ib(type=Dict[str, Any])


@attr.s		@attr.s
Show All 10 Lines