build_jobs.py 文件源码

def post(self, build: Build):
        """
        Create a new job.
        """
        result = self.schema_from_request(job_schema, partial=True)
        if result.errors:
            return self.respond(result.errors, 403)
        data = result.data
        job = Job(build=build, repository_id=build.repository_id, **data)
        if job.status != Status.queued and not job.date_started:
            job.date_started = timezone.now()

        db.session.add(job)

        try:
            db.session.commit()
        except IntegrityError:
            db.session.rollback()
            return self.respond(status=422)

        aggregate_build_stats_for_job.delay(job_id=job.id)

        return self.respond_with_schema(job_schema, job)