def get_push_events_commits_paths()

in osci/datalake/base.py [0:0]

• 25 lines of code
• 9 McCabe index (conditional complexity)

    def get_push_events_commits_paths(self, to_date: datetime, date_period_type: str = DatePeriodType.YTD,
                                      from_date: datetime = None, company=None) -> Iterable[str]:
        year, month, day = None, None, None
        if date_period_type == DatePeriodType.DTD:
            if from_date is None:
                raise ValueError(f'`from_date` must be non None for {DatePeriodType.DTD} passed: {from_date}')
            if to_date < from_date:
                raise ValueError(f'`from_date` {from_date} must be less than `to_date` {to_date}')
            if to_date.year == from_date.year:
                year = to_date.year
                if to_date.month == from_date.month:
                    month = to_date.month
                    if to_date.day == from_date.day:
                        day = to_date.day
        elif date_period_type == DatePeriodType.MTD:
            year, month = to_date.year, to_date.month
        elif date_period_type == DatePeriodType.YTD:
            year = to_date.year
        else:
            raise ValueError(f'Unsupported date period type: {date_period_type} not in {DatePeriodType.all}')

        return filter(
            lambda path: self.filter_push_commit_file(self._get_file_name(path), from_date=from_date, to_date=to_date),
            self._get_date_partitioned_paths(dir_path=self._github_events_commits_base,
                                             year=year, month=month, day=day, company=company)
        )