1
\$\begingroup\$

I'm writing a project management system focused on forking and web-of-trusting-code, which is a fancy way of saying "take someone's code, do whatever you want with it, and see if others agree with you".

I am trying to use classes to organize the code: an "instance" class, a "project" class, and a "repo" class. An instance contains projects and some extra metadata, a project contains repos and some extra metadata, and finally a repo is a git repo, basically. I'm not sure I'm following best practices, however.

In my Project class, for example, I'm doing some mangling with the project_commit for some degree of compatibility with older versions, so that users can freely downgrade, and I'm not sure I should be doing it like this.

Here are my 3 classes:

class Repo:
    def __init__(self, dbconn, project, url, branch, list_metadata=False):
        self.url = url
        self.branch = branch
        self.project_commit = project.commit

        if not branch:
            self.branchname = "gan" + hashlib.sha256(url.encode("utf-8")).hexdigest()
            self.head = "HEAD"
        else:
            self.branchname = "gan" + hmac.new(branch.encode("utf-8"), url.encode("utf-8"), "sha256").hexdigest()
            self.head = "refs/heads/" + branch

        try:
            self.hash = subprocess.check_output(["git", "-C", cache_home, "show", self.branchname, "-s", "--format=%H", "--"], stderr=subprocess.DEVNULL).decode("utf-8").strip()
        except subprocess.CalledProcessError:
            self.hash = None

        self.message = None
        if list_metadata:
            try:
                self.fetch_metadata()
            except GitError:
                pass

    def fetch_metadata(self):
        try:
            self.message = subprocess.check_output(["git", "-C", cache_home, "show", self.branchname, "-s", "--format=%B", "--"], stderr=subprocess.DEVNULL).decode("utf-8", "replace")
        except subprocess.CalledProcessError as e:
            raise GitError from e

    def update(self): # FIXME?
        try:
            subprocess.check_output(["git", "-C", cache_home, "fetch", "-q", self.url, "+" + self.head + ":" + self.branchname], stderr=subprocess.STDOUT)
        except subprocess.CalledProcessError as e:
            # This may error for various reasons, but some are important: dead links, etc
            click.echo(e.output, err=True)
            return None
        pre_hash = self.hash
        post_hash = subprocess.check_output(["git", "-C", cache_home, "show", self.branchname, "-s", "--format=%H", "--"], stderr=subprocess.DEVNULL).decode("utf-8").strip()
        self.hash = post_hash
        if not pre_hash:
            pre_hash = post_hash
        try:
            count = int(subprocess.check_output(["git", "-C", cache_home, "rev-list", "--count", pre_hash + ".." + post_hash, "--"]).decode("utf-8").strip())
        except subprocess.CalledProcessError:
            count = 0  # force-pushed
        try:
            subprocess.check_call(["git", "-C", cache_home, "merge-base", "--is-ancestor", self.project_commit, self.branchname], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
            self.fetch_metadata()
            return count, post_hash, self.message
        except (subprocess.CalledProcessError, GitError) as e:
            click.echo(e, err=True)
            return None

class Project:
    def __init__(self, dbconn, ganarchy, project_commit, list_repos=False):
        self.commit = project_commit
        if ganarchy.project_commit == project_commit:
            project_commit = None
        self.refresh_metadata()
        if list_repos:
            repos = []
            with dbconn:
                for (e, url, branch) in dbconn.execute('''SELECT "max"("e"), "url", "branch" FROM (SELECT "max"("T1"."entry") "e", "T1"."url", "T1"."branch" FROM "repo_history" "T1"
                                                                    WHERE (SELECT "active" FROM "repos" "T2" WHERE "url" = "T1"."url" AND "branch" IS "T1"."branch" AND "project" IS ?1)
                                                                    GROUP BY "T1"."url", "T1"."branch"
                                                                    UNION
                                                                    SELECT null, "T3"."url", "T3"."branch" FROM "repos" "T3" WHERE "active" AND "project" IS ?1)
                                           GROUP BY "url" ORDER BY "e"''', (project_commit,)):
                    repos.append(Repo(dbconn, self, url, branch))
            self.repos = repos
        else:
            self.repos = None

    def refresh_metadata(self):
        try:
            project = subprocess.check_output(["git", "-C", cache_home, "show", self.commit, "-s", "--format=%B", "--"], stderr=subprocess.DEVNULL).decode("utf-8", "replace")
            project_title, project_desc = (lambda x: x.groups() if x is not None else ('', None))(re.fullmatch('^\\[Project\\]\s+(.+?)(?:\n\n(.+))?$', project, flags=re.ASCII|re.DOTALL|re.IGNORECASE))
            if not project_title.strip():
                project_title, project_desc = ("Error parsing project commit",)*2
            if project_desc:
                project_desc = project_desc.strip()
            self.commit_body = project
            self.title = project_title
            self.description = project_desc
        except subprocess.CalledProcessError:
            self.commit_body = None
            self.title = None
            self.description = None

    def update(self):
        # TODO
        self.refresh_metadata()

class GAnarchy:
    def __init__(self, dbconn, list_projects=False):
        with dbconn:
            # TODO
            #(project_commit, base_url, title) = dbconn.execute('''SELECT "git_commit", "base_url", "title" FROM "config"''').fetchone()
            (project_commit, base_url) = dbconn.execute('''SELECT "git_commit", "base_url" FROM "config"''').fetchone()
            title = None
            self.project_commit = project_commit
            self.base_url = base_url
            if not base_url:
                pass ## TODO
            if not title:
                from urllib.parse import urlparse
                title = "GAnarchy on " + urlparse(base_url).hostname
            self.title = title
        if list_projects:
            projects = []
            with dbconn:
                for (project,) in dbconn.execute('''SELECT DISTINCT "project" FROM "repos" '''): # FIXME? *maybe* sort by activity in the future
                    if project == None:
                        project = self.project_commit
                    projects.append(Project(dbconn, ganarchy, project))
            projects.sort(key=lambda project: project.title)
            self.projects = projects
        else:
            self.projects = None

Here's the rest of my code: https://cybre.tech/SoniEx2/ganarchy/src/commit/4bd53221bb86c7fefe6b8294b7576486a3d8eb44/ganarchy.py

\$\endgroup\$
  • \$\begingroup\$ Be sure to include all relevant parts of your code that you want to have reviewed with the question. That includes imports and other referenced classes you have written yourself. \$\endgroup\$ – AlexV Jun 10 at 19:41

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.