def _validate_cron()

in source/scheduler/common/aws_solutions/scheduler/common/schedule.py [0:0]

• 38 lines of code
• 6 McCabe index (conditional complexity)

    def _validate_cron(self) -> str:
        """
        Perform a partial validation of the cron expression
        :param expression: the cron expression e.g. cron(* * * * ? *)
        :return: the expression e.g. * * * ? *
        """
        schedule = self.expression
        # fmt: off
        cron_re = re.compile(
            r"^cron\((?P<minutes>[^ ]+) (?P<hours>[^ ]+) (?P<day_of_month>[^ ]+) (?P<month>[^ ]+) (?P<day_of_week>[^ ]+) (?P<year>[^ ]+)\)$"
        )
        # fmt: on
        match = cron_re.match(schedule)

        if not match:
            self._configuration_errors.append(
                f"invalid cron ScheduleExpression {schedule}. Should have 6 fields"
            )
        else:
            minutes = match.group("minutes")
            hours = match.group("hours")
            day_of_month = match.group("day_of_month")
            month = match.group("month")
            day_of_week = match.group("day_of_week")
            year = match.group("year")

            if day_of_month != CRON_ANY_WILDCARD and day_of_week != CRON_ANY_WILDCARD:
                self._configuration_errors.append(
                    f"invalid cron ScheduleExpression {schedule}. Do not specify day-of-month and day-of week in the same cron expression"
                )

            # validate the majority of the ScheduleExpression
            try:
                cronex.CronExpression(
                    f"{minutes} {hours} {day_of_month} {month} {day_of_week}"
                )
            except ValueError as exc:
                self._configuration_errors.append(
                    f"invalid cron ScheduleExpression: {exc}"
                )

            # cronex does not validate the year - validate separately
            try:
                cronex.parse_atom(year, CRON_MIN_MAX_YEAR)
            except ValueError as exc:
                self._configuration_errors.append(
                    f"invalid cron ScheduleExpression year: {exc}"
                )

            return (
                f"cron({minutes} {hours} {day_of_month} {month} {day_of_week} {year})"
            )