badges: updated badges to new architecture

invenio_github/api.py

@@ -25,6 +25,7 @@
 """Invenio module that adds GitHub integration to the platform."""
 
 import json
+from abc import abstractmethod
 from contextlib import contextmanager
 from copy import deepcopy
 
@@ -56,10 +57,12 @@
 )
 
 
-def check_repo_access_permissions(repo, user_id, repo_id, repo_name):
+def check_repo_access_permissions(repo, user_id):
     """Checks permissions from user on repo."""
     if repo and repo.user_id and repo.user_id != int(user_id):
-        raise RepositoryAccessError(user=user_id, repo=repo_name, repo_id=repo_id)
+        raise RepositoryAccessError(
+            user=user_id, repo=repo.name, repo_id=repo.github_id
+        )
 
 
 class GitHubAPI(object):
@@ -264,9 +267,7 @@ def create_hook(self, repo_id, repo_name):
         if not repo:
             repo = Repository.create(self.user_id, repo_id, repo_name)
 
-        check_repo_access_permissions(
-            repo, self.user_id, repo_id=repo_id, repo_name=repo_name
-        )
+        check_repo_access_permissions(repo, self.user_id)
 
         # Create hook
         hook_config = dict(
@@ -307,9 +308,7 @@ def remove_hook(self, repo_id, name):
         if not repo:
             raise RepositoryNotFoundError(repo_id)
 
-        check_repo_access_permissions(
-            repo, self.user_id, repo_id=repo_id, repo_name=name
-        )
+        check_repo_access_permissions(repo, self.user_id)
 
         ghrepo = self.api.repository_with_id(repo_id)
         if ghrepo:
@@ -322,9 +321,13 @@ def remove_hook(self, repo_id, name):
                 return True
         return False
 
+    def repo_last_published_release(self, repo):
+        """Retrieves the repository last release."""
+        return repo.latest_release(ReleaseStatus.PUBLISHED)
+
     def get_repository_releases(self, repo):
         """Retrieve repository releases. Returns API release objects."""
-        check_repo_access_permissions(repo, self.user_id, repo.github_id, repo.name)
+        check_repo_access_permissions(repo, self.user_id)
 
         # Retrieve releases and sort them by creation date
         release_instances = []
@@ -339,9 +342,7 @@ def get_user_repositories(self):
         repos = deepcopy(self.user_available_repositories)
         if repos:
             # 'Enhance' our repos dict, from our database model
-            db_repos = self.user_enabled_repositories
-            for repo in db_repos:
-                # TODO here
+            for repo in self.user_enabled_repositories:
                 if str(repo.github_id) in repos:
                     release_instance = current_github.release_api_class(
                         repo.latest_release()
@@ -362,14 +363,14 @@ def user_available_repositories(self):
 
     def disable_repo(self, repo):
         """Disables an user repository if the user has permission to do so."""
-        check_repo_access_permissions(repo, self.user_id, repo.github_id, repo.name)
+        check_repo_access_permissions(repo, self.user_id)
 
         repo.hook = None
         repo.user_id = None
 
     def enable_repo(self, repo, hook):
         """Enables an user repository if the user has permission to do so."""
-        check_repo_access_permissions(repo, self.user_id, repo.github_id, repo.name)
+        check_repo_access_permissions(repo, self.user_id)
 
         repo.hook = hook
         repo.user_id = self.user_id
@@ -390,17 +391,17 @@ def get_last_sync_time(self):
         )
         return last_sync
 
-    def get_repository(self, repo_name):
+    def get_repository(self, repo_name=None, repo_github_id=None):
         """Retrieves one repository.
 
         Checks for access permission.
         """
-        repo = Repository.get(name=repo_name)
+        repo = Repository.get(name=repo_name, github_id=repo_github_id)
         if not repo:
             raise RepositoryNotFoundError(repo_name)
 
         # Might raise a RepositoryAccessError
-        check_repo_access_permissions(repo, self.user_id, repo.github_id, repo_name)
+        check_repo_access_permissions(repo, self.user_id)
 
         return repo
 
@@ -573,7 +574,6 @@ def test_zipball(self):
 
     # High level API
 
-    # TODO split maybe
     def release_failed(self):
         """Set release status to FAILED."""
         self.release_object.status = ReleaseStatus.FAILED
@@ -626,3 +626,20 @@ def resolve_record(self):
     def serialize_record(self):
         """Serializes the release record."""
         raise NotImplementedError
+
+    @property
+    @abstractmethod
+    def badge_title(self):
+        """Stores a string to render in the record badge title (e.g. 'DOI')."""
+        return None
+
+    @property
+    @abstractmethod
+    def badge_value(self):
+        """Stores a string to render in the record badge value (e.g. '10.1234/invenio.1234')."""
+        return None
+
+    @property
+    def self_url(self):
+        """Release self url (e.g. github HTML url)."""
+        return self.release_payload.get("html_url")

invenio_github/views/badge.py

@@ -1,7 +1,7 @@
 # -*- coding: utf-8 -*-
 #
 # This file is part of Invenio.
-# Copyright (C) 2014, 2015, 2016 CERN.
+# Copyright (C) 2014-2023 CERN.
 #
 # Invenio is free software; you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
@@ -25,10 +25,11 @@
 
 from __future__ import absolute_import
 
-from flask import Blueprint, abort, redirect, url_for
+from flask import Blueprint, abort, current_app, redirect, url_for
+from flask_login import current_user
 
-from ..api import GitHubRelease
-from ..models import ReleaseStatus, Repository
+from invenio_github.api import GitHubAPI
+from invenio_github.proxies import current_github
 
 blueprint = Blueprint(
     "invenio_github_badge",
@@ -39,56 +40,26 @@
 )
 
 
-def get_pid_of_latest_release_or_404(**kwargs):
-    """Return PID of the latest release."""
-    repo = Repository.query.filter_by(**kwargs).first_or_404()
-    release = repo.latest_release(ReleaseStatus.PUBLISHED)
-    if release:
-        return GitHubRelease(release).pid
-    abort(404)
-
-
-def get_badge_image_url(pid, ext="svg"):
-    """Return the badge for a DOI."""
-    return url_for(
-        "invenio_formatter_badges.badge",
-        title=pid.pid_type,
-        value=pid.pid_value,
-        ext=ext,
-    )
-
-
-def get_doi_url(pid):
-    """Return the badge for a DOI."""
-    return "https://doi.org/{pid.pid_value}".format(pid=pid)
-
-
 #
 # Views
 #
-@blueprint.route("/<int:github_id>.svg")
-def index(github_id):
-    """Generate a badge for a specific GitHub repository."""
-    pid = get_pid_of_latest_release_or_404(github_id=github_id)
-    return redirect(get_badge_image_url(pid))
 
 
-@blueprint.route("/<int:user_id>/<path:repo_name>.svg")
-def index_old(user_id, repo_name):
+@blueprint.route("/<int:repo_github_id>.svg")
+def index(repo_github_id):
     """Generate a badge for a specific GitHub repository."""
-    pid = get_pid_of_latest_release_or_404(name=repo_name)
-    return redirect(get_badge_image_url(pid))
-
-
-@blueprint.route("/latestdoi/<int:github_id>")
-def latest_doi(github_id):
-    """Redirect to the newest record version."""
-    pid = get_pid_of_latest_release_or_404(github_id=github_id)
-    return redirect(get_doi_url(pid))
-
-
-@blueprint.route("/latestdoi/<int:user_id>/<path:repo_name>")
-def latest_doi_old(user_id, repo_name):
-    """Redirect to the newest record version."""
-    pid = get_pid_of_latest_release_or_404(name=repo_name)
-    return redirect(get_doi_url(pid))
+    try:
+        github_api = GitHubAPI(current_user.id)
+        repo = github_api.get_repository(repo_github_id=repo_github_id)
+        release_object = github_api.repo_last_published_release(repo)
+        release = current_github.release_api_class(release_object)
+        badge_url = url_for(
+            "invenio_formatter_badges.badge",
+            title=release.badge_title,
+            value=release.badge_value,
+            ext="svg",
+        )
+        return redirect(badge_url)
+    except Exception as e:
+        current_app.logger.error(str(e), exc_info=True)
+        abort(404)

invenio_github/views/github.py

@@ -84,7 +84,6 @@ def register_ui_routes(blueprint):
 
     @blueprint.route("/")
     @login_required
-    @request_session_token()
     @register_menu(  # TODO modify?
         blueprint,
         "settings.github",
@@ -106,19 +105,18 @@ def get_repositories():
         """Display list of the user's repositories."""
         github = GitHubAPI(user_id=current_user.id)
         ctx = dict(connected=False)
-
-        # Generate the repositories view object
-        repos = github.get_user_repositories()
-        last_sync = github.get_last_sync_time()
-
-        ctx.update(
-            {
-                # TODO maybe can be refactored. e.g. have two templates and render the correct one.
-                "connected": True,
-                "repos": sorted(repos.items(), key=lambda x: x[1]["full_name"]),
-                "last_sync": last_sync,
-            }
-        )
+        if github.session_token:
+            # Generate the repositories view object
+            repos = github.get_user_repositories()
+            last_sync = github.get_last_sync_time()
+
+            ctx.update(
+                {
+                    "connected": True,
+                    "repos": sorted(repos.items(), key=lambda x: x[1]["full_name"]),
+                    "last_sync": last_sync,
+                }
+            )
 
         return render_template(current_app.config["GITHUB_TEMPLATE_INDEX"], **ctx)
 
@@ -142,16 +140,17 @@ def get_repository(repo_name):
                 repo=repo,
                 releases=releases,
             )
-        except RepositoryAccessError as e:
+        except RepositoryAccessError:
             abort(403)
-        except NoResultFound as e:
+        except NoResultFound:
             abort(404)
 
 
 def register_api_routes(blueprint):
     """Register API routes."""
 
     @login_required
+    @request_session_token()
     @blueprint.route("/user/github/repositories/sync", methods=["POST"])
     def sync_user_repositories():
         """Synchronizes user repos.