setup: migrate to ruff

.ruff.toml

@@ -0,0 +1,4 @@
+select = ["ALL"]
+ignore = ["ANN101", "ANN102", "D203", "D211", "D212", "D213", "UP009", "PLR0913", "SIM117", "INP001", "S101"]
+
+exclude = ["docs"]

invenio_pure/api.py

@@ -1,38 +1,45 @@
 # -*- coding: utf-8 -*-
 #
-# Copyright (C) 2022 Graz University of Technology.
+# Copyright (C) 2022-2023 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.
 
 """API functions of the pure connector."""
 
-from typing import Callable
+
+from collections.abc import Callable
+
+from invenio_records_resources.services.records.results import RecordItem
 
 from .types import PureConfigs, PureRecord
 from .utils import download_file, get_research_output_count, get_research_outputs
 
 
 def import_from_pure(
-    import_func: Callable, pure_record: PureRecord, configs: PureConfigs
-):
+    import_func: Callable,
+    pure_record: PureRecord,
+    configs: PureConfigs,
+) -> RecordItem:
     """Import record from pure."""
     return import_func(pure_record, configs, download_file)
 
 
-def pure_records(configs):
-    """Generator to yield records from pure."""
+def pure_records(configs: dict) -> None:
+    """Yield records from pure."""
     research_count = get_research_output_count(configs.endpoint, configs.token)
     size = 100
     offset = 0
 
     while research_count > 0:
         research_output = get_research_outputs(
-            configs.endpoint, configs.token, size, offset
+            configs.endpoint,
+            configs.token,
+            size,
+            offset,
         )
 
-        for record in research_output:
-            yield record
+        yield from research_output
 
         offset += size
         research_count -= size

invenio_pure/cli.py

@@ -1,14 +1,14 @@
 # -*- coding: utf-8 -*-
 #
-# Copyright (C) 2021-2022 Graz University of Technology.
+# Copyright (C) 2021-2023 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
@@ -19,7 +19,7 @@
 
 
 @group()
-def pure():
+def pure() -> None:
     """Commands for InvenioRdmPure."""
 
 
@@ -31,15 +31,30 @@ def pure():
 @option("--pure_password", type=STRING)
 @option("--user-email", type=STRING)
 @option("--pure-record", type=JSON())
+@option("--no-color", is_flag=True, default=False)
 def import_records_from_pure(
-    endpoint, token, pure_username, pure_password, user_email, pure_record
-):
+    endpoint: str,
+    token: str,
+    pure_username: str,
+    pure_password: str,
+    user_email: str,
+    pure_record: dict,
+    *,
+    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
+        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}")
+    color = "green" if not no_color else "black"
+    secho(f"record.id: {record.id}", fg=color)

invenio_pure/click_options.py

@@ -8,20 +8,21 @@
 """Click options."""
 
 from json import JSONDecodeError, loads
-from os.path import dirname, isabs, isfile, join
+from pathlib import Path
+from typing import Any
 
 from click import ParamType, secho
 
 
-def load_file_as_string(path):
+def load_file_as_string(path: str) -> str:
     """Open a file and return the content as UTF-8 encoded string."""
-    if not isabs(path):
-        path = join(dirname(__file__), path)
+    if not Path(path).is_absolute(path):
+        path = Path(__file__).parent / path
 
-    if not isfile(path):
+    if not Path(path).is_file():
         return ""
 
-    with open(path, "rb") as file_pointer:
+    with Path(path).open(mode="rb") as file_pointer:
         content = file_pointer.read()
         return content.decode("utf-8")
 
@@ -31,9 +32,14 @@ class JSON(ParamType):
 
     name = "JSON"
 
-    def convert(self, value, param, ctx):
-        """This method converts the json-str or json-file to the dictionary representation."""
-        if isfile(value):
+    def convert(
+        self,
+        value: Any,  # noqa: ANN401
+        param: Any,  # noqa: ANN401, ARG002
+        ctx: Any,  # noqa: ANN401, ARG002
+    ) -> str:
+        """Convert the json-str or json-file to the dictionary representation."""
+        if Path(value).is_file():
             value = load_file_as_string(value)
 
         try:

invenio_pure/config.py

@@ -6,7 +6,7 @@
 # under the terms of the MIT License; see LICENSE file for more details.
 
 """Invenio module that adds pure."""
-from typing import Callable
+from collections.abc import Callable
 
 from .types import (
     URL,

invenio_pure/errors.py

@@ -1,24 +1,26 @@
 # -*- coding: utf-8 -*-
 #
-# Copyright (C) 2022 Technische Universität Graz
+# Copyright (C) 2022-2023 Technische Universität Graz
 #
 # 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.
 
 """REST errors."""
 
+from .types import PureRecord
+
 
 class PureRuntimeError(RuntimeError):
     """Pure runtime error."""
 
-    def __init__(self, record):
-        """Constructor for pure runtime error."""
+    def __init__(self, record: PureRecord) -> None:
+        """Construct for pure runtime error."""
         self.record = record
 
 
 class PureAPIError(Exception):
     """Pure API error class."""
 
-    def __init__(self, code, msg):
-        """Constructor for pure api error."""
+    def __init__(self, code: int, msg: str) -> None:
+        """Construct for pure api error."""
         super().__init__(f"Pure API error code={code} msg='{msg}'")

invenio_pure/ext.py

@@ -1,21 +1,23 @@
 # -*- coding: utf-8 -*-
 #
-# Copyright (C) 2021-2022 Technische Universität Graz.
+# Copyright (C) 2021-2023 Technische Universität Graz.
 #
 # 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 flask import Flask
+
 
 class InvenioPure:
     """invenio-pure extension."""
 
-    def __init__(self, app=None):
+    def __init__(self, app: Flask = None) -> None:
         """Extension initialization."""
         if app:
             self.init_app(app)
 
-    def init_app(self, app):
+    def init_app(self, app: Flask) -> None:
         """Flask application initialization."""
         app.extensions["invenio-pure"] = self

invenio_pure/tasks.py

@@ -1,13 +1,13 @@
 # -*- coding: utf-8 -*-
 #
-# Copyright (C) 2021-2022 Technische Universität Graz
+# Copyright (C) 2021-2023 Technische Universität Graz
 #
 # 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.
 
 """Scheduled tasks for celery."""
 
-from typing import Callable
+from collections.abc import Callable
 
 from celery import shared_task
 from flask import current_app
@@ -18,7 +18,7 @@
 
 
 def config_variables() -> tuple[Callable, Callable, PureConfigs]:
-    """Configuration variables."""
+    """Collect variables."""
     import_func = current_app.config["PURE_IMPORT_FUNC"]
     sieve_func = current_app.config["PURE_SIEVE_FUNC"]
     endpoint = current_app.config["PURE_PURE_ENDPOINT"]
@@ -30,14 +30,20 @@ def config_variables() -> tuple[Callable, Callable, PureConfigs]:
     sender = current_app.config["PURE_ERROR_MAIL_SENDER"]
 
     pure_configs = PureConfigs(
-        endpoint, token, pure_username, pure_password, user_email, recipients, sender
+        endpoint,
+        token,
+        pure_username,
+        pure_password,
+        user_email,
+        recipients,
+        sender,
     )
 
     return import_func, sieve_func, pure_configs
 
 
 @shared_task(ignore_result=True)
-def import_records_from_pure():
+def import_records_from_pure() -> None:
     """Import records from pure."""
     import_func, sieve_func, configs = config_variables()
 

invenio_pure/utils.py

@@ -1,13 +1,14 @@
 # -*- coding: utf-8 -*-
 #
-# Copyright (C) 2021-2022 Technische Universität Graz
+# Copyright (C) 2021-2023 Technische Universität Graz
 #
 # 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.
 
 """Utility methods."""
-
+from http import HTTPStatus
 from json import loads
+from pathlib import Path
 from shutil import copyfileobj
 
 from requests import get
@@ -32,14 +33,17 @@ def get_research_output_count(pure_api_key: str, pure_api_url: str) -> int:
     url = f"{pure_api_url}/research-outputs"
     response = get(url, headers=headers(pure_api_key), timeout=10)
 
-    if response.status_code != 200:
+    if response.status_code != HTTPStatus.OK:
         return -1
 
     return int(loads(response.text)["count"])
 
 
 def get_research_outputs(
-    pure_api_key: str, pure_api_url: str, size: int, offset: int
+    pure_api_key: str,
+    pure_api_url: str,
+    size: int,
+    offset: int,
 ) -> list[dict]:
     """Get a list of research outputs.
 
@@ -51,18 +55,21 @@ def get_research_outputs(
     url = f"{pure_api_url}/research-outputs?size={size}&offset={offset}"
     response = get(url, headers=headers(pure_api_key), timeout=10)
 
-    if response.status_code != 200:
+    if response.status_code != HTTPStatus.OK:
         return []
 
     response_json = loads(response.text)
-    items = response_json["items"]
-    return items
+    return response_json["items"]
 
 
-def store_file_temporarily(file_url: URL, file_path: FilePath, auth: HTTPBasicAuth):
+def store_file_temporarily(
+    file_url: URL,
+    file_path: FilePath,
+    auth: HTTPBasicAuth,
+) -> None:
     """Download file."""
     with get(file_url, stream=True, auth=auth, timeout=10) as response:
-        with open(file_path, "wb") as file_pointer:
+        with Path(file_path).open(mode="wb") as file_pointer:
             copyfileobj(response.raw, file_pointer)
 
 
@@ -76,7 +83,7 @@ def download_file(
 
     Return path to the downloaded file upon success, empty string upon failure.
     """
-    file_path = f"/tmp/{pure_id}.pdf"
+    file_path = f"/tmp/{pure_id}.pdf"  # noqa: S108
     auth = HTTPBasicAuth(pure_username, pure_password)
     store_file_temporarily(file_url, file_path, auth)
 

run-tests.sh

@@ -1,7 +1,7 @@
 #!/usr/bin/env sh
 # -*- coding: utf-8 -*-
 #
-# Copyright (C) 2020-2022 Technische Universität Graz
+# Copyright (C) 2020-2023 Technische Universität Graz
 #
 # 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.
@@ -12,6 +12,8 @@ set -o errexit
 # Quit on unbound symbols
 set -o nounset
 
+ruff .
+
 python -m check_manifest
 python -m sphinx.cmd.build -qnNW docs docs/_build/html
 python -m pytest