def checkpoint_field()

in Solutions/Mimecast/Data Connectors/MimecastAudit/MimecastAudit/mimecast_audit_to_sentinel.py [0:0]

• 126 lines of code
• 10 McCabe index (conditional complexity)

    def checkpoint_field(self):
        """Set the start date and end date in checkpoint file.

        Returns:
            json : Parsed checkpoint
        """
        __method_name = inspect.currentframe().f_code.co_name
        try:
            checkpoint = self.get_checkpoint_data(self.state_manager, load_flag=True)

            if not checkpoint:
                checkpoint = {}

            utc_timestamp = (
                datetime.datetime.utcnow()
                .replace(tzinfo=datetime.timezone.utc)
                .isoformat()
            )

            checkpoint_updated = False
            if checkpoint.get("start_time") is None:
                applogger.info(
                    self.log_format.format(
                        consts.LOGS_STARTS_WITH,
                        __method_name,
                        self.azure_function_name,
                        "Checkpoint is empty ",
                    )
                )

                start_date = None
                if not consts.START_DATE:
                    start_date = self.get_utc_time_in_past(days=consts.DAYS_BACK)
                    applogger.info(
                        self.log_format.format(
                            consts.LOGS_STARTS_WITH,
                            __method_name,
                            self.azure_function_name,
                            "Date set to {} days in the past".format(consts.DAYS_BACK),
                        )
                    )
                else:
                    start_date = (
                        datetime.datetime.strptime(consts.START_DATE, "%Y-%m-%d")
                        .replace(
                            hour=0, minute=0, second=0, tzinfo=datetime.timezone.utc
                        )
                        .strftime(consts.TIME_FORMAT)
                    )
                    now = datetime.datetime.utcnow().strftime(consts.TIME_FORMAT)
                    last_valid_date = self.get_utc_time_in_past(
                        days=consts.VALID_PREVIOUS_DAY
                    )
                    if start_date > now:
                        applogger.error(
                            self.log_format.format(
                                consts.LOGS_STARTS_WITH,
                                __method_name,
                                self.azure_function_name,
                                "Error Occurred while validating params. StartTime cannot be in the future.",
                            )
                        )
                        raise MimecastException()
                    elif start_date < last_valid_date:
                        applogger.info(
                            self.log_format.format(
                                consts.LOGS_STARTS_WITH,
                                __method_name,
                                self.azure_function_name,
                                "Date provided is older than 60 days. "
                                "Ingestion will start from this date: {}".format(
                                    last_valid_date
                                ),
                            )
                        )

                    applogger.info(
                        self.log_format.format(
                            consts.LOGS_STARTS_WITH,
                            __method_name,
                            self.azure_function_name,
                            "Date taken from user input",
                        )
                    )

                mimecast_start_date = datetime.datetime.strptime(
                    start_date, consts.TIME_FORMAT
                )
                checkpoint["start_time"] = mimecast_start_date.strftime(
                    consts.TIME_FORMAT
                )

                end_date = datetime.datetime.fromisoformat(utc_timestamp)
                mimecast_end_date = end_date.strftime(consts.TIME_FORMAT)
                checkpoint["end_time"] = mimecast_end_date

                checkpoint_updated = True
                applogger.info(
                    self.log_format.format(
                        consts.LOGS_STARTS_WITH,
                        __method_name,
                        self.azure_function_name,
                        "Start and end dates initialized in the checkpoint file.",
                    )
                )

            self.start_date = checkpoint["start_time"]
            if checkpoint_updated:
                self.post_checkpoint_data(self.state_manager, data=checkpoint)

                applogger.info(
                    self.log_format.format(
                        consts.LOGS_STARTS_WITH,
                        __method_name,
                        self.azure_function_name,
                        "Checkpoint data fetched and written",
                    )
                )

            return checkpoint
        except MimecastException:
            raise MimecastException()
        except KeyError as key_error:
            applogger.error(
                self.log_format.format(
                    consts.LOGS_STARTS_WITH,
                    __method_name,
                    self.azure_function_name,
                    consts.KEY_ERROR_MSG.format(key_error),
                )
            )
            raise MimecastException()
        except Exception as err:
            applogger.error(
                self.log_format.format(
                    consts.LOGS_STARTS_WITH,
                    __method_name,
                    self.azure_function_name,
                    consts.UNEXPECTED_ERROR_MSG.format(err),
                )
            )
            raise MimecastException()