Differential D8740 Diff 31544 swh/storage/interface.py

Changeset View

Standalone View

swh/storage/interface.py

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

import datetime		import datetime
from enum import Enum		from enum import Enum
from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple, TypeVar		from typing import (
		TYPE_CHECKING,
		Any,
		Dict,
		Iterable,
		List,
		Optional,
		Sequence,
		Tuple,
		TypeVar,
		)

import attr		import attr
from typing_extensions import Protocol, TypedDict, runtime_checkable		from typing_extensions import Protocol, TypedDict, runtime_checkable

from swh.core.api import remote_api_endpoint		from swh.core.api import remote_api_endpoint
from swh.core.api.classes import PagedResult as CorePagedResult		from swh.core.api.classes import PagedResult as CorePagedResult
from swh.model.model import (		from swh.model.model import (
Content,		Content,
Show All 12 Lines	from swh.model.model import (
Sha1,		Sha1,
Sha1Git,		Sha1Git,
SkippedContent,		SkippedContent,
Snapshot,		Snapshot,
SnapshotBranch,		SnapshotBranch,
)		)
from swh.model.swhids import ExtendedSWHID, ObjectType		from swh.model.swhids import ExtendedSWHID, ObjectType

		if TYPE_CHECKING:
		from swh.storage.writer import JournalWriter


class ListOrder(Enum):		class ListOrder(Enum):
"""Specifies the order for paginated endpoints returning sorted results."""		"""Specifies the order for paginated endpoints returning sorted results."""

ASC = "asc"		ASC = "asc"
DESC = "desc"		DESC = "desc"


Show All 26 Lines

def deprecated(f):		def deprecated(f):
f.deprecated_endpoint = True		f.deprecated_endpoint = True
return f		return f


@runtime_checkable		@runtime_checkable
class StorageInterface(Protocol):		class StorageInterface(Protocol):
		journal_writer: Optional["JournalWriter"] = None

@remote_api_endpoint("check_config")		@remote_api_endpoint("check_config")
def check_config(self, *, check_write: bool) -> bool:		def check_config(self, *, check_write: bool) -> bool:
"""Check that the storage is configured and ready to go."""		"""Check that the storage is configured and ready to go."""
...		...

@remote_api_endpoint("content/add")		@remote_api_endpoint("content/add")
def content_add(self, content: List[Content]) -> Dict[str, int]:		def content_add(self, content: List[Content]) -> Dict[str, int]:
"""Add content blobs to the storage		"""Add content blobs to the storage
▲ Show 20 Lines • Show All 1,320 Lines • Show Last 20 Lines