improve cli

docs/conf.py

@@ -1,6 +1,7 @@
 # -*- coding: utf-8 -*-
 #
 # Copyright (C) 2020-2022 Technische Universität Graz.
+# Copyright (C) 2024 Graz University of Technology.
 #
 # invenio-pure is free software; you can redistribute it and/or modify it
 # under the terms of the MIT License; see LICENSE file for more details.
@@ -30,6 +31,11 @@
     "sphinx.ext.viewcode",
 ]
 
+# add nitpick ignore
+nitpick_ignore = [
+    ("py:class", "flask.app.Flask"),
+]
+
 # Add any paths that contain templates here, relative to this directory.
 templates_path = ["_templates"]
 

invenio_pure/__init__.py

@@ -1,6 +1,6 @@
 # -*- coding: utf-8 -*-
 #
-# Copyright (C) 2021 Technische Universität Graz.
+# Copyright (C) 2021-2024 Graz University of Technology.
 #
 # invenio-pure is free software; you can redistribute it and/or modify it
 # under the terms of the MIT License; see LICENSE file for more details.
@@ -9,15 +9,13 @@
 
 from .errors import PureAPIError, PureRuntimeError
 from .ext import InvenioPure
-from .types import URL, PureConfigs, PureID, PureRecord
+from .types import URL, PureID
 
 __version__ = "0.1.2"
 
 __all__ = (
     "__version__",
     "InvenioPure",
-    "PureConfigs",
-    "PureRecord",
     "PureID",
     "PureAPIError",
     "PureRuntimeError",

invenio_pure/cli.py

@@ -1,45 +1,93 @@
 # -*- coding: utf-8 -*-
 #
-# Copyright (C) 2021-2022 Graz University of Technology.
+# Copyright (C) 2021-2025 Graz University of Technology.
 #
 # invenio-pure is free software; you can redistribute it and/or modify it
 # under the terms of the MIT License; see LICENSE file for more details.
 
 """CLI commands for Invenio-RDM-Pure."""
 
 
-from click import STRING, group, option
+from click import STRING, group, option, secho
 from click_params import URL
 from flask import current_app
 from flask.cli import with_appcontext
+from invenio_access.utils import get_identity
+from invenio_accounts import current_accounts
 
-from .api import import_from_pure
-from .click_options import JSON
-from .types import PureConfigs
+from .services import PureRESTService, build_service
 
 
 @group()
-def pure():
+def pure() -> None:
     """Commands for InvenioRdmPure."""
 
 
 @pure.command("import")
 @with_appcontext
-@option("--endpoint", type=URL)
-@option("--token", type=STRING)
-@option("--pure_username", type=STRING)
-@option("--pure_password", type=STRING)
-@option("--user-email", type=STRING)
-@option("--pure-record", type=JSON())
+@option("--endpoint", type=URL, required=True)
+@option("--token", type=STRING, required=True)
+@option("--user-email", type=STRING, required=True)
+@option("--pure-id", type=STRING, required=True)
+@option("--no-color", is_flag=True, default=False)
+@build_service
 def import_records_from_pure(
-    endpoint, token, pure_username, pure_password, user_email, pure_record
-):
+    pure_service: PureRESTService,
+    user_email: str,
+    pure_id: str,
+    *,
+    no_color: bool,
+) -> None:
     """Import a record from given JSON file or JSON string."""
     import_func = current_app.config["PURE_IMPORT_FUNC"]
-    recipients = current_app.config["PURE_ERROR_MAIL_RECIPIENTS"]
-    sender = current_app.config["PURE_ERROR_MAIL_SENDER"]
-    configs = PureConfigs(
-        endpoint, token, pure_username, pure_password, user_email, recipients, sender
-    )
-    record = import_from_pure(import_func, pure_record, configs)
-    print(f"record.id: {record.id}")
+    user = current_accounts.datastore.get_user_by_email(user_email)
+    identity = get_identity(user)
+    try:
+        record = import_func(identity, pure_id, pure_service)
+        color = "green" if not no_color else "black"
+        secho(f"record.id: {record.id}", fg=color)
+    except RuntimeError as e:
+        msg = f"ERROR pure pure_id: {pure_id} with message: {e!r}"
+        color = "red" if not no_color else "black"
+        secho(msg, fg=color)
+
+
+@pure.command("list")
+@with_appcontext
+@option("--endpoint", type=URL, required=True)
+@option("--token", type=STRING, required=True)
+@option("--user-email", type=STRING, required=True)
+@build_service
+def list_all_available_records(pure_service: PureRESTService, user_email: str) -> None:
+    """List all possible to import records."""
+    filter_records = current_app.config["PURE_FILTER_RECORDS"]
+    user = current_accounts.datastore.get_user_by_email(user_email)
+    identity = get_identity(user)
+
+    for pure_id in pure_service.fetch_all_ids(identity, filter_records):
+        secho(f"pure_id: {pure_id}", fg="green")
+
+
+@pure.command("sync")
+@with_appcontext
+@option("--endpoint", type=URL, required=True)
+@option("--token", type=STRING, required=True)
+@option("--user-email", type=STRING, required=True)
+@build_service
+def sync(pure_service: PureRESTService, user_email: str) -> None:
+    """Sync Pure with the repo."""
+    import_func = current_app.config["PURE_IMPORT_FUNC"]
+    filter_records = current_app.config["PURE_FILTER_RECORDS"]
+    user = current_accounts.datastore.get_user_by_email(user_email)
+    identity = get_identity(user)
+
+    ids = pure_service.fetch_all_ids(identity, filter_records)
+
+    for pure_id in ids:
+        try:
+            import_func(identity, pure_id, pure_service)
+            msg = f"SUCCESS pure pure_id: {pure_id} imported successfully."
+            secho(msg, fg="green")
+        except RuntimeError as e:
+            msg = f"ERROR pure pure_id: {pure_id} with message: {e!r}"
+            secho(msg, fg="red")

invenio_pure/config.py

@@ -1,26 +1,16 @@
 # -*- coding: utf-8 -*-
 #
-# Copyright (C) 2021-2022 Technische Universität Graz.
+# Copyright (C) 2021-2025 Graz University of Technology.
 #
 # invenio-pure is free software; you can redistribute it and/or modify it
 # under the terms of the MIT License; see LICENSE file for more details.
 
 """Invenio module that adds pure."""
-from typing import Callable
 
-from .types import (
-    URL,
-    EmailAddress,
-    FilePath,
-    PureConfigs,
-    PureID,
-    PurePassword,
-    PureRecord,
-    PureToken,
-    PureUsername,
-)
 
-PURE_CELERY_BEAT_SCHEDULE = {}
+from .types import URL, EmailAddress, PureToken
+
+PURE_CELERY_BEAT_SCHEDULE: dict[str, dict] = {}
 """The celery beat schedule is used to configure the import schedule.
 
 Following is a example configuration. One option to add this to the
@@ -35,44 +25,11 @@
     }
 """
 
-PURE_IMPORT_FUNC: Callable[
-    [
-        PureRecord,
-        PureConfigs,
-        Callable[[PureID, URL, PureUsername, PurePassword], FilePath],
-    ],
-    None,
-] = None
-"""This function is called to import the pure record into the repository.
-
-It needs as an parameter the pure record as a json, the pure configuration
-and a callable which is provided from this package to download the file
-from pure."""
-
-PURE_SIEVE_FUNC: Callable[[PureRecord], bool] = None
-"""This function implements the import criteria.
-
-The import criteria handles the conditions which have to be true
-as that the record will be imported into the repository.
-"""
-
 PURE_PURE_ENDPOINT: URL = ""
 """This is the endpoint of the pure instance."""
 
 PURE_PURE_TOKEN: PureToken = ""
 """This is the token to be allowed to use the API."""
 
-PURE_PURE_USERNAME: PureUsername = ""
-"""This is the pure username which is necessary to download files."""
-
-PURE_PURE_PASSWORD: PurePassword = ""
-"""This is the pure password which is necessary to download files."""
-
 PURE_USER_EMAIL: EmailAddress = ""
 """This is the user email of the pure user within the InvenioRDM instance."""
-
-PURE_ERROR_MAIL_RECIPIENTS: list[EmailAddress] = []
-"""The list of recipients to send emails if errors happen in the import process."""
-
-PURE_ERROR_MAIL_SENDER: EmailAddress = ""
-"""The sender of the error emails."""