确定资源组的成本
组
def get_resource_group_cost(credential, subscription_id: str, resource_group_name: str, days: int = 7) -> Dict:
try:
cost_client = CostManagementClient(credential)
end_time = datetime.now(timezone.utc).replace(hour=0, minute=0, second=0, microsecond=0)
start_time = end_time - timedelta(days=days)
scope = f"/subscriptions/{subscription_id}/resourceGroups/{resource_group_name}"
parameters = {
"type": "ActualCost",
"timeframe": "Custom",
"timePeriod": {
"from": start_time.isoformat(),
"to": end_time.isoformat()
},
"dataset": {
"granularity": "None",
"aggregation": {
"totalCost": {
"name": "Cost",
"function": "Sum"
}
},
"grouping": [
{
"type": "Dimension",
"name": "ResourceId"
}
]
}
}
cost_data = cost_client.query.usage(scope=scope, parameters=parameters)
# Process the results into a dictionary of resource costs
resource_costs = {}
if cost_data.rows:
for row in cost_data.rows:
resource_id = row[1] # ResourceId is in the second column
cost = row[0] # Cost is in the first column
currency = row[2] # Currency is in the third column
resource_costs[resource_id] = {
'total_cost': cost,
'avg_daily_cost': cost / days,
'currency': currency
}
return resource_costs
except Exception as e:
print(f"Error getting cost data for resource group {resource_group_name}: {str(e)}")
return None