Skip to content

AWS - Reference

This section provides a structured breakdown of the main application module and its supporting utilities used in the eraXplor project.

🌟eraXplor source code

🔹Main Application Module

▶️ Entry Point

eraXplor - AWS Cost Export Tool

The official CLI interface for exporting AWS cost and usage data via AWS Cost Explorer API. Provides flexible filtering, grouping, and output options for cost analysis.

Command Line Arguments

--start-date, -s DATE Start date in YYYY-MM-DD format. Default: 3 months prior

--end-date, -e DATE End date in YYYY-MM-DD format. Default: Current date

--profile, -p PROFILE AWS credential profile name. Default: 'default'

--groupby, -g DIMENSION Cost grouping dimension. Options: - LINKED_ACCOUNT (default) - SERVICE - PURCHASE_TYPE - USAGE_TYPE - LINKED_ACCOUNT-With-SERVICE - LINKED_ACCOUNT-With-PURCHASE_TYPE - LINKED_ACCOUNT-With-USAGE_TYPE

--out, -o FILENAME Output CSV filename. Default: 'cost_report_.csv'

--granularity, -G GRAN Time granularity. Options: - MONTHLY (default) - DAILY

Examples:

  1. Basic usage with default settings: eraXplor-aws

  2. Custom date range and profile: eraXplor-aws -s 2025-01-01 -e 2025-03-30 -p production

  3. Service-level breakdown with daily granularity: eraXplor-aws -g SERVICE -G DAILY -o service_costs.csv

  4. Account+Service combined analysis: eraXplor-aws -g LINKED_ACCOUNT-With-SERVICE

Notes
  • Requires AWS credentials configured via CLI or IAM role
  • Date range cannot exceed 14 months per AWS limitations
  • Output files contain unblended costs in USD

main()

Orchestrates & Manage depends of cost export workflow.

Source code in src/eraXplor_aws/__main__.py 
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def main() -> None:
    """Orchestrates & Manage depends of cost export workflow."""
    # Banner
    _banner_format, _copyright_notice = generate_banner()
    print(f"\n\n {termcolor.colored(_banner_format, color="green")}")
    print(f"{termcolor.colored(_copyright_notice, color="green")}", end="\n\n")

    # fetch Parsed parameters by command line
    arg_parser = parser().parse_args()

    # Select start date handler
    start_date_input = parser_start_date_handler(arg_parser)

    # Select end date handler
    end_date_input = parser_end_date_handler(arg_parser)

    # Select profile name
    aws_profile_name_input = parser_profile_handler(arg_parser)

    # Select cost groupby key
    cost_groupby_key_input = parser_groupby_handler(arg_parser)

    # Select output filename
    filename = parser_filename_handler(arg_parser)

    # check granularity
    granularity = parser_granularity_handler(arg_parser)

    # Fetch monthly account cost usage
    results = monthly_account_cost_export(
        start_date_input, end_date_input,
        aws_profile_name_input,
        cost_groupby_key_input,
        granularity)

    print(json.dumps(results, indent=4, default=str), end="\n\n\n")

    # Export results to CSV
    csv_export(results, filename)

This is the primary script responsible for orchestrating the user workflow. It handles user input, invokes AWS cost data retrieval, and manages data export functionality.

🛠 Utility Modules

Module to display a banner and copyright notice.

banner()

Generates a banner and copyright notice for the application.

Source code in src/eraXplor_aws/utils/banner_utils.py 
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
def banner():
    """Generates a banner and copyright notice for the application."""

    copyright_notice = """╔══════════════════════════════════════════════════╗
║  © 2025 Mohamed eraki                            ║
║  mohamed-ibrahim2021@outlook.com                 ║
║  Version: 3.2.0                                  ║
║  eraXplor - AWS Cost exporter Tool               ║
╚══════════════════════════════════════════════════╝
    """
    banner_format = pyfiglet.figlet_format("eraXplor", font='slant')
    return banner_format, copyright_notice

Responsible for rendering styled ASCII banners and displaying copyright information used in the CLI interface.

📊 Cost Export Utilities

Module to retrieve AWS account cost data using AWS Cost Explorer API.

monthly_account_cost_export(start_date_input, end_date_input, aws_profile_name_input, cost_groupby_key_input, granularity)

Retrieve AWS cost and usage data via Cost Explorer API.

Fetches unblended costs across all linked accounts in an AWS organization, with flexible grouping and granularity options. Data is returned in standardized records suitable for analysis or export.

Parameters:

Name Type Description Default
start_date_input Union[str, datetime]

[REQUIRED] Start date for cost report (inclusive). Default: "3 Months ago" Format: YYYY-MM-DD datetime object. Note: AWS limits historical data to 14 months.

 required
end_date_input Union[str, datetime]

[REQUIRED] End date for cost report (inclusive). Default: "Today date" Format: YYYY-MM-DD datetime object. Note: Cannot be earlier than start date.

 required
aws_profile_name_input str

[REQUIRED] AWS credential profile name from local configuration. Default: "default"

 required
cost_groupby_key_input str

[REQUIRED] Dimension for cost aggregation. Valid values: Default: "LINKED_ACCOUNT" - 'LINKED_ACCOUNT' (default): Costs by AWS account - 'SERVICE': Costs by AWS service (e.g. EC2, S3) - 'PURCHASE_TYPE': Costs by purchase option - 'USAGE_TYPE': Costs by usage category - Composite keys (e.g. 'LINKED_ACCOUNT-With-SERVICE') - Composite keys (e.g. 'LINKED_ACCOUNT-With-PURCHASE_TYPE') - Composite keys (e.g. 'LINKED_ACCOUNT-With-USAGE_TYPE')

 required
granularity str

[REQUIRED] Time interval for cost breakdown: Default: 'MONTHLY' - 'MONTHLY' (default): Monthly aggregates - 'DAILY': Daily cost records

 required

Returns:

Type Description
List[_CostRecord]

List[_CostRecord]: Structured cost records containing: - TIME_PERIOD: Dict with 'Start'/'End' date strings - ID: Resource identifier (account, service, etc.) - GROUPBY_FILTER: Composite values. - COST: Unblended cost as string (USD)

Raises:

Type Description
ValueError

For invalid date ranges or parameters
ClientError

For AWS API authentication/access issues
DataNotAvailableError

If requested data exceeds retention period
Example

costs = monthly_account_cost_export( ... start_date_input="2023-01-01", ... end_date_input="2023-03-31", ... aws_profile_name_input="production", ... cost_groupby_key_input="SERVICE", ... granularity="MONTHLY" ... ) len(costs) > 0 True

Source code in src/eraXplor_aws/utils/cost_export_utils.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
def monthly_account_cost_export(
    start_date_input: Union[str, datetime],  # str | datetime
    end_date_input: Union[str, datetime],
    aws_profile_name_input: str,
    cost_groupby_key_input: str,
    granularity: str,
) -> List[_CostRecord]:
    """Retrieve AWS cost and usage data via Cost Explorer API.

    Fetches unblended costs across all linked accounts in an AWS organization, with flexible
    grouping and granularity options. Data is returned in standardized records suitable for
    analysis or export.

    Args:
        start_date_input (Union[str, datetime]): 
            [REQUIRED] Start date for cost report (inclusive).
            Default: "3 Months ago"
            Format: YYYY-MM-DD datetime object.
            Note: AWS limits historical data to 14 months.

        end_date_input (Union[str, datetime]): 
            [REQUIRED] End date for cost report (inclusive).
            Default: "Today date"
            Format: YYYY-MM-DD datetime object.
            Note: Cannot be earlier than start date.

        aws_profile_name_input (str): 
            [REQUIRED] AWS credential profile name from local configuration.
            Default: "default"

        cost_groupby_key_input (str): 
            [REQUIRED] Dimension for cost aggregation. Valid values:
            Default: "LINKED_ACCOUNT"
            - 'LINKED_ACCOUNT' (default): Costs by AWS account
            - 'SERVICE': Costs by AWS service (e.g. EC2, S3)
            - 'PURCHASE_TYPE': Costs by purchase option
            - 'USAGE_TYPE': Costs by usage category
            - Composite keys (e.g. 'LINKED_ACCOUNT-With-SERVICE')
            - Composite keys (e.g. 'LINKED_ACCOUNT-With-PURCHASE_TYPE')
            - Composite keys (e.g. 'LINKED_ACCOUNT-With-USAGE_TYPE')

        granularity (str): 
            [REQUIRED] Time interval for cost breakdown:
            Default: 'MONTHLY'
            - 'MONTHLY' (default): Monthly aggregates
            - 'DAILY': Daily cost records

    Returns:
        List[_CostRecord]: Structured cost records containing:
            - TIME_PERIOD: Dict with 'Start'/'End' date strings
            - ID: Resource identifier (account, service, etc.)
            - GROUPBY_FILTER: Composite values.
            - COST: Unblended cost as string (USD)

    Raises:
        ValueError: For invalid date ranges or parameters
        ClientError: For AWS API authentication/access issues
        DataNotAvailableError: If requested data exceeds retention period

    Example:
        >>> costs = monthly_account_cost_export(
        ...     start_date_input="2023-01-01",
        ...     end_date_input="2023-03-31",
        ...     aws_profile_name_input="production",
        ...     cost_groupby_key_input="SERVICE",
        ...     granularity="MONTHLY"
        ... )
        >>> len(costs) > 0
        True
    """

    _profile_session = boto3.Session(profile_name=str(aws_profile_name_input))
    _ce_client = _profile_session.client("ce")

    # if condition determine the type of groupby key
    results = []
    with Live(
        Spinner(
            "bouncingBar",
            text=f"Fetching AWS costs grouped by {cost_groupby_key_input}...\n\n",
        ),
        refresh_per_second=10,
    ):

        def _fetch_account():
            if cost_groupby_key_input == "LINKED_ACCOUNT-With-SERVICE":
                _account_cost_usage = _ce_client.get_cost_and_usage(
                    TimePeriod={
                        "Start": str(start_date_input),
                        "End": str(end_date_input),
                    },
                    # Granularity="MONTHLY",
                    Granularity=granularity,
                    Metrics=["UnblendedCost"],
                    GroupBy=[
                        {"Type": "DIMENSION", "Key": "LINKED_ACCOUNT"},
                        {"Type": "DIMENSION", "Key": "SERVICE"},
                    ],
                )
                for _item in _account_cost_usage["ResultsByTime"]:
                    time_period = _item["TimePeriod"]
                    for _group in _item["Groups"]:
                        ID = _group["Keys"][0]
                        service = _group["Keys"][1]
                        cost = float(_group["Metrics"]["UnblendedCost"]["Amount"])
                        currency = _group["Metrics"]["UnblendedCost"]["Unit"]
                        results.append(
                            {
                                "TIME_PERIOD": time_period,
                                "ID": ID,
                                "GROUPBY_FILTER": service,
                                "COST": f"{cost:.2f} {currency}",
                            }
                        )
            if cost_groupby_key_input == "LINKED_ACCOUNT-With-PURCHASE_TYPE":
                _account_cost_usage = _ce_client.get_cost_and_usage(
                    TimePeriod={
                        "Start": str(start_date_input),
                        "End": str(end_date_input),
                    },
                    # Granularity="MONTHLY",
                    Granularity=granularity,
                    Metrics=["UnblendedCost"],
                    GroupBy=[
                        {"Type": "DIMENSION", "Key": "LINKED_ACCOUNT"},
                        {"Type": "DIMENSION", "Key": "PURCHASE_TYPE"},
                    ],
                )
                for _item in _account_cost_usage["ResultsByTime"]:
                    time_period = _item["TimePeriod"]
                    for _group in _item["Groups"]:
                        ID = _group["Keys"][0]
                        purchase_type = _group["Keys"][1]
                        cost = float(_group["Metrics"]["UnblendedCost"]["Amount"])
                        currency = _group["Metrics"]["UnblendedCost"]["Unit"]
                        results.append(
                            {
                                "TIME_PERIOD": time_period,
                                "ID": ID,
                                "GROUPBY_FILTER": purchase_type,
                                "COST": f"{cost:.2f} {currency}",
                            }
                        )
            if cost_groupby_key_input == "LINKED_ACCOUNT-With-USAGE_TYPE":
                _account_cost_usage = _ce_client.get_cost_and_usage(
                    TimePeriod={
                        "Start": str(start_date_input),
                        "End": str(end_date_input),
                    },
                    # Granularity="MONTHLY",
                    Granularity=granularity,
                    Metrics=["UnblendedCost"],
                    GroupBy=[
                        {"Type": "DIMENSION", "Key": "LINKED_ACCOUNT"},
                        {"Type": "DIMENSION", "Key": "USAGE_TYPE"},
                    ],
                )
                for _item in _account_cost_usage["ResultsByTime"]:
                    time_period = _item["TimePeriod"]
                    for _group in _item["Groups"]:
                        ID = _group["Keys"][0]
                        usage_type = _group["Keys"][1]
                        cost = float(_group["Metrics"]["UnblendedCost"]["Amount"])
                        currency = _group["Metrics"]["UnblendedCost"]["Unit"]
                        results.append(
                            {
                                "TIME_PERIOD": time_period,
                                "ID": ID,
                                "GROUPBY_FILTER": usage_type,
                                "COST": f"{cost:.2f} {currency}",
                            }
                        )
            if cost_groupby_key_input == "LINKED_ACCOUNT":
                _account_cost_usage = _ce_client.get_cost_and_usage(
                    TimePeriod={
                        "Start": str(start_date_input),
                        "End": str(end_date_input),
                    },
                    # Granularity="MONTHLY",
                    Granularity=granularity,
                    Metrics=["UnblendedCost"],
                    GroupBy=[
                        {"Type": "DIMENSION", "Key": "LINKED_ACCOUNT"},
                    ],
                )
                for _item in _account_cost_usage["ResultsByTime"]:
                    time_period = _item["TimePeriod"]
                    for _group in _item["Groups"]:
                        ID = _group["Keys"][0]
                        # usage_type = group["Keys"][1]
                        cost = float(_group["Metrics"]["UnblendedCost"]["Amount"])
                        currency = _group["Metrics"]["UnblendedCost"]["Unit"]
                        results.append(
                            {
                                "TIME_PERIOD": time_period,
                                "ID": ID,
                                "GROUPBY_FILTER": "NONE",
                                "COST": f"{cost:.2f} {currency}",
                            }
                        )
            if cost_groupby_key_input == "SERVICE":
                _account_cost_usage = _ce_client.get_cost_and_usage(
                    TimePeriod={
                        "Start": str(start_date_input),
                        "End": str(end_date_input),
                    },
                    # Granularity="MONTHLY",
                    Granularity=granularity,
                    Metrics=["UnblendedCost"],
                    GroupBy=[
                        {"Type": "DIMENSION", "Key": "SERVICE"},
                    ],
                )
                for _item in _account_cost_usage["ResultsByTime"]:
                    time_period = _item["TimePeriod"]
                    for _group in _item["Groups"]:
                        ID = _group["Keys"][0]
                        # usage_type = group["Keys"][1]
                        cost = float(_group["Metrics"]["UnblendedCost"]["Amount"])
                        currency = _group["Metrics"]["UnblendedCost"]["Unit"]
                        results.append(
                            {
                                "TIME_PERIOD": time_period,
                                "ID": ID,
                                "GROUPBY_FILTER": "NONE",
                                "COST": f"{cost:.2f} {currency}",
                            }
                        )
            if cost_groupby_key_input == "PURCHASE_TYPE":
                _account_cost_usage = _ce_client.get_cost_and_usage(
                    TimePeriod={
                        "Start": str(start_date_input),
                        "End": str(end_date_input),
                    },
                    # Granularity="MONTHLY",
                    Granularity=granularity,
                    Metrics=["UnblendedCost"],
                    GroupBy=[
                        {"Type": "DIMENSION", "Key": "PURCHASE_TYPE"},
                    ],
                )
                for _item in _account_cost_usage["ResultsByTime"]:
                    time_period = _item["TimePeriod"]
                    for _group in _item["Groups"]:
                        ID = _group["Keys"][0]
                        # usage_type = group["Keys"][1]
                        cost = float(_group["Metrics"]["UnblendedCost"]["Amount"])
                        currency = _group["Metrics"]["UnblendedCost"]["Unit"]
                        results.append(
                            {
                                "TIME_PERIOD": time_period,
                                "ID": ID,
                                "GROUPBY_FILTER": "NONE",
                                "COST": f"{cost:.2f} {currency}",
                            }
                        )
            if cost_groupby_key_input == "USAGE_TYPE":
                _account_cost_usage = _ce_client.get_cost_and_usage(
                    TimePeriod={
                        "Start": str(start_date_input),
                        "End": str(end_date_input),
                    },
                    # Granularity="MONTHLY",
                    Granularity=granularity,
                    Metrics=["UnblendedCost"],
                    GroupBy=[
                        {"Type": "DIMENSION", "Key": "USAGE_TYPE"},
                    ],
                )
                for _item in _account_cost_usage["ResultsByTime"]:
                    time_period = _item["TimePeriod"]
                    for _group in _item["Groups"]:
                        ID = _group["Keys"][0]
                        # usage_type = group["Keys"][1]
                        cost = float(_group["Metrics"]["UnblendedCost"]["Amount"])
                        currency = _group["Metrics"]["UnblendedCost"]["Unit"]
                        results.append(
                            {
                                "TIME_PERIOD": time_period,
                                "ID": ID,
                                "GROUPBY_FILTER": "NONE",
                                "COST": f"{cost:.2f} {currency}",
                            }
                        )
        # progress.update(task, advance=1)
        _thread = threading.Thread(target=_fetch_account)
        _thread.start()
        _thread.join()
    return results

Contains functions for retrieving cost and usage reports from AWS Cost Explorer using boto3, grouped by various dimensions such as:

  • Linked AWS accounts
  • AWS services
  • Purchase types
  • Usage types

🧾 CSV Export Utilities

Module for exporting AWS cost data to CSV format.

csv_export(results, filename)

Exports AWS cost data to a CSV file with standardized formatting.

Takes the output from monthly_account_cost_export() (i.e. depends handle by main) and writes it to a CSV file with consistent column headers and proper formatting. The CSV will contain the time period, Account/Service/Purchase_type/Usage_type, and associated costs.

Parameters:

Name Type Description Default
fetch_monthly_account_cost_usage list

List of cost data dictionaries as returned by monthly_account_cost_export(). Each dictionary should contain: - time_period (dict): With 'Start' and 'End' keys - ID : AWS account ID, service name, purchase type name, usage type name. - GROUPBY_FILTER (str): Grouping filter used in the query, e.g., 'Account', 'Service', etc. - COST (float): The cost associated with the ID for the given time period.

 required
filename str

Output filename for the CSV. Defaults to 'cost_report.csv'.

 required

Returns:

Name Type Description
None None

Writes directly to file but doesn't return any value.
Source code in src/eraXplor_aws/utils/csv_export_utils.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def csv_export(
    results: List[Dict[str, Any]],
    filename: str
    ) -> None:
    """Exports AWS cost data to a CSV file with standardized formatting.

    Takes the output from monthly_account_cost_export() _(i.e. depends handle by main)_
    and writes it to a CSV file with consistent column headers and proper formatting.
    The CSV will contain the time period, Account/Service/Purchase_type/Usage_type,
    and associated costs.

    Args:
        fetch_monthly_account_cost_usage (list): List of cost data dictionaries as returned
            by monthly_account_cost_export(). Each dictionary should contain:
            - time_period (dict): With 'Start' and 'End' keys
            - ID : AWS account ID, service name, purchase type name, usage type name.
            - GROUPBY_FILTER (str): Grouping filter used in the query, e.g., 'Account', 'Service', etc.
            - COST (float): The cost associated with the ID for the given time period.

        filename (str, optional): Output filename for the CSV. Defaults to 'cost_report.csv'.

    Returns:
        None: Writes directly to file but doesn't return any value.
    """
    # Create a CSV file with write mode
    with open(filename, mode="w", newline="", encoding="utf-8") as _csvfile:
        writer = csv.writer(_csvfile)
        writer.writerow(
            [
                "Start Date",
                "End Date",
                "ID",
                "GROUPBY_FILTER",
                "Cost",
            ]
        )
        for _row in results:
            time_period = _row["TIME_PERIOD"]
            ID = _row.get("ID")
            groupby = _row.get("GROUPBY_FILTER")
            cost = _row.get("COST")
            writer.writerow([time_period["Start"], time_period["End"], ID, groupby, cost])
    print(f"\n✅ Data exported to {filename}")

Provides functionality to export retrieved cost data into a structured CSV format.

📅 Date Utilities

Module providing date utility functions.

get_end_date_from_user()

Prompts the user to enter an end date and validates the input format.

Continuously prompts the user up to 4 times until a valid date is provided in the specified format or until the user interrupts with keyboard input. Handles both format validation and user interruption gracefully.

Returns:

Type Description

datetime.date, 'Too many invalid attempts', or None:

Returns a date object if valid input is provided, returns None if the user

interrupts the input (Ctrl+C).

Raises:

Type Description
ValueError

If the input date format is invalid.
KeyboardInterrupt

If the user interrupts the input prompt (though this is caught and handled within the function).
Source code in src/eraXplor_aws/utils/date_utils.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def get_end_date_from_user():
    """Prompts the user to enter an end date and validates the input format.

    Continuously prompts the user up to 4 times until a valid date is provided in the specified
    format or until the user interrupts with keyboard input. Handles both format
    validation and user interruption gracefully.

    Returns:
        datetime.date, 'Too many invalid attempts', or None: 
        Returns a date object if valid input is provided, returns None if the user 
        interrupts the input (Ctrl+C).

    Raises:
        ValueError: If the input date format is invalid.

        KeyboardInterrupt: If the user interrupts the input prompt (though this is
            caught and handled within the function).
    """
    attempts = 0
    while attempts < 4:
        try:
            date_string = input("Enter an end date value with YYYY-MM-DD format: ")
            date_object = datetime.strptime(date_string, "%Y-%m-%d").date()
            return date_object
        except ValueError:
            print("Invalid date format, Please use YYYY-MM-DD")
        except KeyboardInterrupt:
            print("\nUser interrupted. Exiting")
            break
        attempts += 1
    print("Too many invalid attempts. Exiting.")
    return None

get_start_date_from_user()

Prompts the user to enter a start date and validates the input format.

Continuously prompts the user up to 4 times until a valid date is provided in the specified format or until the user interrupts with keyboard input. Handles both format validation and user interruption gracefully.

Returns:

Type Description

datetime.date, 'Too many invalid attempts', or None:

Returns a date object if valid input is provided, returns None if

the user interrupts the input (Ctrl+C).

Raises:

Type Description
ValueError

If the input date format is invalid.
KeyboardInterrupt

If the user interrupts the input prompt (though this is caught and handled within the function).
Source code in src/eraXplor_aws/utils/date_utils.py 
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
def get_start_date_from_user():
    """Prompts the user to enter a start date and validates the input format.

    Continuously prompts the user up to 4 times until a valid date is provided in the specified
    format or until the user interrupts with keyboard input. Handles both format
    validation and user interruption gracefully.

    Returns:
        datetime.date, 'Too many invalid attempts', or None: 
        Returns a date object if valid input is provided, returns None if 
        the user interrupts the input (Ctrl+C).

    Raises:
        ValueError: If the input date format is invalid.

        KeyboardInterrupt: If the user interrupts the input prompt (though this is
            caught and handled within the function).
    """
    attempts = 0
    while attempts < 4:
        try:
            date_string = input("Enter a start date value with YYYY-MM-DD format: ")
            date_object = datetime.strptime(date_string, "%Y-%m-%d").date()
            return date_object
        except ValueError:
            print("Invalid date format, Please use YYYY-MM-DD")
        except KeyboardInterrupt:
            print("\nUser interrupted. Exiting")
            break
        attempts +=1
    print("Too many invalid attempts. Exiting.")
    return None

Includes interactive functions for prompting and validating date input from users, ensuring format compliance and error handling.

Azure - Reference

This section provides a structured breakdown of the main application module and its supporting utilities used in the eraXplor_azure project.

🌟eraXplor source code

🔹Main Application Module

▶️ Entry Point

eraXplor - Azure Cost Export Tool

This is the main entry point for the eraXplor_azure CLI tool, which allows users to export Azure cost and usage data using Azure CostManagementClient client.

Command Line Arguments

--start-date, -s DATE Start date in YYYY,MM,DD format. Default: 3 months prior

--end-date, -e DATE End date in YYYY,MM,DD format. Default: Today date.

--subscription-id, -S SUBSCRIPTION_ID Azure subscription ID for cost export.

--out, -o FILENAME Output CSV filename. Default: az_cost_report.csv

--granularity, -G GRANULARITY Time granularity. Options: - Monthly (default) - Daily

Examples:

  1. Basic usage with default settings: eraXplor-azure -S SUBSCRIPTION_ID

  2. Custom date range and profile: eraXplor-azure -s 2025,01,01 -e 2025,03,30 -S SUBSCRIPTION_ID

Notes
  • Ensure that the environment is properly authenticated with Azure using DefaultAzureCredential.
  • Date strings must follow the exact "YYYY,MM,DD" format to avoid parsing errors.
  • Depending on the size of the date range and granularity, response time may vary.

main()

Orchestrates & Manage depends of cost export workflow.

Source code in src/eraXplor_azure/__main__.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def main() -> None:
    """Orchestrates & Manage depends of cost export workflow."""

    # Banner
    banner_format, copyright_notice = generate_banner()
    print(f"\n\n {termcolor.colored(banner_format, color="green")}")
    print(f"{termcolor.colored(copyright_notice, color="green")}", end="\n\n")

    # Fetch Parsed parameters by command line
    arg_parser = parser().parse_args()
    start_date_input = arg_parser.start_date
    end_date_input = arg_parser.end_date
    subscription_id_input = arg_parser.subscription_id
    granularity_input = arg_parser.granularity
    filename_input = arg_parser.out

    subscriptions_list_detailed, subscriptions_with_tags_list = subs_cost_export()

    # Parsing data to cost export func
    cm_client_query_results = cost_export(
        subscription_id=subscription_id_input,
        subscriptions_list_detailed=subscriptions_list_detailed,
        start_date=start_date_input,
        end_date=end_date_input,
        granularity=granularity_input,
    )

    # print(json.dumps(cm_client_query_results, indent=4, default=str), end="\n\n\n")

    csv_export(cm_client_query_results=cm_client_query_results, filename=filename_input)

This is the primary script responsible for orchestrating the user workflow. It handles user input, invokes Azure cost data retrieval, and manages data export functionality.

🛠 Utility Modules

Module to display a banner and copyright notice.

banner()

Generates a banner and copyright notice for the application.

Source code in src/eraXplor_azure/utils/banner_utils.py 
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
def banner():
    """Generates a banner and copyright notice for the application."""

    copyright_notice = """╔══════════════════════════════════════════════════╗
║  © 2025 Mohamed eraki                            ║
║  mohamed-ibrahim2021@outlook.com                 ║
║  Version: 3.2.0                                  ║
║  eraXplor - Azure Cost exporter Tool             ║
╚══════════════════════════════════════════════════╝
    """
    banner_format = pyfiglet.figlet_format("eraXplor", font='slant')
    return banner_format, copyright_notice

Responsible for rendering styled ASCII banners and displaying copyright information used in the CLI interface.

📊 Cost Export Utilities

Module for exporting Azure cost data using the Azure Cost Management API.

cost_export(subscription_id=None, subscriptions_list_detailed=None, start_date=None, end_date=None, granularity='Monthly')

Retrieve Azure cost data for a given subscription and time range.

Executes a cost management query using the Azure Cost Management API to extract cost data for a specific subscription, aggregated by the selected granularity (Daily or Monthly).

Parameters:

Name Type Description Default
subscription_id str

[REQUIRED] Azure subscription ID to query cost data for.

 None
start_date str

[REQUIRED] Start date of the report period (inclusive). Format: "YYYY,MM,DD" Default: 3 Months ago from today.

 None
end_date str

[REQUIRED] End date of the report period (inclusive). Format: "YYYY,MM,DD" Default: Today date.

 None
granularity str

[OPTIONAL] Level of time granularity for aggregation. Default: "Monthly" Valid values: - "Daily": Daily cost records - "Monthly": Monthly aggregated cost records

 'Monthly'

Returns:

Type Description
List[_CostRecord]

List[_CostRecord]: A list of structured cost records, where each record contains: - TIME_PERIOD (Dict[str, str]): Date or date range for the record. - COST (str): Formatted string representing the total cost and currency (e.g., "123.45 USD").

Raises:

Type Description
Exception

For any API errors or authentication failures.
Example

cost_export( ... subscription_id="SUB_ID", ... start_date="2025,01,01", ... end_date="2025,04,30", ... granularity="Monthly" ... ) [{'TIME_PERIOD': {'Start': '2025-01-01', 'End': '2025-01-31'}, 'COST': '456.78 USD'}, ...]

Notes
  • Ensure that the environment is properly authenticated with Azure using DefaultAzureCredential.
  • Date strings must follow the exact "YYYY,MM,DD" format to avoid parsing errors.
  • Depending on the size of the date range and granularity, response time may vary.
Source code in src/eraXplor_azure/utils/cost_export_utils.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
def cost_export(
    subscription_id: str | None = None, 
    subscriptions_list_detailed: List[dict[str, Any]] = None, 
    start_date: str = None, 
    end_date: str = None,
    granularity: str = 'Monthly',
) -> List[_CostRecord]:
    """
        Retrieve Azure cost data for a given subscription and time range.

        Executes a cost management query using the Azure Cost Management API to 
        extract cost data for a specific subscription, aggregated by the 
        selected granularity (Daily or Monthly).

        Args:
            subscription_id (str): 
                [REQUIRED] Azure subscription ID to query cost data for.

            start_date (str): 
                [REQUIRED] Start date of the report period (inclusive).
                Format: "YYYY,MM,DD"
                Default: 3 Months ago from today.

            end_date (str): 
                [REQUIRED] End date of the report period (inclusive).
                Format: "YYYY,MM,DD"
                Default: Today date.

            granularity (str): 
                [OPTIONAL] Level of time granularity for aggregation.
                Default: "Monthly"
                Valid values:
                    - "Daily": Daily cost records
                    - "Monthly": Monthly aggregated cost records

        Returns:
            List[_CostRecord]: 
                A list of structured cost records, where each record contains:
                    - TIME_PERIOD (Dict[str, str]): Date or date range for the record.
                    - COST (str): Formatted string representing the total cost and currency (e.g., "123.45 USD").

        Raises:
            Exception: For any API errors or authentication failures.

        Example:
            >>> cost_export(
            ...     subscription_id="SUB_ID",
            ...     start_date="2025,01,01",
            ...     end_date="2025,04,30",
            ...     granularity="Monthly"
            ... )
            [{'TIME_PERIOD': {'Start': '2025-01-01', 'End': '2025-01-31'}, 'COST': '456.78 USD'}, ...]

        Notes:
            - Ensure that the environment is properly authenticated with Azure using `DefaultAzureCredential`.
            - Date strings must follow the exact "YYYY,MM,DD" format to avoid parsing errors.
            - Depending on the size of the date range and granularity, response time may vary.
        """

    credential = DefaultAzureCredential()
    cm_client = CostManagementClient(credential)
    cm_client_query_results = []

    if subscription_id is not None:
        start_date = datetime.datetime.strptime(start_date, "%Y,%m,%d")
        end_date = datetime.datetime.strptime(end_date, "%Y,%m,%d")
        scope = f"/subscriptions/{subscription_id}"
        # scope = f"/subscriptions/{subscriptionId}/resourceGroups/{resourceGroupName}"
        # scope = f"/providers/Microsoft.Billing/billingAccounts/{billingAccountId}"
        # scope = f"/providers/Microsoft.Billing/billingAccounts/{billingAccountId}/departments/{departmentId}"
        # scope = f"/providers/Microsoft.Billing/billingAccounts/{billingAccountId}/enrollmentAccounts/{enrollmentAccountId}"
        # scope = f"/providers/Microsoft.Management/managementGroups/{managementGroupId}"
        # scope = f"/providers/Microsoft.Billing/billingAccounts/{billingAccountId}/billingProfiles/{billingProfileId}"
        # scope = f"/providers/Microsoft.Billing/billingAccounts/{billingAccountId}/billingProfiles/{billingProfileId}/invoiceSections/{invoiceSectionId}"
        # scope = f"/providers/Microsoft.Billing/billingAccounts/{billingAccountId}/customers/{customerId}"

        with Live(Spinner
                ("bouncingBar", text=f"Fetching Azure costs of subscriptions: {subscription_id}...\n\n"),
                    refresh_per_second=10):
            def _sub_cost_export():
                """Internal function to handle the cost export query."""

                try:
                    cm_client_query = cm_client.query.usage(
                        scope=scope,
                        parameters=models.QueryDefinition(
                            type='Usage',
                            timeframe='Custom',
                            time_period=models.QueryTimePeriod(
                                from_property=start_date,
                                to=end_date,
                            ),
                            dataset=models.QueryDataset(
                                granularity=granularity,
                                aggregation={
                                    'totalcost': models.QueryAggregation(name='PreTaxCost', function='Sum')
                                }
                            )
                        )
                    )
                    cm_client_query_rows = cm_client_query.rows
                    for row in cm_client_query_rows:
                        time_period = row[1]
                        cost = row[0]
                        currency = row[2]

                        cm_client_query_results.append(
                            {
                                "TIME_PERIOD": time_period,
                                "SUBSCRIPTION_ID": subscription_id,
                                "DISPLAY_NAME": "NONE",
                                "COST": f"{cost:.2f} {currency}",
                                "TAGS": "None"
                            }
                        )
                    print(json.dumps(cm_client_query_results, indent=4, default=str), end="\n\n\n")
                except Exception as e:
                    print(f"An error occurred: {e}")
                    return {"error": str(e)}

            # progress.update(task, advance=1)
            _thread = threading.Thread(target=_sub_cost_export)
            _thread.start()
            _thread.join()

    if subscription_id is None:
        start_date = datetime.datetime.strptime(start_date, "%Y,%m,%d")
        end_date = datetime.datetime.strptime(end_date, "%Y,%m,%d")

        for sub in subscriptions_list_detailed:
            subscription_id = sub['Subscription_ID']
            subscription_name = sub['Display_Name']
            subscription_tags = sub.get('Tags', {})
            scope = f"/subscriptions/{subscription_id}"

            with Live(Spinner
                    ("bouncingBar", text=f"Fetching Azure costs of subscriptions: {subscription_name}({subscription_id})...\n\n"),
                        refresh_per_second=10):
                def _sub_cost_export():
                    """Internal function to handle the cost export query."""

                    try:
                        cm_client_query = cm_client.query.usage(
                            scope=scope,
                            parameters=models.QueryDefinition(
                                type='Usage',
                                timeframe='Custom',
                                time_period=models.QueryTimePeriod(
                                    from_property=start_date,
                                    to=end_date,
                                ),
                                dataset=models.QueryDataset(
                                    granularity=granularity,
                                    aggregation={
                                        'totalcost': models.QueryAggregation(name='PreTaxCost', function='Sum')
                                    }
                                )
                            )
                        )
                        cm_client_query_rows = cm_client_query.rows
                        for row in cm_client_query_rows:
                            time_period = row[1]
                            cost = row[0]
                            currency = row[2]

                            cm_client_query_results.append(
                                {
                                    "TIME_PERIOD": time_period,
                                    "SUBSCRIPTION_ID": subscription_id,
                                    "DISPLAY_NAME": subscription_name,
                                    "COST": f"{cost:.2f} {currency}",
                                    "TAGS": subscription_tags if subscription_tags else "None"
                                }
                            )
                            # print(json.dumps(cm_client_query_results, indent=4, default=str), end="\n\n\n")
                        # Combine results of for loop and print
                        print(json.dumps([
                            {
                                "TIME_PERIOD": row[1],
                                "SUBSCRIPTION_ID": subscription_id,
                                "DISPLAY_NAME": subscription_name,
                                "COST": f"{row[0]:.2f} {row[2]}",
                                "TAGS": subscription_tags if subscription_tags else "None",
                            } for row in cm_client_query_rows
                        ], indent=4, default=str), end="\n\n\n")

                    except Exception as e:
                        print(f"An error occurred: {e}")
                        return {"error": str(e)}

                # progress.update(task, advance=1)
                _thread = threading.Thread(target=_sub_cost_export)
                _thread.start()
                _thread.join()

    return cm_client_query_results

subs_cost_export()

Function to retrieve Azure subscription details

Source code in src/eraXplor_azure/utils/cost_export_utils.py 
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
def subs_cost_export():
    """Function to retrieve Azure subscription details"""
    _credential = DefaultAzureCredential()
    _subscription_client = SubscriptionClient(_credential)
    _subscriptions = list(_subscription_client.subscriptions.list())

    subscriptions_list_detailed = []
    subscriptions_with_tags_list = []

    for sub in _subscriptions:
        subscriptions_list_detailed.append(
            {
                "Subscription_ID": sub.subscription_id,
                "Display_Name": sub.display_name,
                "Tenant_ID": sub.tenant_id,
                "Tags": sub.tags,
            }
        )
        subscriptions_with_tags_list.append(
            {
                "Subscription_ID": sub.subscription_id,
                "Display_Name": sub.display_name,
                "Tags": sub.tags,
            }
        )
    return subscriptions_list_detailed, subscriptions_with_tags_list

Contains functions for retrieving cost and usage reports from Azure Cost Explorer using CostManagementClient

🧾 CSV Export Utilities

Module for exporting Azure cost data to CSV format.

csv_export(cm_client_query_results, filename)

Exports Azure cost data to a CSV file with standardized formatting.

Takes the output from cost_export() (i.e. depends handle by main) and writes it to a CSV file with consistent column headers and proper formatting. The CSV will contain the time period, and associated costs.

Parameters:

Name Type Description Default
cm_client_query_results list

List of cost data dictionaries as returned by cost_export(). Each dictionary should contain: - time_period (dict): date as string. - COST (str): Cost amount as string.

 required
filename str

Output filename for the CSV. Defaults to 'az_cost_report.csv'.

 required

Returns:

Name Type Description
None None

Writes directly to file but doesn't return any value.
Source code in src/eraXplor_azure/utils/csv_export_utils.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def csv_export(
    cm_client_query_results: List[Dict[str, Any]],
    filename: str,
    ) -> None:
    """Exports Azure cost data to a CSV file with standardized formatting.

    Takes the output from cost_export() _(i.e. depends handle by main)_
    and writes it to a CSV file with consistent column headers and proper formatting.
    The CSV will contain the time period, and associated costs.

    Args:
        cm_client_query_results (list): List of cost data dictionaries as returned
            by cost_export(). Each dictionary should contain:
            - time_period (dict): date as string.
            - COST (str): Cost amount as string.

        filename (str, optional): Output filename for the CSV. Defaults to 'az_cost_report.csv'.

    Returns:
        None: Writes directly to file but doesn't return any value.
    """
    # Create a CSV file with write mode
    with open(filename, mode="w", newline="", encoding="utf-8") as csvfile:
        writer = csv.writer(csvfile)
        writer.writerow(
            [
                "TIME_PERIOD",
                "SUBSCRIPTION_ID",
                "DISPLAY_NAME",
                "COST",
                "TAGS",
            ]
        )
        for row in cm_client_query_results:
            time_period = row["TIME_PERIOD"]
            sub_id = row["SUBSCRIPTION_ID"]
            display_name = row["DISPLAY_NAME"]
            cost = row.get("COST")
            tags = row.get("TAGS", {})
            writer.writerow([time_period, sub_id, display_name, cost, tags])
    print(f"\n✅ Data exported to {filename}")

Provides functionality to export retrieved cost data into a structured CSV format.