Newer
Older
def instantiate_flow_page(
location: str, repo: Repo_ish, page_desc: FlowPageDesc, commit_sha: bytes
) -> PageBase:
class_ = get_flow_page_class(repo, page_desc.type, commit_sha)
from course.page.base import PageBase
if not issubclass(class_, PageBase):
raise ClassNotFoundError(f"'{page_desc.type}' is not a PageBase subclass")
return class_(None, location, page_desc)
class CourseCommitSHADoesNotExist(Exception):
Andreas Klöckner
committed
def get_course_commit_sha(
course: Course,
participation: Participation | None,
repo: Repo_ish | None = None,
raise_on_nonexistent_preview_commit: bool | None = False
) -> bytes:
sha = course.active_git_commit_sha
def is_commit_sha_valid(repo: Repo_ish, commit_sha: str) -> bool:
if isinstance(repo, SubdirRepoWrapper):
repo = repo.repo
try:
repo[commit_sha.encode()]
except KeyError:
if raise_on_nonexistent_preview_commit:
raise CourseCommitSHADoesNotExist(
_("Preview revision '{}' does not exist--"
"showing active course content instead.").format(commit_sha))
return False
return True
Andreas Klöckner
committed
if participation is not None:
if participation.preview_git_commit_sha:
preview_sha = participation.preview_git_commit_sha
preview_sha_valid = is_commit_sha_valid(repo, preview_sha)
else:
with get_course_repo(course) as repo:
preview_sha_valid = is_commit_sha_valid(repo, preview_sha)
Andreas Klöckner
committed
sha = preview_sha
def list_flow_ids(repo: Repo_ish, commit_sha: bytes) -> list[str]:
flows_tree = get_repo_tree(repo, "flows", commit_sha)
except ObjectDoesNotExist:
# That's OK--no flows yet.
pass
else:
for entry in flows_tree.items():
flow_ids.append(entry.path[:-4].decode("utf-8"))