Add find_notebooks and remove more usages of re

Author: shreve

nbgrader/apps/api.py

@@ -1,11 +1,8 @@
-import glob
-import re
 import sys
 import os
 import logging
 import warnings
 import typing
-from pathlib import Path
 
 from traitlets.config import LoggingConfigurable, Config, get_config
 from traitlets import Instance, Enum, Unicode, observe
@@ -386,6 +383,7 @@ def get_notebooks(self, assignment_id):
             A list of dictionaries containing information about each notebook
 
         """
+        notebooks = []
         with self.gradebook as gb:
             try:
                 assignment = gb.find_assignment(assignment_id)
@@ -394,7 +392,6 @@ def get_notebooks(self, assignment_id):
 
             # if the assignment exists in the database
             if assignment and assignment.notebooks:
-                notebooks = []
                 for notebook in assignment.notebooks:
                     x = notebook.to_dict()
                     x["average_score"] = gb.average_notebook_score(notebook.name, assignment.name)
@@ -405,16 +402,13 @@ def get_notebooks(self, assignment_id):
 
             # if it doesn't exist in the database
             else:
-                sourcedir = Path(self.coursedir.format_path(
-                    self.coursedir.source_directory,
+                for notebook in self.coursedir.find_notebooks(
+                    nbgrader_step=self.coursedir.source_directory,
                     student_id='.',
-                    assignment_id=assignment_id))
-
-                notebooks = []
-                for filename in sourcedir.glob("*.ipynb"):
-                    notebook_id = filename.stem
+                    assignment_id=assignment_id
+                ):
                     notebooks.append({
-                        "name": notebook_id,
+                        "name": notebook['notebook_id'],
                         "id": None,
                         "average_score": 0,
                         "average_code_score": 0,

nbgrader/converters/generate_assignment.py

@@ -1,5 +1,3 @@
-import os
-import re
 from textwrap import dedent
 from pathlib import Path
 

nbgrader/converters/generate_solution.py

@@ -1,5 +1,3 @@
-import os
-import re
 from textwrap import dedent
 
 from traitlets import List, Bool, default

nbgrader/converters/generate_source_with_tests.py

@@ -1,6 +1,3 @@
-import os
-import re
-
 from traitlets import List, default
 
 from .base import BaseConverter

nbgrader/coursedir.py

@@ -341,7 +341,7 @@ def find_assignments(self,
         assignment_id: str = "*",
     ) -> typing.List[typing.Dict]:
         """
-        Find all entries that match the given criteria.
+        Find all directories that match the given criteria.
 
         The default value for each acts as a wildcard. To exclude a directory, use
         a value of "." (e.g. nbgrader_step="source", student_id=".").
@@ -389,6 +389,65 @@ def find_assignments(self,
 
         return results
 
+    def find_notebooks(
+        self,
+        nbgrader_step: str = "*",
+        student_id: str = "*",
+        assignment_id: str = "*",
+        notebook_id: str = "*",
+        ext: str = "ipynb",
+    ) -> typing.List[typing.Dict]:
+        """
+        Find all notebooks that match the given criteria.
+
+        The default value for each acts as a wildcard. To exclude a directory, use
+        a value of "." (e.g. nbgrader_step="source", student_id=".").
+
+        Returns:
+            A list of dicts containing input values, one per matching directory.
+        """
+
+        results = []
+
+        kwargs = dict(
+            nbgrader_step=nbgrader_step,
+            student_id=student_id,
+            assignment_id=assignment_id,
+            notebook_id=notebook_id,
+            ext=ext,
+        )
+
+        pattern = os.path.join(self.directory_structure, "{notebook_id}.{ext}")
+
+        # Locate all matching files using a glob
+        files = list(
+            filter(
+                lambda p: p.is_file() and not is_ignored(p.name, self.ignore),
+                Path(self.root).glob(pattern.format(**kwargs))
+            )
+        )
+
+        if not files:
+            return results
+
+        pattern_args = {
+            key: value.replace("*", f"(?P<{key}>.*)")
+            for key, value in kwargs.items()
+        }
+
+        # Convert to a Path and back to a string to remove any instances of `/.`
+        pattern = str(Path(pattern.format(**pattern_args)))
+
+        if sys.platform == 'win32':
+            # Escape backslashes on Windows
+            pattern = pattern.replace('\\', r'\\')
+
+        for file in files:
+            match = re.match(pattern, str(file.relative_to(self.root)))
+            if match:
+                results.append({ **kwargs, **match.groupdict(), "path": file})
+
+        return results
 
     def get_existing_timestamp(self, dest_path: str) -> typing.Optional[datetime.datetime]:
         """Get the timestamp, as a datetime object, of an existing submission."""

nbgrader/exchange/default/release_feedback.py

@@ -1,12 +1,10 @@
 import os
 import shutil
-import glob
-import re
 from stat import S_IRUSR, S_IWUSR, S_IXUSR, S_IRGRP, S_IWGRP, S_IXGRP, S_IXOTH, S_ISGID
 
 from nbgrader.exchange.abc import ExchangeReleaseFeedback as ABCExchangeReleaseFeedback
 from .exchange import Exchange
-from nbgrader.utils import notebook_hash, make_unique_key
+from nbgrader.utils import notebook_hash
 
 
 class ExchangeReleaseFeedback(Exchange, ABCExchangeReleaseFeedback):
@@ -37,40 +35,37 @@ def copy_files(self):
         else:
             exclude_students = set()
 
-        html_files = glob.glob(os.path.join(self.src_path, "*.html"))
-        for html_file in html_files:
-            regexp = re.escape(os.path.sep).join([
-                self.coursedir.format_path(
-                    self.coursedir.feedback_directory,
-                    "(?P<student_id>.*)",
-                    self.coursedir.assignment_id, escape=True),
-                "(?P<notebook_id>.*).html"
-            ])
-
-            m = re.match(regexp, html_file)
-            if m is None:
-                msg = "Could not match '%s' with regexp '%s'" % (html_file, regexp)
-                self.log.error(msg)
-                continue
-
-            gd = m.groupdict()
-            student_id = gd['student_id']
-            notebook_id = gd['notebook_id']
+        for feedback in self.coursedir.find_notebooks(
+            nbgrader_step=self.coursedir.feedback_directory,
+            student_id=self.coursedir.student_id or "*",
+            assignment_id=self.coursedir.assignment_id,
+            notebook_id="*",
+            ext="html"
+        ):
+            html_file = feedback['path']
+            student_id = feedback['student_id']
             if student_id in exclude_students:
                 self.log.debug("Skipping student '{}'".format(student_id))
                 continue
 
-            feedback_dir = os.path.split(html_file)[0]
+            notebook_id = feedback['notebook_id']
+            feedback_dir = html_file.parent
 
-            timestamp = open(os.path.join(feedback_dir, 'timestamp.txt')).read()
-            with open(os.path.join(feedback_dir, "submission_secret.txt")) as fh:
-                submission_secret = fh.read()
+            timestamp = feedback_dir.joinpath('timestamp.txt').read_text()
+            submission_secret = feedback_dir.joinpath("submission_secret.txt").read_text()
 
             checksum = notebook_hash(secret=submission_secret, notebook_id=notebook_id)
             dest = os.path.join(self.dest_path, "{}-tmp.html".format(checksum))
 
-            self.log.info("Releasing feedback for student '{}' on assignment '{}/{}/{}' ({})".format(
-                student_id, self.coursedir.course_id, self.coursedir.assignment_id, notebook_id, timestamp))
+            self.log.info(
+                "Releasing feedback for student '%s' on assignment '%s/%s/%s' (%s)",
+                student_id,
+                self.coursedir.course_id,
+                feedback["assignment_id"],
+                notebook_id,
+                timestamp
+            )
+
             shutil.copy(html_file, dest)
             # Copy to temporary location and mv to update atomically.
             updated_feedback = os.path.join(self.dest_path, "{}.html". format(checksum))