def create()

in sync/trypush.py [0:0]
48 lines of code
8 McCabe index (conditional complexity)

    def create(cls,
               lock: SyncLock,
               sync: DownstreamSync | LandingSync,
               affected_tests: dict[str, list[str]] | None = None,
               stability: bool = False,
               hacks: bool = True,
               try_cls: type = TryFuzzyCommit,
               rebuild_count: int | None = None,
               check_open: bool = True,
               **kwargs: Any
               ) -> TryPush:
        logger.info("Creating try push for PR %s" % sync.pr)
        if check_open and not tree.is_open("try"):
            logger.info("try is closed")
            raise RetryableError(AbortError("Try is closed"))

        # Ensure the required indexes exist
        TaskGroupIndex.get_or_create(sync.git_gecko)
        try_idx = TryCommitIndex.get_or_create(sync.git_gecko)

        git_work = sync.gecko_worktree.get()

        if rebuild_count is None:
            rebuild_count = 0 if not stability else env.config['gecko']['try']['stability_count']
            if not isinstance(rebuild_count, int):
                logger.error("Could not find config for Stability rebuild count, using default 5")
                rebuild_count = 5
        with try_cls(sync.git_gecko, git_work, affected_tests, rebuild_count, hacks=hacks,
                     **kwargs) as c:
            try_rev = c.push()

        data = {
            "try-rev": try_rev,
            "stability": stability,
            "gecko-head": sync.gecko_commits.head.sha1,
            "wpt-head": sync.wpt_commits.head.sha1,
            "status": "open",
            "bug": sync.bug,
        }
        process_name = base.ProcessName.with_seq_id(sync.git_gecko,
                                                    cls.obj_type,
                                                    sync.sync_type,
                                                    str(getattr(sync, sync.obj_id)))
        rv = super().create(lock, sync.git_gecko, process_name, data)
        try_idx.insert(try_idx.make_key(try_rev), process_name)

        with rv.as_mut(lock):
            rv.created = taskcluster.fromNowJSON("0 days")

        if sync.bug is not None:
            env.bz.comment(sync.bug,
                           "Pushed to try%s %s" %
                           (" (stability)" if stability else "",
                            rv.treeherder_url))

        return rv