Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions _quarto.yml
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,7 @@ quartodoc:
- title: "CLI"
contents:
- zenodo_publish
- get
- title: "Zenodo API"
contents:
- ZenodoClient
Expand Down
25 changes: 1 addition & 24 deletions docs/design/interface/cli.qmd
Original file line number Diff line number Diff line change
Expand Up @@ -68,7 +68,7 @@ def list(sandbox: bool = False) -> None:
pretty_print(deposits)
```

## {{< var planned >}} `get`
## {{< var done >}} `get`

``` {.bash filename="Terminal"}
zen-do get [METADATA_FILE] --sandbox
Expand All @@ -83,29 +83,6 @@ uses the Python functions within `list()` to get the list of deposits
and uses that to find the deposit that matches the content of the
metadata file using the URN ID.

The Python interface, with some implementation comments and docstring,
will be:

```python
@app.command()
def get(metadata_file: Path = Path(".zenodo.toml"), \, *, sandbox: bool = False) -> None:
"""Get the Zenodo deposit JSON based on the metadata file.

Args:
metadata_file: The path to the metadata file.
sandbox: Whether to use the Zenodo sandbox environment for testing purposes.
"""
# Read metadata file (checking if the URN exists or not).
# metadata: Metadata = read_metadata(path=metadata_file)
# Fetch token from system keyring or environment variable.
# token: Token = get_token(sandbox=sandbox)
# Get request to Zenodo servers.
# TODO: Either dict or a custom class for the deposit, e.g. `Deposit`?
# deposits: list[Deposit] = list_deposits(sandbox=sandbox, token=token)
# deposit: Deposit = find_deposit(deposits=deposits, metadata=metadata)
pretty_print(deposit)
```

## {{< var planned >}} `convert`

``` {.bash filename="Terminal"}
Expand Down
1 change: 1 addition & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@ dependencies = [
"keyring>=25.7.0",
"pydantic>=2.12.5",
"requests>=2.32.5",
"rich>=15.0.0",
"seedcase-soil>=0.10.0",
"toml>=0.10.2",
]
Expand Down
3 changes: 2 additions & 1 deletion src/zen_do/__init__.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
"""Module containing all source code."""

from .cli import zenodo_publish
from .cli import get, zenodo_publish
from .examples import example_deposit, example_metadata
from .zenodo_client import (
ZenodoClient,
Expand All @@ -9,6 +9,7 @@
from .zenodo_metadata import ZenodoCreator, ZenodoMetadata, ZenodoRelatedIdentifier

__all__ = [
"get",
"example_metadata",
"example_deposit",
"zenodo_publish",
Expand Down
26 changes: 26 additions & 0 deletions src/zen_do/cli.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,7 @@
from pathlib import Path

import seedcase_soil as so
from rich import print_json
from seedcase_soil import (
# print_if_verbose,
run_without_tracebacks,
Expand Down Expand Up @@ -26,6 +30,28 @@ def zenodo_publish(sandbox: bool = False) -> None:
print("New Zenodo record created successfully!")


@app.command()
def get(
metadata_file: Path = Path(".zenodo.toml"), /, *, sandbox: bool = False
) -> None:
"""Get the Zenodo deposit JSON based on the metadata file.

Args:
metadata_file: The path to the metadata file.
sandbox: Whether to use the Zenodo sandbox environment for testing purposes.
"""
token = get_token(sandbox)
client = ZenodoClient(token, sandbox)
deposit = zenodo_get_deposit(client.get_deposits(), metadata_file)

if deposit:
print_json(data=deposit)
else:
so.pretty_print(
f"No deposit found on Zenodo for metadata file '{metadata_file}'."
)


def main() -> None:
"""Create an entry point to run the cli without tracebacks."""
run_without_tracebacks(app)
6 changes: 4 additions & 2 deletions src/zen_do/examples.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,9 @@
)


def example_metadata(title: str = "Test Book") -> ZenodoMetadata:
def example_metadata(
title: str = "Test Book", urn: str = "urn:zenodo:my-org:project:book"
) -> ZenodoMetadata:
"""A set of example Zenodo metadata."""
return ZenodoMetadata(
title=title,
Expand All @@ -23,7 +25,7 @@ def example_metadata(title: str = "Test Book") -> ZenodoMetadata:
],
related_identifiers=[
ZenodoRelatedIdentifier(
identifier="urn:zenodo:my-org:project:book",
identifier=urn,
relation="isIdenticalTo",
resource_type="other",
scheme="urn",
Expand Down
30 changes: 8 additions & 22 deletions src/zen_do/zenodo_get_deposit.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,22 +5,25 @@
import seedcase_soil as so

from zen_do.zenodo_client import ZenodoResponse, _get_zenodo_field
from zen_do.zenodo_metadata import ZenodoMetadata, ZenodoRelatedIdentifier
from zen_do.zenodo_metadata import ZenodoMetadata, ZenodoRelatedIdentifier, _is_urn


def zenodo_get_deposit(deposits: list[ZenodoResponse]) -> Optional[ZenodoResponse]:
def zenodo_get_deposit(
deposits: list[ZenodoResponse], metadata_file: Path = Path(".zenodo.toml")
) -> Optional[ZenodoResponse]:
"""Gets the Zenodo deposit for the repository if it exists.

Gets the URN identifier from the `.zenodo.toml` file. If one
doesn't exist, this function will not work.

Args:
deposits: All the deposits on Zenodo associated with an access token.
metadata_file: The path to the metadata file.

Returns:
The Zenodo deposit for the repo if it exists, None otherwise.
"""
urn = _get_urn()
urn = _load_zenodo_toml(metadata_file).urn

matching_deposits = so.keep(
deposits,
Expand All @@ -47,25 +50,8 @@ def _urn_matches(id_response: ZenodoResponse, target_urn: str) -> bool:
return _is_urn(id) and id.identifier == target_urn


def _is_urn(id: ZenodoRelatedIdentifier) -> bool:
return id.relation == "isIdenticalTo" and id.scheme == "urn"


def _get_urn() -> str:
metadata = _load_zenodo_toml()
ids = so.keep(metadata.related_identifiers, _is_urn)
if len(ids) != 1:
raise ValueError(
"Expected exactly one `isIdenticalTo` URN in `.zenodo.toml` under "
f"`related_identifiers`, but found {len(ids)}. Ensure there is a single "
"unique URN, as it is used to identify the corresponding deposit on Zenodo."
)

return ids[0].identifier


def _load_zenodo_toml() -> ZenodoMetadata:
with open(Path(".zenodo.toml"), mode="rb") as file:
def _load_zenodo_toml(metadata_file: Path) -> ZenodoMetadata:
with open(metadata_file, mode="rb") as file:
toml_file = tomllib.load(file)

return ZenodoMetadata.model_validate(toml_file)
23 changes: 23 additions & 0 deletions src/zen_do/zenodo_metadata.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
import re
from typing import Optional, Self

import seedcase_soil as so
from pydantic import BaseModel, ConfigDict, model_validator


Expand Down Expand Up @@ -71,3 +72,25 @@ class ZenodoMetadata(KebabModel, frozen=True):
upload_type: str
creators: list[ZenodoCreator]
related_identifiers: list[ZenodoRelatedIdentifier] = []

@property
def urn(self) -> str:
"""The URN related identifier of the deposit."""
urns = so.keep(self.related_identifiers, _is_urn)
return urns[0].identifier

@model_validator(mode="after")
def _check_unique_urn(self) -> Self:
urns = so.keep(self.related_identifiers, _is_urn)
if len(urns) != 1:
raise ValueError(
"Expected exactly one `isIdenticalTo` URN in the Zenodo metadata file "
f"under `related_identifiers`, but found {len(urns)}. Ensure there is "
"a single unique URN, as it is used to identify the corresponding "
"deposit on Zenodo."
)
return self


def _is_urn(id: ZenodoRelatedIdentifier) -> bool:
return id.relation == "isIdenticalTo" and id.scheme == "urn"
31 changes: 30 additions & 1 deletion tests/test_cli.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
from pytest import fixture, raises

from zen_do.cli import app
from zen_do.examples import example_deposit


@fixture
Expand All @@ -9,7 +10,8 @@ def _mock_zenodo_get_deposit(mocker):


@fixture
def _mock_client(mocker):
def _mock_client(mocker, monkeypatch):
monkeypatch.setenv("ZENODO_TOKEN", "token")
return mocker.patch("zen_do.cli.ZenodoClient")


Expand All @@ -35,3 +37,30 @@ def test_zenodo_publish_new_deposit(
def test_zenodo_publish_needs_token():
with raises(RuntimeError):
app("zenodo-publish", result_action="return_value")


def test_get_when_deposit_found(
capsys,
_mock_client,
_mock_zenodo_get_deposit,
):
deposit = example_deposit()
_mock_zenodo_get_deposit.return_value = deposit

app("get", result_action="return_value")
out = capsys.readouterr().out

assert str(deposit["id"]) in out


def test_get_when_deposit_not_found(
capsys,
_mock_client,
_mock_zenodo_get_deposit,
):
_mock_zenodo_get_deposit.return_value = None

app("get", result_action="return_value")
out = capsys.readouterr().out

assert "{" not in out
81 changes: 23 additions & 58 deletions tests/test_zenodo_get_deposit.py
Original file line number Diff line number Diff line change
@@ -1,21 +1,18 @@
from pathlib import Path

import pydantic
import toml
from pytest import MonkeyPatch, fixture, mark, raises
from pytest import MonkeyPatch, fixture, raises

from zen_do.examples import example_deposit, example_metadata
from zen_do.zenodo_get_deposit import zenodo_get_deposit
from zen_do.zenodo_metadata import ZenodoMetadata, ZenodoRelatedIdentifier


def _write_metadata(path: Path, metadata: ZenodoMetadata) -> None:
(path / ".zenodo.toml").write_text(toml.dumps(metadata.model_dump()))
from zen_do.zenodo_metadata import ZenodoRelatedIdentifier


@fixture
def _zenodo_toml(monkeypatch: MonkeyPatch, tmp_path: Path) -> None:
monkeypatch.chdir(tmp_path)
_write_metadata(tmp_path, example_metadata())
(tmp_path / ".zenodo.toml").write_text(toml.dumps(example_metadata().model_dump()))


def test_returns_deposit_if_matching_deposit_has_exactly_one_matching_identifier(
Expand All @@ -32,6 +29,17 @@ def test_returns_deposit_if_matching_deposit_has_exactly_one_matching_identifier
assert deposit["id"] == 12


def test_can_use_non_default_file_location(tmp_path):
(tmp_path / "book").mkdir()
metadata_path = tmp_path / "book" / ".book.zenodo.toml"
metadata_path.write_text(toml.dumps(example_metadata().model_dump()))

deposit = zenodo_get_deposit([example_deposit(id=12)], metadata_path)

assert deposit
assert deposit["id"] == 12


def test_returns_deposit_if_matching_deposit_has_at_least_one_matching_identifier(
_zenodo_toml,
):
Expand Down Expand Up @@ -69,6 +77,14 @@ def test_raises_error_if_multiple_matching_deposits(_zenodo_toml):
zenodo_get_deposit(deposits)


def test_raises_error_if_metadata_file_empty(tmp_path):
metadata_path = tmp_path / ".zenodo.toml"
metadata_path.touch()

with raises(pydantic.ValidationError):
zenodo_get_deposit([example_deposit()], metadata_path)


def test_returns_none_if_no_deposits_on_zenodo(_zenodo_toml):
deposit = zenodo_get_deposit([])

Expand All @@ -81,54 +97,3 @@ def test_returns_none_if_no_matching_deposits(_zenodo_toml):
deposit = zenodo_get_deposit(deposits)

assert deposit is None


def test_raises_error_if_zenodo_json_has_no_urn_id(monkeypatch, tmp_path):
metadata = example_metadata()
monkeypatch.chdir(tmp_path)
del metadata.related_identifiers[0]
_write_metadata(tmp_path, metadata)

with raises(ValueError):
zenodo_get_deposit([])


def test_raises_error_if_zenodo_json_has_multiple_urn_ids(monkeypatch, tmp_path):
metadata = example_metadata()
monkeypatch.chdir(tmp_path)
metadata.related_identifiers.append(
ZenodoRelatedIdentifier(
identifier="urn:zenodo:my-org:project:poster",
relation="isIdenticalTo",
resource_type="other",
scheme="urn",
)
)
_write_metadata(tmp_path, metadata)

with raises(ValueError):
zenodo_get_deposit([])


@mark.parametrize(
"urn",
[
"",
"not a URN",
"urn",
"urn:",
"urn:unknown",
"urn:zenodo",
"urn:zenodo:",
"urn:zenodo:a:",
"urn:zenodo:a/b",
],
)
def test_flags_incorrect_urn(monkeypatch, tmp_path, urn):
metadata_json = example_metadata().model_dump()
metadata_json["related_identifiers"][0]["identifier"] = urn
monkeypatch.chdir(tmp_path)
(tmp_path / ".zenodo.toml").write_text(toml.dumps(metadata_json))

with raises(ValueError):
zenodo_get_deposit([])
Loading
Loading