sources: Bring back fetching the Git repositories

Signed-off-by: Michael Tremer <michael.tremer@ipfire.org>

diff --git a/src/buildservice/__init__.py b/src/buildservice/__init__.py
index fc266de747c3f14645de72946aa56179549a2b09..b1f80bf376a661d1f7417da204b87e50cacaf6d6 100644 (file)
--- a/src/buildservice/__init__.py
+++ b/src/buildservice/__init__.py
@@ -106,7 +106,7 @@ class Backend(object):
                self.run_periodic_task(300, self.mirrors.check)
 
                # Regularly fetch sources
-               #self.run_periodic_task(300, self.sources.fetch)
+               self.run_periodic_task(300, self.sources.fetch)
 
                # Regularly check for new releases
                # XXX Disabled for now

diff --git a/src/buildservice/sources.py b/src/buildservice/sources.py
index 033a385bb4582577b4c3f77875336813d8bc2857..117116d31b027aafbabd3ac0952e9882820ac924 100644 (file)
--- a/src/buildservice/sources.py
+++ b/src/buildservice/sources.py
@@ -39,19 +39,20 @@ VALID_TAGS = (
 
 class Sources(base.Object):
        def __aiter__(self):
-               sources = self._get_sources("""
-                       SELECT
-                               *
-                       FROM
-                               sources
-                       WHERE
-                               deleted_at IS NULL
-                       ORDER BY
-                               created_at
-                       """,
+               stmt = (
+                       sqlalchemy
+                       .select(
+                               Source,
+                       )
+                       .where(
+                               Source.deleted_at == None,
+                       )
+                       .order_by(
+                               Source.created_at.asc(),
+                       )
                )
 
-               return aiter(sources)
+               return self.db.fetch(stmt)
 
        async def get_by_slug(self, slug):
                stmt = (
@@ -97,10 +98,12 @@ class Sources(base.Object):
                """
                        Fetches any new commits from all sources
                """
-               # Fetch all sources concurrently
-               async with asyncio.TaskGroup() as tg:
-                       for source in self:
-                               tg.create_task(source.fetch())
+               # Fetch all sources
+               async for source in self:
+                       await source.fetch()
+
+                       # Commit!
+                       await self.db.commit()
 
                # Process any pending jobs
                if run_jobs:
@@ -264,12 +267,15 @@ class Source(database.Base, database.BackendMixin, database.SoftDeleteMixin):
                        await self.git.show_attribute(revision, r"%aI"),
                )
 
+               # Convert the date to UTC
+               date = date.astimezone(datetime.timezone.utc).replace(tzinfo=None)
+
                # Create a new build group
-               group = self.backend.builds.create_group()
+               builds = await self.backend.builds.create_group()
 
                # Insert into the database
                commit = await self.db.insert(
-                       Commit,
+                       SourceCommit,
                        source    = self,
                        revision  = revision,
                        author    = author,
@@ -277,7 +283,7 @@ class Source(database.Base, database.BackendMixin, database.SoftDeleteMixin):
                        subject   = subject,
                        body      = body,
                        date      = date,
-                       group     = group,
+                       builds    = builds,
                )
 
                # If we are processing the initial commit, we get a list of all files in the tree
@@ -312,7 +318,7 @@ class Source(database.Base, database.BackendMixin, database.SoftDeleteMixin):
                                raise RuntimeError("Unhandled status %s for file %s" % (status, filename))
 
                        # Create job
-                       commit._create_job(action, name)
+                       await commit._create_job(action=action, name=name)
 
                return commit
 
@@ -362,26 +368,25 @@ class Source(database.Base, database.BackendMixin, database.SoftDeleteMixin):
                else:
                        await self.git.clone()
 
-               with self.db.transaction():
-                       # Did we already import something?
-                       initial_commit = not self.revision
+               # Did we already import something?
+               initial_commit = not self.revision
 
-                       # Determine which commits there are to process
-                       revisions = await self.git.revisions(self.revision, self.branch)
+               # Determine which commits there are to process
+               revisions = await self.git.revisions(self.revision, self.branch)
 
-                       # Import all revisions
-                       for revision in revisions:
-                               # Import the commit
-                               await self._create_commit(revision, initial_commit=initial_commit)
+               # Import all revisions
+               for revision in revisions:
+                       # Import the commit
+                       await self._create_commit(revision, initial_commit=initial_commit)
 
-                               # Store the updated revision
-                               self._set_attribute("revision", revision)
+                       # Store the updated revision
+                       self.revision = revision
 
-                               # Only the first revision would the initial commit
-                               initial_commit = False
+                       # Only the first revision would the initial commit
+                       initial_commit = False
 
-                       # Store when we fetched
-                       self._set_attribute_now("last_fetched_at")
+               # Store when we fetched
+               self.last_fetched_at = sqlalchemy.func.current_timestamp()
 
 
 class SourceCommit(database.Base, database.BackendMixin, database.SoftDeleteMixin):
@@ -550,31 +555,18 @@ class SourceCommit(database.Base, database.BackendMixin, database.SoftDeleteMixi
 
        # Jobs
 
-       async def _create_job(self, action, name):
+       async def _create_job(self, **kwargs):
                """
                        Creates a new job
                """
-               log.info("%s: %s: Created '%s' job for '%s'" \
-                       % (self.source, self.revision, action, name))
-
-               job = await self.backend.sources._get_job("""
-                       INSERT INTO
-                               source_commit_jobs
-                       (
-                               commit_id,
-                               action,
-                               name
-                       )
-                       VALUES
-                       (
-                               %s, %s, %s
-                       )
-                       RETURNING *
-                       """, self.id, action, name,
+               job = await self.db.insert(
+                       SourceJob,
+                       commit = self,
+                       **kwargs,
                )
 
-               # Append to cache
-               self.jobs.add(job)
+               log.info("%s: %s: Created '%s' job for '%s'" \
+                       % (self.source, self.revision, job.action, job.name))
 
                return job