Page MenuHomeSoftware Heritage
Differential D3383 Diff 12007 swh/storage/interface.py

Changeset View

Standalone View

View Options

swh/storage/interface.py

Show First 20 Lines • Show All 1,170 Lines • ▼ Show 20 Lines ) -> Dict[str, Any]:
'format': '...', 'format': '...',
'metadata': b'...', 'metadata': b'...',
'context': { ... }, 'context': { ... },
} }
""" """
... ...
@remote_api_endpoint("directory/metadata/add")
def directory_metadata_add(
self,
id: str,
context: Dict[str, Union[str, bytes, int]],
discovery_date: datetime.datetime,
authority: Dict[str, Any],
fetcher: Dict[str, Any],
format: str,
metadata: bytes,
) -> None:
"""Add a directory_metadata for the directory at discovery_date,
obtained using the `fetcher` from the `authority`.
The authority and fetcher must be known to the storage before
using this endpoint.
If there is already directory metadata for the same directory, authority,
fetcher, and at the same date; the new one will be either dropped or
will replace the existing one
(it is unspecified which one of these two behaviors happens).
Args:
discovery_date: when the metadata was fetched.
authority: a dict containing keys `type` and `url`.
fetcher: a dict containing keys `name` and `version`.
format: text field indicating the format of the directory of the
metadata: blob of raw metadata
"""
...
@remote_api_endpoint("directory/metadata/get")
def directory_metadata_get(
self,
id: str,
authority: Dict[str, str],
after: Optional[datetime.datetime] = None,
page_token: Optional[bytes] = None,
limit: int = 1000,
) -> Dict[str, Any]:
"""Retrieve list of all directory_metadata entries for the id
Args:
id: the directory's SWHID
authority: a dict containing keys `type` and `url`.
after: minimum discovery_date for a result to be returned
page_token: opaque token, used to get the next page of results
limit: maximum number of results to be returned
Returns:
dict with keys `next_page_token` and `results`.
`next_page_token` is an opaque token that is used to get the
next page of results, or `None` if there are no more results.
`results` is a list of dicts in the format:
.. code-block: python
{
'authority': {'type': ..., 'url': ...},
'fetcher': {'name': ..., 'version': ...},
'discovery_date': ...,
'format': '...',
'metadata': b'...',
'context': { ... },
}
"""
...
@remote_api_endpoint("revision/metadata/add")
def revision_metadata_add(
self,
id: str,
context: Dict[str, Union[str, bytes, int]],
discovery_date: datetime.datetime,
authority: Dict[str, Any],
fetcher: Dict[str, Any],
format: str,
metadata: bytes,
) -> None:
"""Add a revision_metadata for the revision at discovery_date,
obtained using the `fetcher` from the `authority`.
The authority and fetcher must be known to the storage before
using this endpoint.
If there is already revision metadata for the same revision, authority,
fetcher, and at the same date; the new one will be either dropped or
will replace the existing one
(it is unspecified which one of these two behaviors happens).
Args:
discovery_date: when the metadata was fetched.
authority: a dict containing keys `type` and `url`.
fetcher: a dict containing keys `name` and `version`.
format: text field indicating the format of the revision of the
metadata: blob of raw metadata
"""
...
@remote_api_endpoint("revision/metadata/get")
def revision_metadata_get(
self,
id: str,
authority: Dict[str, str],
after: Optional[datetime.datetime] = None,
page_token: Optional[bytes] = None,
limit: int = 1000,
) -> Dict[str, Any]:
"""Retrieve list of all revision_metadata entries for the id
Args:
id: the revision's SWHID
authority: a dict containing keys `type` and `url`.
after: minimum discovery_date for a result to be returned
page_token: opaque token, used to get the next page of results
limit: maximum number of results to be returned
Returns:
dict with keys `next_page_token` and `results`.
`next_page_token` is an opaque token that is used to get the
next page of results, or `None` if there are no more results.
`results` is a list of dicts in the format:
.. code-block: python
{
'authority': {'type': ..., 'url': ...},
'fetcher': {'name': ..., 'version': ...},
'discovery_date': ...,
'format': '...',
'metadata': b'...',
'context': { ... },
}
"""
...
@remote_api_endpoint("release/metadata/add")
def release_metadata_add(
self,
id: str,
context: Dict[str, Union[str, bytes, int]],
discovery_date: datetime.datetime,
authority: Dict[str, Any],
fetcher: Dict[str, Any],
format: str,
metadata: bytes,
) -> None:
"""Add a release_metadata for the release at discovery_date,
obtained using the `fetcher` from the `authority`.
The authority and fetcher must be known to the storage before
using this endpoint.
If there is already release metadata for the same release, authority,
fetcher, and at the same date; the new one will be either dropped or
will replace the existing one
(it is unspecified which one of these two behaviors happens).
Args:
discovery_date: when the metadata was fetched.
authority: a dict containing keys `type` and `url`.
fetcher: a dict containing keys `name` and `version`.
format: text field indicating the format of the release of the
metadata: blob of raw metadata
"""
...
@remote_api_endpoint("release/metadata/get")
def release_metadata_get(
self,
id: str,
authority: Dict[str, str],
after: Optional[datetime.datetime] = None,
page_token: Optional[bytes] = None,
limit: int = 1000,
) -> Dict[str, Any]:
"""Retrieve list of all release_metadata entries for the id
Args:
id: the release's SWHID
authority: a dict containing keys `type` and `url`.
after: minimum discovery_date for a result to be returned
page_token: opaque token, used to get the next page of results
limit: maximum number of results to be returned
Returns:
dict with keys `next_page_token` and `results`.
`next_page_token` is an opaque token that is used to get the
next page of results, or `None` if there are no more results.
`results` is a list of dicts in the format:
.. code-block: python
{
'authority': {'type': ..., 'url': ...},
'fetcher': {'name': ..., 'version': ...},
'discovery_date': ...,
'format': '...',
'metadata': b'...',
'context': { ... },
}
"""
...
@remote_api_endpoint("snapshot/metadata/add")
def snapshot_metadata_add(
self,
id: str,
context: Dict[str, Union[str, bytes, int]],
discovery_date: datetime.datetime,
authority: Dict[str, Any],
fetcher: Dict[str, Any],
format: str,
metadata: bytes,
) -> None:
"""Add a snapshot_metadata for the snapshot at discovery_date,
obtained using the `fetcher` from the `authority`.
The authority and fetcher must be known to the storage before
using this endpoint.
If there is already snapshot metadata for the same snapshot, authority,
fetcher, and at the same date; the new one will be either dropped or
will replace the existing one
(it is unspecified which one of these two behaviors happens).
Args:
discovery_date: when the metadata was fetched.
authority: a dict containing keys `type` and `url`.
fetcher: a dict containing keys `name` and `version`.
format: text field indicating the format of the snapshot of the
metadata: blob of raw metadata
"""
...
@remote_api_endpoint("snapshot/metadata/get")
def snapshot_metadata_get(
self,
id: str,
authority: Dict[str, str],
after: Optional[datetime.datetime] = None,
page_token: Optional[bytes] = None,
limit: int = 1000,
) -> Dict[str, Any]:
"""Retrieve list of all snapshot_metadata entries for the id
Args:
id: the snapshot's SWHID
authority: a dict containing keys `type` and `url`.
after: minimum discovery_date for a result to be returned
page_token: opaque token, used to get the next page of results
limit: maximum number of results to be returned
Returns:
dict with keys `next_page_token` and `results`.
`next_page_token` is an opaque token that is used to get the
next page of results, or `None` if there are no more results.
`results` is a list of dicts in the format:
.. code-block: python
{
'authority': {'type': ..., 'url': ...},
'fetcher': {'name': ..., 'version': ...},
'discovery_date': ...,
'format': '...',
'metadata': b'...',
'context': { ... },
}
"""
...
@remote_api_endpoint("origin/metadata/add") @remote_api_endpoint("origin/metadata/add")
def origin_metadata_add( def origin_metadata_add(
self, self,
origin_url: str, origin_url: str,
discovery_date: datetime.datetime, discovery_date: datetime.datetime,
authority: Dict[str, Any], authority: Dict[str, Any],
fetcher: Dict[str, Any], fetcher: Dict[str, Any],
format: str, format: str,
▲ Show 20 Lines • Show All 188 Lines • Show Last 20 Lines
About · Developer information · Legal