Jobs mutations

backend/app/neo4j_driver.py

@@ -214,6 +214,74 @@ def get_github_organization_by_slug(self, slug: str):
             )
             return result.single()
 
+    def merge_source(self, kind: str, name: str ):
+        with self.driver.session() as session:
+            result = session.run(
+                """
+                MERGE (s:Source {kind: $kind})
+                ON CREATE SET s.name = $name
+                RETURN s
+                """,
+                kind=kind,
+                name=name
+            )
+            return result.single()["s"]
+
+    def import_repository_job(self, repo_url: str, job_id: Optional[str] = None):
+        if job_id is None:
+            job_id = uuid4()
+
+        with self.driver.session() as session:
+            result = session.run(
+                """
+                MATCH (r:GithubRepository {url: $repo_url})
+                MERGE (s:Source {kind: 'GIT'})
+                CREATE (j:Job {job_id: $job_id, created_at: datetime()})
+                MERGE (j)<-[ib:IMPORTED_BY]-(r)
+                MERGE (j)-[:IMPORTS_SOURCE]->(s)
+                ON CREATE SET ib.import_job_scheduled = datetime()
+                RETURN j
+                """,
+                repo_url=repo_url,
+                job_id=job_id
+            )
+            return result.single()["j"]
+
+
+    def start_job(self, job_id: str):
+        with self.driver.session() as session:
+            result = session.run(
+                """
+                MATCH (j:Job {job_id: $job_id})
+                SET j.started_at = datetime()
+                RETURN j
+                """,
+                job_id=job_id
+            )
+            return result.single()["j"]
+
+    def complete_job(self, job_id: str):
+        with self.driver.session() as session:
+            result = session.run(
+                """
+                MATCH (j:Job {job_id: $job_id})
+                SET j.completed_at = datetime()
+                RETURN j
+                """,
+                job_id=job_id
+            )
+            return result.single()["j"]
+
+    def get_unstarted_jobs_and_sources(self):
+        with self.driver.session() as session:
+            result = session.run(
+                """
+                MATCH (j:Job)-[:IMPORTS_SOURCE]->(s:Source)
+                WHERE NOT EXISTS(j.started_at)
+                RETURN j.job_id AS job_id, s.kind AS kind, s.url AS url
+                """
+            )
+            return [record.data() for record in result]
 
     def get_github_repository_by_url(self, url: str):
         with self.driver.session() as session:

backend/app/schema.py

@@ -7,6 +7,22 @@
 import gitfame
 from contextlib import redirect_stdout
 
+class SourceKind(graphene.Enum):
+    """Enumerates the possible kinds of data sources we support."""
+    GIT = "GIT"
+
+
+class Source(graphene.ObjectType):
+    """Represents a data source (e.g. a Git repository, an external API)."""
+    name = graphene.String()
+    kind = graphene.Field(SourceKind)
+
+
+class Job(graphene.ObjectType):
+    """Represents an asynchronous import job."""
+    run_id = graphene.String()
+    repo_url = graphene.String()
+    status = graphene.String()
 
 class FameDetail(graphene.ObjectType):
     email = graphene.String()
@@ -57,7 +73,7 @@ class Query(graphene.ObjectType):
 
     def resolve_organizations(self, info):
         db = Neo4jDriver()
-        orgs = db.get_all_organizations()
+        orgs = db.get_all_github_organizations()
         db.close()
         return [GithubOrganization(**org) for org in orgs]
 
@@ -67,7 +83,7 @@ def resolve_organizations(self, info):
 
     def resolve_github_repositories(self, info):
         db = Neo4jDriver()
-        repos = db.get_all_repositories()
+        repos = db.get_all_github_repositories()
         db.close()
         return [GithubRepository(**repo) for repo in repos]
 
@@ -197,9 +213,29 @@ def mutate(self, info, org_slug, name, url, description=""):
         db.close()
         return CreateGithubRepository(github_repository=GithubRepository(name=repo["name"], url=repo["url"], description=repo["description"]))
 
+class ImportRepositoryJob(graphene.Mutation):
+    """Queue a job to import a repository. Returns the enqueued Job object."""
+
+    class Arguments:
+        repo_url = graphene.String(required=True)
+        run_id = graphene.String(required=True)
+
+    job = graphene.Field(Job)
+
+    def mutate(self, info, repo_url, run_id):
+        db = Neo4jDriver()
+        job_data = db.import_repository_job(repo_url, run_id)
+        db.close()
+        if not job_data:
+            raise GraphQLError("Failed to create import job")
+        return ImportRepositoryJob(job=Job(**job_data))
+
+
+
 class Mutation(graphene.ObjectType):
     create_github_organization = CreateGithubOrganization.Field()
     create_github_repository = CreateGithubRepository.Field()
+    import_repository_job = ImportRepositoryJob.Field()
 
 schema = graphene.Schema(query=Query, mutation=Mutation)
 ma = graphene.Schema(query=Query)