import logging from calendar import monthrange from datetime import date, datetime, timedelta import boto3 from billing_report.config import Config def assumed_role_session(role_arn: str, base_session: boto3.session.Session = None) -> boto3.session.Session: base_session = base_session or boto3.Session() sts_client = base_session.client("sts") # Call the assume_role method of the STSConnection object and pass the role # ARN and a role session name. assumed_role_object = sts_client.assume_role(RoleArn=role_arn, RoleSessionName=Config.ROLE_SESSION_NAME) # From the response that contains the assumed role, get the temporary # credentials that can be used to make subsequent API calls credentials = assumed_role_object["Credentials"] return boto3.Session( aws_access_key_id=credentials["AccessKeyId"], aws_secret_access_key=credentials["SecretAccessKey"], aws_session_token=credentials["SessionToken"], region_name=Config.AWS_REGION, ) def month_year_iter(start_month: int, start_year: int, end_month: int, end_year: int) -> tuple: ym_start = 12 * start_year + start_month - 1 ym_end = 12 * end_year + end_month # (- 1) removed to include the last month for ym in range(ym_start, ym_end): y, m = divmod(ym, 12) yield y, m + 1 def date_range_split_montly(logger: logging.Logger, start_date: str, end_date: str) -> tuple: """Generate start date,start date for the next month, and year-month for each month between given dates. End date is adjusted to today in case it is current month. """ start_date_obj = datetime.strptime(start_date, "%Y-%m-%d").date() if end_date != "": end_date_obj = datetime.strptime(end_date, "%Y-%m-%d").date() else: end_date_obj = date.today() if start_date_obj >= end_date_obj: raise "Start date can not be earlier than end date" for year, month in month_year_iter( start_year=start_date_obj.year, start_month=start_date_obj.month, end_year=end_date_obj.year, end_month=end_date_obj.month, ): current_month_start = date(year, month, 1).strftime("%Y-%m-%d") current_month = date(year, month, 1).strftime("%Y-%m") if year != end_date_obj.year or month != end_date_obj.month: # The end date is excluded, so we set the first day of the next month next_month = date(year, month, 1) + timedelta(monthrange(year, month)[1]) current_month_end = next_month.strftime("%Y-%m-%d") logger.info("{} {} {} {}".format(year, month, end_date_obj.year, end_date_obj.month)) logger.info("{} {} {}".format(current_month, current_month_start, current_month_end)) yield current_month, current_month_start, current_month_end elif end_date_obj.day > Config.MIN_MONTHLY_REPORT_DAYS: logger.info("{}".format(str(end_date_obj))) current_month_end = end_date_obj.strftime("%Y-%m-%d") logger.info(current_month_end) yield current_month, current_month_start, current_month_end else: logger.info( "Month {0} skipped because MIN_MONTHLY_REPORT_DAYS has not been reached.".format( end_date_obj.strftime("%Y-%m-%d") ) )