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 manages dependencies of Azure FOCUS export workflow.

Source code in src/core/services/azure/__main__.py 
 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
def main() -> None:
    """Orchestrates and manages dependencies of Azure FOCUS export workflow."""
    try:
        # 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")

        # Parse CLI args
        arg_parser = parser().parse_args()

        # Resolve parser-managed values
        rg_name_input = parser_resource_group_name_handler(arg_parser)
        location_input = parser_location_handler(arg_parser)
        storage_account_name_input = parser_storage_account_name_handler(arg_parser)
        container_name_input = parser_container_name_handler(arg_parser)
        folder_name_input = parser_folder_name_handler(arg_parser)
        subscription_id_input = parser_subscription_id_handler(arg_parser)
        granularity_input = parser_granularity_handler(arg_parser)
        command_mode = parser_command_handler(arg_parser)

        if command_mode == "configure":
            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,
            )
            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,
                granularity=granularity_input,
            )
            print(
                "FOCUS export configured successfully. "
                "Run again with '--command download' after data is available.",
                end="\n\n",
            )
            return

        if command_mode == "download":
            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,
            )
            return

        raise ValueError(
            f"Unsupported command mode '{command_mode}'. "
            "Use '--command configure' or '--command download'."
        )
    except ValueError as exc:
        print(f"Error: {exc}")
        sys.exit(2)
    except Exception as exc:
        print(f"Unexpected error: {exc}")
        sys.exit(1)

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

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, granularity='Monthly')

Build the request payload for Azure FOCUS export creation.

The payload configures: - FOCUS cost dataset - Scheduled export (recurrence matches granularity) - 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
granularity str

Dataset granularity. One of "Hourly", "Daily", or "Monthly". Defaults to "Monthly".

 'Monthly'

Returns:

Type Description
dict[str, Any]

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

    The payload configures:
    - FOCUS cost dataset
    - Scheduled export (recurrence matches granularity)
    - 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.
        granularity (str): Dataset granularity. One of "Hourly", "Daily", or "Monthly".
                           Defaults to "Monthly".

    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": granularity,
                    "configuration": {
                        "dataVersion": "1.0",
                    },
                },
            },
            "deliveryInfo": {
                "destination": {
                    "resourceId": storage_resource_id,
                    "container": container_name,
                    "rootFolderPath": folder_name,
                }
            },
            "format": "Parquet",
            "partitionData": True,
            "schedule": {
                "status": "Active",
                "recurrence": granularity,  # "Daily" or "Monthly"
                "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, granularity='Monthly', 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/azure_focus_export_utils.py 
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
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,
    granularity: str = "Monthly",
    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,
        granularity=granularity,
    )

    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/azure_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/azure_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/azure_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 helpers for discovering Azure billing identifiers and creating Azure FOCUS export definitions.

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/azure_focus_depends.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
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/azure_focus_depends.py 
 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
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, granularity='Monthly')

Build the request payload for Azure FOCUS export creation.

The payload configures: - FOCUS cost dataset - Scheduled export (recurrence matches granularity) - 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
granularity str

Dataset granularity. One of "Hourly", "Daily", or "Monthly". Defaults to "Monthly".

 'Monthly'

Returns:

Type Description
dict[str, Any]

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

    The payload configures:
    - FOCUS cost dataset
    - Scheduled export (recurrence matches granularity)
    - 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.
        granularity (str): Dataset granularity. One of "Hourly", "Daily", or "Monthly".
                           Defaults to "Monthly".

    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": granularity,
                    "configuration": {
                        "dataVersion": "1.0",
                    },
                },
            },
            "deliveryInfo": {
                "destination": {
                    "resourceId": storage_resource_id,
                    "container": container_name,
                    "rootFolderPath": folder_name,
                }
            },
            "format": "Parquet",
            "partitionData": True,
            "schedule": {
                "status": "Active",
                "recurrence": granularity,  # "Daily" or "Monthly"
                "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, granularity='Monthly', 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/azure_focus_export_utils.py 
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
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,
    granularity: str = "Monthly",
    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,
        granularity=granularity,
    )

    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/azure_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/azure_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/azure_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/azure_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.