Skip to content

Azure - Reference

This section documents the current Azure FOCUS workflow modules used by eraXplor.

eraXplor source code

Main Application Module

Entry Point

Azure service package for the eraXplor FOCUS workflow.

This package exposes the Azure CLI entry point implemented in core.services.azure.__main__ and the supporting utility modules under core.services.azure.utils.

main()

Orchestrates and manage the cost export workflow.

This function serves as the main entry point for the eraXplor_azure CLI tool. It coordinates the entire cost export process by: 1. Displaying the application banner with version information 2. Parsing command-line arguments for configuration 3. Retrieving all accessible Azure subscriptions 4. Fetching cost data using the Azure Cost Management API 5. Exporting the results to a CSV file

The function uses the following workflow
  • generate_banner(): Displays the eraXplor banner
  • parser(): Parses CLI arguments
  • list_subs(): Retrieves subscription details
  • cost_export(): Fetches cost data for all subscriptions
  • csv_export(): Writes results to CSV format

Returns:

Name Type Description
None None

This function does not return a value. It prints output directly to the console and writes the cost report to a CSV file.
Example

if name == "main": ... main()

Note

This function is typically called from the command line and should not be imported directly for programmatic use. For programmatic use, import and call the individual utility functions directly.

Source code in src/core/services/azure/__main__.py 
 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
def main() -> None:
    """
    Orchestrates and manage the cost export workflow.

    This function serves as the main entry point for the eraXplor_azure CLI tool.
    It coordinates the entire cost export process by:
    1. Displaying the application banner with version information
    2. Parsing command-line arguments for configuration
    3. Retrieving all accessible Azure subscriptions
    4. Fetching cost data using the Azure Cost Management API
    5. Exporting the results to a CSV file

    The function uses the following workflow:
        - generate_banner(): Displays the eraXplor banner
        - parser(): Parses CLI arguments
        - list_subs(): Retrieves subscription details
        - cost_export(): Fetches cost data for all subscriptions
        - csv_export(): Writes results to CSV format

    Returns:
        None: This function does not return a value. It prints output directly
              to the console and writes the cost report to a CSV file.

    Raises:
        Any exceptions raised by the underlying Azure SDK calls or file I/O
        operations are propagated to the caller.

    Example:
        >>> if __name__ == "__main__":
        ...     main()

    Note:
        This function is typically called from the command line and should
        not be imported directly for programmatic use. For programmatic use,
        import and call the individual utility functions directly.
    """

    # 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()
    rg_name_input = arg_parser.resource_group_name
    location_input = arg_parser.location
    storage_account_name_input = arg_parser.storage_account_name
    container_name_input = arg_parser.container_name
    folder_name_input = arg_parser.folder_name
    subscription_id_input = arg_parser.subscription_id

    run_backend = arg_parser.all or arg_parser.backend
    run_export = arg_parser.all or arg_parser.export_config
    run_fetch = arg_parser.all or arg_parser.fetch

    if not (run_backend or run_export or run_fetch):
        run_backend = run_export = run_fetch = True

    if run_backend:
        print("\n=== Running backend provisioning stage ===")
        create_resource_group(
            resource_group_name=rg_name_input,
            location=location_input,
            subscription_id=subscription_id_input,
        )
        create_storage_account_container_folder(
            resource_group_name=rg_name_input,
            location=location_input,
            storage_account_name=storage_account_name_input,
            container_name=container_name_input,
            folder_name=folder_name_input,
            subscription_id=subscription_id_input,
        )

    if run_export:
        print("\n=== Running FOCUS export creation stage ===")
        billing_ids = get_default_billing_account_and_profile_ids()
        create_focus_export(
            billing_account_id=billing_ids["billing_account_id"],
            billing_profile_id=billing_ids["billing_profile_id"],
            subscription_id=subscription_id_input,
            resource_group_name=rg_name_input,
            storage_account_name=storage_account_name_input,
            container_name=container_name_input,
            folder_name=folder_name_input,
        )

    if run_fetch:
        print("\n=== Running fetch stage for Parquet files ===")
        download_parquet_files(
            storage_account_name=storage_account_name_input,
            container_name=container_name_input,
            folder_name=folder_name_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

Shared utility helpers for eraXplor service packages.

banner()

Generates a banner and copyright notice for the eraXplor application.

Creates an ASCII art banner using the 'slant' font with the application name, along with a formatted copyright notice containing version information and contact details.

Returns:

Name Type Description
tuple callable

A tuple containing two strings: - banner_format (str): ASCII art banner with the text "eraXplor" - copyright_notice (str): Formatted copyright and version information
Example

banner_format, copyright_notice = banner() print(banner_format) print(copyright_notice)

Note

The copyright year is currently set to 2025 and should be updated annually. The version number reflects the current release version of the eraXplor package.

Source code in src/core/services/utils/banner_utils.py 
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
def banner():
    """
    Generates a banner and copyright notice for the eraXplor application.

    Creates an ASCII art banner using the 'slant' font with the application name,
    along with a formatted copyright notice containing version information and
    contact details.

    Returns:
        tuple: A tuple containing two strings:
            - banner_format (str): ASCII art banner with the text "eraXplor"
            - copyright_notice (str): Formatted copyright and version information

    Example:
        >>> banner_format, copyright_notice = banner()
        >>> print(banner_format)
        >>> print(copyright_notice)

    Note:
        The copyright year is currently set to 2025 and should be updated
        annually. The version number reflects the current release version
        of the eraXplor package.
    """

    copyright_notice = """╔══════════════════════════════════════════════════╗
║  © 2026 Mohamed Eraki                            ║
║  mohamed-ibrahim2021@outlook.com                 ║
║  Version: 4.0.0                                  ║
║  eraXplor - FinOps 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.

This module provides functionality to query and retrieve Azure cost and usage data using the Azure Cost Management API. It supports multiple grouping dimensions including subscription, ServiceName, and ResourceGroupName, with both daily and monthly granularity options.

The module includes
  • cost_export: Main function to fetch cost data across all subscriptions
  • list_subs: Retrieves details of all accessible Azure subscriptions
  • _subs_cost_export: Internal function for subscription-level cost export
  • _cost_export_subfunc: Internal function for dimension-based cost export
Dependencies
  • azure-identity: For DefaultAzureCredential authentication
  • azure-mgmt-costmanagement: For Cost Management API access
  • azure-mgmt-resource: For Subscription Client access
  • rich: For live progress display
Example

from eraXplor_azure.utils.cost_export_utils import cost_export, list_subs subscriptions = list_subs() cost_data = cost_export( ... group_by='subscription', ... subscriptions_list_detailed=subscriptions, ... start_date='2025,01,01', ... end_date='2025,01,31', ... granularity='Monthly' ... )

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

Retrieve Azure cost data for all subscriptions over a specified time range.

Executes cost management queries using the Azure Cost Management API to extract cost data for all accessible subscriptions, aggregated by the selected dimension and granularity (Daily or Monthly).

This is the main entry point for fetching cost data. The function delegates to internal helper functions based on the group_by parameter: - 'subscription': Uses _subs_cost_export for per-subscription breakdown - 'ServiceName' or 'ResourceGroupName': Uses _cost_export_subfunc

Parameters:

Name Type Description Default
group_by str

Dimension to group costs by. Valid values: - 'subscription' (default): Group by Azure subscription - 'ServiceName': Group by Azure service name - 'ResourceGroupName': Group by resource group

 'subscription'
subscriptions_list_detailed List[dict[str, Any]]

List of subscription dictionaries as returned by list_subs(). Each dictionary should contain 'Subscription_ID', 'Display_Name', and 'Tags'. If None, the function will attempt to retrieve subscriptions automatically.

 None
start_date str

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

 None
end_date str

End date of the report period (inclusive). Format: "YYYY,MM,DD" Default: Today's date.

 None
granularity str

Level of time granularity for aggregation. Valid values: - 'Monthly' (default): Monthly aggregated cost records - 'Daily': Daily cost records

 'Monthly'

Returns:

Type Description
List[_CostRecord]

List[_CostRecord]: A list of structured cost records, where each record contains: - TIME_PERIOD: Date or date range (dict with 'Start'/'End' keys for monthly, string for daily) - GROUP_BY: The grouping dimension used - SUBSCRIPTION_ID: Azure subscription ID - DISPLAY_NAME: Subscription display name - PreTaxCost: Formatted cost string with currency (e.g. "123.45 USD") - TAGS: Dictionary of subscription tags or "None"

Raises:

Type Description
AzureError

For Azure API errors.
Exception

For any authentication failures or network issues.
Example

from eraXplor_azure.utils.cost_export_utils import cost_export, list_subs subs = list_subs() costs = cost_export( ... group_by='subscription', ... subscriptions_list_detailed=subs, ... start_date='2025,01,01', ... end_date='2025,01,31', ... granularity='Monthly' ... ) for record in costs: ... print(f"{record['DISPLAY_NAME']}: {record['PreTaxCost']}")

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.
  • The function displays progress using rich.live for real-time feedback.
Source code in src/core/services/azure/utils/cost_export_utils.py 
 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
def cost_export(
    group_by: str = 'subscription',
    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 all subscriptions over a specified time range.

    Executes cost management queries using the Azure Cost Management API to extract
    cost data for all accessible subscriptions, aggregated by the selected dimension
    and granularity (Daily or Monthly).

    This is the main entry point for fetching cost data. The function delegates to
    internal helper functions based on the group_by parameter:
        - 'subscription': Uses _subs_cost_export for per-subscription breakdown
        - 'ServiceName' or 'ResourceGroupName': Uses _cost_export_subfunc

    Args:
        group_by (str, optional):
            Dimension to group costs by. Valid values:
            - 'subscription' (default): Group by Azure subscription
            - 'ServiceName': Group by Azure service name
            - 'ResourceGroupName': Group by resource group

        subscriptions_list_detailed (List[dict[str, Any]], optional):
            List of subscription dictionaries as returned by list_subs().
            Each dictionary should contain 'Subscription_ID', 'Display_Name',
            and 'Tags'. If None, the function will attempt to retrieve
            subscriptions automatically.

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

        end_date (str, optional):
            End date of the report period (inclusive).
            Format: "YYYY,MM,DD"
            Default: Today's date.

        granularity (str, optional):
            Level of time granularity for aggregation. Valid values:
            - 'Monthly' (default): Monthly aggregated cost records
            - 'Daily': Daily cost records

    Returns:
        List[_CostRecord]:
            A list of structured cost records, where each record contains:
            - TIME_PERIOD: Date or date range (dict with 'Start'/'End' keys for monthly,
              string for daily)
            - GROUP_BY: The grouping dimension used
            - SUBSCRIPTION_ID: Azure subscription ID
            - DISPLAY_NAME: Subscription display name
            - PreTaxCost: Formatted cost string with currency (e.g. "123.45 USD")
            - TAGS: Dictionary of subscription tags or "None"

    Raises:
        azure.core.exceptions.AzureError: For Azure API errors.
        Exception: For any authentication failures or network issues.

    Example:
        >>> from eraXplor_azure.utils.cost_export_utils import cost_export, list_subs
        >>> subs = list_subs()
        >>> costs = cost_export(
        ...     group_by='subscription',
        ...     subscriptions_list_detailed=subs,
        ...     start_date='2025,01,01',
        ...     end_date='2025,01,31',
        ...     granularity='Monthly'
        ... )
        >>> for record in costs:
        ...     print(f"{record['DISPLAY_NAME']}: {record['PreTaxCost']}")

    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.
        - The function displays progress using rich.live for real-time feedback.
    """

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

    if group_by == 'subscription':
        _subs_cost_export(
            group_by=group_by,
            subscriptions_list_detailed=subscriptions_list_detailed,
            start_date=start_date,
            end_date=end_date,
            granularity=granularity,
            cm_client=cm_client,
            cm_client_query_results=cm_client_query_results,
        )
        return cm_client_query_results


    if group_by == 'ServiceName':
        _cost_export_subfunc(
            group_by=group_by,
            subscriptions_list_detailed=subscriptions_list_detailed,
            start_date=start_date,
            end_date=end_date,
            granularity=granularity,
            cm_client=cm_client,
            cm_client_query_results=cm_client_query_results,
        )
        return cm_client_query_results    

    if group_by == 'ResourceGroupName':
        _cost_export_subfunc(
            group_by=group_by,
            subscriptions_list_detailed=subscriptions_list_detailed,
            start_date=start_date,
            end_date=end_date,
            granularity=granularity,
            cm_client=cm_client,
            cm_client_query_results=cm_client_query_results,
        )
        return cm_client_query_results

list_subs()

Retrieve details of all Azure subscriptions accessible by the authenticated principal.

Uses the Azure SubscriptionClient if available, otherwise falls back to the ARM REST API. Returns detailed information including subscription ID, display name, tenant ID, and tags.

Returns:

Type Description

List[dict[str, Any]]: A list of subscription dictionaries, where each dictionary contains: - 'Subscription_ID' (str): The unique Azure subscription ID - 'Display_Name' (str): The human-readable subscription name - 'Tenant_ID' (str): The Azure tenant ID associated with the subscription - 'Tags' (dict or None): Dictionary of subscription tags if any exist

Raises:

Type Description
AzureError

For Azure API errors.
Exception

For authentication failures.
Example

from eraXplor_azure.utils.cost_export_utils import list_subs subscriptions = list_subs() for sub in subscriptions: ... print(f"{sub['Display_Name']}: {sub['Subscription_ID']}")

Note
  • Requires appropriate Azure RBAC permissions to list subscriptions.
  • The authenticated principal must have Reader role or equivalent on the subscriptions to be listed.
  • Tags are optional and may be None for subscriptions without tags.
Source code in src/core/services/azure/utils/cost_export_utils.py 
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
def list_subs():
    """
    Retrieve details of all Azure subscriptions accessible by the authenticated principal.

    Uses the Azure SubscriptionClient if available, otherwise falls back to the ARM REST API.
    Returns detailed information including subscription ID, display name, tenant ID, and tags.

    Returns:
        List[dict[str, Any]]:
            A list of subscription dictionaries, where each dictionary contains:
            - 'Subscription_ID' (str): The unique Azure subscription ID
            - 'Display_Name' (str): The human-readable subscription name
            - 'Tenant_ID' (str): The Azure tenant ID associated with the subscription
            - 'Tags' (dict or None): Dictionary of subscription tags if any exist

    Raises:
        azure.core.exceptions.AzureError: For Azure API errors.
        Exception: For authentication failures.

    Example:
        >>> from eraXplor_azure.utils.cost_export_utils import list_subs
        >>> subscriptions = list_subs()
        >>> for sub in subscriptions:
        ...     print(f"{sub['Display_Name']}: {sub['Subscription_ID']}")

    Note:
        - Requires appropriate Azure RBAC permissions to list subscriptions.
        - The authenticated principal must have Reader role or equivalent
          on the subscriptions to be listed.
        - Tags are optional and may be None for subscriptions without tags.
    """
    _credential = DefaultAzureCredential()
    subscriptions_list_detailed = []

    # Try using SubscriptionClient if available
    if HAS_SUBSCRIPTION_CLIENT:
        try:
            _subscription_client = SubscriptionClient(_credential)
            _subscriptions = list(_subscription_client.subscriptions.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,
                    }
                )
            return subscriptions_list_detailed
        except Exception:
            pass  # Fall through to REST API fallback

    # Fallback to ARM REST API
    from core.services.utils.get_access_token import get_access_token
    import requests

    token = get_access_token()
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
    }

    url = "https://management.azure.com/subscriptions?api-version=2020-01-01"
    response = requests.get(url, headers=headers)
    response.raise_for_status()

    data = response.json()
    for sub in data.get("value", []):
        subscriptions_list_detailed.append(
            {
                "Subscription_ID": sub.get("subscriptionId"),
                "Display_Name": sub.get("displayName"),
                "Tenant_ID": sub.get("tenantId"),
                "Tags": sub.get("tags"),
            }
        )

    return subscriptions_list_detailed

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.

This module provides functionality to write Azure cost and usage data to CSV files with standardized formatting. It is typically used in conjunction with the cost_export() function to persist cost data for further analysis or reporting.

The CSV output includes the following columns
  • TIME_PERIOD: Date or date range for the cost record
  • GROUP_BY: The grouping dimension used (e.g. 'SUBSCRIPTION_ID', currency)
  • SUBSCRIPTION_ID: The Azure subscription ID
  • DISPLAY_NAME: The subscription display name
  • PreTaxCost: The cost amount with currency
  • TAGS: Subscription tags (if available)
Example

from eraXplor_azure.utils.cost_export_utils import cost_export, list_subs from eraXplor_azure.utils.csv_export_utils import csv_export subs = list_subs() costs = cost_export( ... group_by='subscription', ... subscriptions_list_detailed=subs, ... start_date='2025,01,01', ... end_date='2025,01,31' ... ) csv_export(cm_client_query_results=costs, filename='cost_report.csv')

csv_export(cm_client_query_results, filename)

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

Takes the output from cost_export() and writes it to a CSV file with consistent column headers and proper formatting. The CSV will contain cost records with their associated metadata including time period, grouping information, subscription details, and tags.

Parameters:

Name Type Description Default
cm_client_query_results List[Dict[str, Any]]

List of cost data dictionaries as returned by cost_export(). Each dictionary should contain the following keys: - TIME_PERIOD (str): Date or date range for the cost record - GROUP_BY (str): The grouping dimension used - SUBSCRIPTION_ID (str): The Azure subscription ID - DISPLAY_NAME (str): The subscription display name - PreTaxCost (str): Cost amount formatted with currency - TAGS (dict or str): Subscription tags or "None"

 required
filename str

Output filename for the CSV file. Defaults to 'az_cost_report.csv' if not specified. The file will be created in the current working directory unless a path is included in the filename.

 required

Returns:

Name Type Description
None None

This function writes directly to file and prints a confirmation message to stdout, but does not return any value.

Raises:

Type Description
IOError

If the file cannot be created or written to.
KeyError

If required keys are missing from the input dictionaries.
Example

from eraXplor_azure.utils.csv_export_utils import csv_export costs = [ ... { ... 'TIME_PERIOD': {'Start': '2025-01-01', 'End': '2025-01-31'}, ... 'GROUP_BY': 'SUBSCRIPTION_ID', ... 'SUBSCRIPTION_ID': 'sub-12345', ... 'DISPLAY_NAME': 'My Subscription', ... 'PreTaxCost': '123.45 USD', ... 'TAGS': {'env': 'production'} ... } ... ] csv_export(cm_client_query_results=costs, filename='report.csv') Data exported to report.csv

Notes
  • The function uses UTF-8 encoding for proper handling of special characters.
  • A confirmation message is printed to console upon successful export.
  • Existing files with the same name will be overwritten.
Source code in src/core/services/azure/utils/csv_export_utils.py 
 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
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() and writes it to a CSV file with
    consistent column headers and proper formatting. The CSV will contain
    cost records with their associated metadata including time period,
    grouping information, subscription details, and tags.

    Args:
        cm_client_query_results (List[Dict[str, Any]]):
            List of cost data dictionaries as returned by cost_export().
            Each dictionary should contain the following keys:
            - TIME_PERIOD (str): Date or date range for the cost record
            - GROUP_BY (str): The grouping dimension used
            - SUBSCRIPTION_ID (str): The Azure subscription ID
            - DISPLAY_NAME (str): The subscription display name
            - PreTaxCost (str): Cost amount formatted with currency
            - TAGS (dict or str): Subscription tags or "None"

        filename (str):
            Output filename for the CSV file. Defaults to 'az_cost_report.csv'
            if not specified. The file will be created in the current working
            directory unless a path is included in the filename.

    Returns:
        None: This function writes directly to file and prints a confirmation
              message to stdout, but does not return any value.

    Raises:
        IOError: If the file cannot be created or written to.
        KeyError: If required keys are missing from the input dictionaries.

    Example:
        >>> from eraXplor_azure.utils.csv_export_utils import csv_export
        >>> costs = [
        ...     {
        ...         'TIME_PERIOD': {'Start': '2025-01-01', 'End': '2025-01-31'},
        ...         'GROUP_BY': 'SUBSCRIPTION_ID',
        ...         'SUBSCRIPTION_ID': 'sub-12345',
        ...         'DISPLAY_NAME': 'My Subscription',
        ...         'PreTaxCost': '123.45 USD',
        ...         'TAGS': {'env': 'production'}
        ...     }
        ... ]
        >>> csv_export(cm_client_query_results=costs, filename='report.csv')
        Data exported to report.csv

    Notes:
        - The function uses UTF-8 encoding for proper handling of special characters.
        - A confirmation message is printed to console upon successful export.
        - Existing files with the same name will be overwritten.
    """
    # 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",
                "GROUP_BY",
                "SUBSCRIPTION_ID",
                "DISPLAY_NAME",
                "PreTaxCost",
                "TAGS",
            ]
        )
        for row in cm_client_query_results:
            time_period = row["TIME_PERIOD"]
            group_by = row["GROUP_BY"]
            subscription_id = row["SUBSCRIPTION_ID"]
            display_name = row["DISPLAY_NAME"]
            PreTaxCost = row.get("PreTaxCost")
            tags = row.get("TAGS", {})
            writer.writerow(
                [time_period, group_by, subscription_id, display_name, PreTaxCost, tags]
                )
    print(f"\n Data exported to {filename}")

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

Resource Provisioning Utilities

Module for installing dependencies required for Azure FOCUS data export.

This module provides functions to create necessary Azure resources for FOCUS data export, including: - Resource Group - Storage Account - Blob Container - Folder (Virtual Directory) within the Blob Container

ARGS: - subscription_id: Azure Subscription ID - resource_group_name: Name of the Resource Group to create - location: Azure region for the resources - storage_account_name: Name of the Storage Account to create - container_name: Name of the Blob Container to create - folder_name: Name of the Folder (Virtual Directory) to create within the Blob Container

Dependencies: - azure-identity - azure-mgmt-storage - azure-mgmt-resource - azure-storage-blob

create_resource_group(subscription_id, resource_group_name='focus-data-export-rg', location='eastus')

Creates a resource group if it does not exist.

Source code in src/core/services/azure/utils/focus_depends.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def create_resource_group(
        subscription_id: str,
        resource_group_name: str = "focus-data-export-rg",
        location: str = "eastus",
        ) -> None:
    """
    Creates a resource group if it does not exist.
    """

    credential = DefaultAzureCredential()
    resource_client = ResourceManagementClient(credential, subscription_id)

    resource_client.resource_groups.create_or_update(
        resource_group_name,
        {"location": location},
    )
    print(f"Resource group '{resource_group_name}' created or already exists.")

create_storage_account_container_folder(subscription_id, resource_group_name='focus-data-export-rg', location='eastus', storage_account_name='focusdataexportstorage', container_name='focus-data', folder_name='focus-exports')

Creates storage account, container, and folder (virtual directory).

Source code in src/core/services/azure/utils/focus_depends.py 
 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
def create_storage_account_container_folder(
    subscription_id: str,
    resource_group_name: str = "focus-data-export-rg",
    location: str = "eastus",
    storage_account_name: str = "focusdataexportstorage",
    container_name: str = "focus-data",
    folder_name: str = "focus-exports",
    ) -> None:
    """
    Creates storage account, container, and folder (virtual directory).
    """

    credential = DefaultAzureCredential()
    account_name = storage_account_name.lower()
    storage_client = StorageManagementClient(credential, subscription_id)

    # Create Storage Account
    print(f"Creating storage account '{account_name}'...")
    poller = storage_client.storage_accounts.begin_create(
        resource_group_name,
        account_name,
        StorageAccountCreateParameters(
            sku=Sku(name="Standard_LRS"),
            kind="StorageV2",
            location=location,
        ),
    )
    poller.result()
    print("Storage account created successfully.")

    # Get Storage Account Key
    keys = storage_client.storage_accounts.list_keys(
        resource_group_name,
        account_name,
    )
    account_key = keys.keys[0].value
    account_url = f"https://{account_name}.blob.core.windows.net"
    blob_service_client = BlobServiceClient(
        account_url=account_url,
        credential=account_key,
    )

    # Create Container
    try:
        blob_service_client.create_container(container_name)
        print(f"Container '{container_name}' created.")
    except ResourceExistsError:
        print(f"Container '{container_name}' already exists.")

    # Create Folder (Virtual Directory)
    folder_blob_name = f"{folder_name}/"
    blob_client = blob_service_client.get_blob_client(
        container=container_name,
        blob=folder_blob_name,
    )
    blob_client.upload_blob(b"", overwrite=True)
    print(f"Folder '{folder_name}/' created successfully.")

Creates the Azure resource group, storage account, container, and folder required by the export workflow.

Export Configuration Utilities

Utility functions for configuring Azure FOCUS exports.

This module contains helper functions for: - Azure authentication - Building Azure Cost Management export payloads - Creating scheduled FOCUS exports in Parquet format

build_export_payload(storage_resource_id, container_name, folder_name)

Build the request payload for Azure FOCUS export creation.

The payload configures: - FOCUS cost dataset - Daily export schedule - Parquet output format - Destination storage container and folder

Parameters:

Name Type Description Default
storage_resource_id str

Azure Storage Account resource ID.

 required
container_name str

Blob container name.

 required
folder_name str

Root folder path for exported files.

 required

Returns:

Type Description
dict[str, Any]

dict[str, Any]: JSON payload for Azure export API request.
Source code in src/core/services/azure/utils/focus_export_utils.py 
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
def build_export_payload(
    storage_resource_id: str,
    container_name: str,
    folder_name: str,
) -> dict[str, Any]:
    """
    Build the request payload for Azure FOCUS export creation.

    The payload configures:
    - FOCUS cost dataset
    - Daily export schedule
    - Parquet output format
    - Destination storage container and folder

    Args:
        storage_resource_id (str): Azure Storage Account resource ID.
        container_name (str): Blob container name.
        folder_name (str): Root folder path for exported files.

    Returns:
        dict[str, Any]: JSON payload for Azure export API request.
    """
    start_date = (
        datetime.now(UTC) + timedelta(days=1)
    ).strftime("%Y-%m-%dT00:00:00Z")

    end_date = "2030-01-01T00:00:00Z"

    return {
        "location": "global",
        "identity": {
            "type": "SystemAssigned",
        },
        "properties": {
            "definition": {
                "type": "FocusCost",
                "timeframe": "MonthToDate",
                "dataset": {
                    "granularity": "Daily",
                    "configuration": {
                        "dataVersion": "1.0",
                    },
                },
            },
            "deliveryInfo": {
                "destination": {
                    "resourceId": storage_resource_id,
                    "container": container_name,
                    "rootFolderPath": folder_name,
                }
            },
            "format": "Parquet",
            "partitionData": True,
            "schedule": {
                "status": "Active",
                "recurrence": "Daily",
                "recurrencePeriod": {
                    "from": start_date,
                    "to": end_date,
                },
            },
        },
    }

create_focus_export(billing_account_id, billing_profile_id, subscription_id, resource_group_name, storage_account_name, container_name, folder_name, export_name='focus-cost-export')

Create an Azure Cost Management FOCUS export.

This function authenticates to Azure, builds the required export configuration, and creates a scheduled FOCUS export that writes Parquet files into the specified Azure Storage container.

Parameters:

Name Type Description Default
billing_account_id str

Azure billing account identifier.

 required
billing_profile_id str

Azure billing profile identifier.

 required
subscription_id str

Azure subscription ID.

 required
resource_group_name str

Resource group containing storage account.

 required
storage_account_name str

Destination storage account name.

 required
container_name str

Blob container name.

 required
folder_name str

Folder path inside container.

 required
export_name str

Export job name. Defaults to "focus-cost-export".

 'focus-cost-export'

Returns:

Type Description
dict[str, Any]

dict[str, Any]: Result containing: - status_code (int): HTTP response code. - data (dict[str, Any]): Parsed API response.

Raises:

Type Description
RuntimeError

If Azure API request fails.
Source code in src/core/services/azure/utils/focus_export_utils.py 
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
def create_focus_export(
    billing_account_id: str,
    billing_profile_id: str,
    subscription_id: str,
    resource_group_name: str,
    storage_account_name: str,
    container_name: str,
    folder_name: str,
    export_name: str = "focus-cost-export",
) -> dict[str, Any]:
    """
    Create an Azure Cost Management FOCUS export.

    This function authenticates to Azure, builds the required export
    configuration, and creates a scheduled FOCUS export that writes
    Parquet files into the specified Azure Storage container.

    Args:
        billing_account_id (str): Azure billing account identifier.
        billing_profile_id (str): Azure billing profile identifier.
        subscription_id (str): Azure subscription ID.
        resource_group_name (str): Resource group containing storage account.
        storage_account_name (str): Destination storage account name.
        container_name (str): Blob container name.
        folder_name (str): Folder path inside container.
        export_name (str, optional): Export job name.
            Defaults to "focus-cost-export".

    Returns:
        dict[str, Any]: Result containing:
            - status_code (int): HTTP response code.
            - data (dict[str, Any]): Parsed API response.

    Raises:
        RuntimeError: If Azure API request fails.
    """
    print("Authenticating to Azure...")
    token = get_access_token()

    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
    }

    scope = _build_scope(
        billing_account_id,
        billing_profile_id,
    )

    storage_resource_id = _build_storage_resource_id(
        subscription_id,
        resource_group_name,
        storage_account_name,
    )

    payload = build_export_payload(
        storage_resource_id,
        container_name,
        folder_name,
    )

    url = (
        f"{settings.AZURE_MANAGEMENT_BASE_URL}{scope}"
        f"/providers/Microsoft.CostManagement/"
        f"exports/{export_name}"
        f"?api-version={settings.AZURE_COST_API_VERSION}"
    )

    print("Creating FOCUS export...")

    response = requests.put(
        url=url,
        headers=headers,
        json=payload,
        timeout=60,
    )

    if not response.ok:
        raise RuntimeError(
            f"FOCUS export failed: "
            f"{response.status_code} - {response.text}"
        )

    print("FOCUS export created successfully.")

    try:
        data = response.json()
    except ValueError:
        data = {"message": response.text}

    return {
        "status_code": response.status_code,
        "data": data,
    }

get_billing_accounts()

Return billing accounts available to the authenticated Azure identity.

Source code in src/core/services/azure/utils/focus_export_utils.py 
84
85
86
87
88
89
90
91
92
93
def get_billing_accounts() -> list[dict[str, Any]]:
    """Return billing accounts available to the authenticated Azure identity."""
    token = get_access_token()
    url = (
        f"{settings.AZURE_MANAGEMENT_BASE_URL}"
        "/providers/Microsoft.Billing/billingAccounts"
        "?api-version=2020-05-01"
    )
    data = _fetch_arm_json(url, _get_arm_headers(token))
    return data.get("value", [])

get_billing_profiles(billing_account_id)

Return billing profiles for a given Azure billing account.

Source code in src/core/services/azure/utils/focus_export_utils.py 
 96
 97
 98
 99
100
101
102
103
104
105
def get_billing_profiles(billing_account_id: str) -> list[dict[str, Any]]:
    """Return billing profiles for a given Azure billing account."""
    token = get_access_token()
    url = (
        f"{settings.AZURE_MANAGEMENT_BASE_URL}"
        f"/providers/Microsoft.Billing/billingAccounts/{billing_account_id}"
        "/billingProfiles?api-version=2020-05-01"
    )
    data = _fetch_arm_json(url, _get_arm_headers(token))
    return data.get("value", [])

get_default_billing_account_and_profile_ids()

Fetch the first available billing account and profile IDs for the authenticated credential.

Source code in src/core/services/azure/utils/focus_export_utils.py 
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
def get_default_billing_account_and_profile_ids() -> dict[str, str]:
    """Fetch the first available billing account and profile IDs for the authenticated credential."""
    accounts = get_billing_accounts()
    if not accounts:
        raise RuntimeError(
            "No Azure billing accounts were found for the authenticated identity."
        )

    billing_account = accounts[0]
    billing_account_id = billing_account.get("name")
    if not billing_account_id:
        raise RuntimeError(
            "Billing account response missing required 'name' field."
        )

    profiles = get_billing_profiles(billing_account_id)
    if not profiles:
        raise RuntimeError(
            f"No billing profiles found for billing account '{billing_account_id}'."
        )

    billing_profile = profiles[0]
    billing_profile_id = billing_profile.get("name")
    if not billing_profile_id:
        raise RuntimeError(
            "Billing profile response missing required 'name' field."
        )

    return {
        "billing_account_id": billing_account_id,
        "billing_profile_id": billing_profile_id,
    }

Contains the helpers that discover billing identifiers and create Azure FOCUS export definitions.

Fetch Utilities

Module for fetching Parquet files from Azure Blob Storage for FOCUS data export. This module provides a function to download Parquet files from a specified folder (virtual directory) within an Azure Blob Storage container.

ARGS: - storage_account_name: Name of the Storage Account - container_name: Name of the Blob Container - folder_name: Name of the Folder (Virtual Directory) within the Blob Container

Dependencies: - azure-identity - azure-storage-blob

download_parquet_files(storage_account_name, container_name, folder_name)

Downloads Parquet files from Azure Blob Storage

Parameters:

Name Type Description Default
- storage_account_name

Name of the Storage Account

 required
- container_name

Name of the Blob Container

 required
- folder_name

Name of the Folder (Virtual Directory) within the Blob Container

 required

Returns:

Type Description
list
  • List of paths to the downloaded Parquet files
Source code in src/core/services/azure/utils/focus_fetch.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
def download_parquet_files(
    storage_account_name: str,
    container_name: str,
    folder_name: str,
    ) -> list:
    """
    Downloads Parquet files from Azure Blob Storage

    Args:
        - storage_account_name: Name of the Storage Account
        - container_name: Name of the Blob Container
        - folder_name: Name of the Folder (Virtual Directory) within the Blob Container

    Returns:
        - List of paths to the downloaded Parquet files 
    """

    local_download_path = "./downloaded_parquet_files"
    os.makedirs(local_download_path, exist_ok=True)

    credential = DefaultAzureCredential()
    container_client = ContainerClient(
        account_url=f"https://{storage_account_name}.blob.core.windows.net",
        container_name=container_name,
        credential=credential,
    )

    print(f"Listing blobs in container '{container_name}' under folder '{folder_name}'...")
    blob_list = container_client.list_blobs(name_starts_with=f"{folder_name}/")

    downloaded_files = []
    for blob in blob_list:
        blob_name = blob.name
        download_file_path = os.path.join(local_download_path, os.path.basename(blob_name))
        with open(download_file_path, "wb") as f:
            f.write(container_client.download_blob(blob_name).readall())
        downloaded_files.append(download_file_path)
        print(f"Downloaded: {download_file_path}")

    return downloaded_files

Provides functionality for downloading generated Parquet export files from the configured Azure Storage container.