Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
namespace = resource_id.split("/")[6]
service = resource_id.split("/")[7]
except IndexError:
raise MsticpyAzureException(
"Provided Resource ID isn't in the correct format.",
"It should look like:",
"/subscriptions/SUB_ID/resourceGroups/RESOURCE_GROUP/"
+ "providers/NAMESPACE/SERVICE_NAME/RESOURCE_NAME ",
)
elif resource_provider is not None:
try:
namespace = resource_provider.split("/")[0]
service = resource_provider.split("/")[1]
except IndexError:
raise MsticpyAzureException(
"""Provided Resource Provider isn't in the correct format. It should look like:
NAMESPACE/SERVICE_NAME"""
)
else:
raise ValueError(
"Please provide an resource ID or resource provider namespace"
)
# Get list of API versions for the service
provider = self.resource_client.providers.get(namespace) # type: ignore
resource_types = next(
(t for t in provider.resource_types if t.resource_type == service), None
)
# Get first API version that isn't in preview
if resource_types:
resource_id: str, optional
The ID of the resources to get an API version for
sub_id: str, optional
The ID of the subscription to get details from
resource_provider: str, optional
The resource provider namespace and service to get an API version for
Returns
-------
api_ver: str
The latest avaliable non-preview API version
"""
# Check if connection and client required are already present
if self.connected is False:
raise MsticpyAzureException("Please connect before continuing")
self._check_client("resource_client", sub_id) # type: ignore
# Normalise elements depending on user input type
if resource_id is not None:
try:
namespace = resource_id.split("/")[6]
service = resource_id.split("/")[7]
except IndexError:
raise MsticpyAzureException(
"Provided Resource ID isn't in the correct format.",
"It should look like:",
"/subscriptions/SUB_ID/resourceGroups/RESOURCE_GROUP/"
+ "providers/NAMESPACE/SERVICE_NAME/RESOURCE_NAME ",
)
The latest avaliable non-preview API version
"""
# Check if connection and client required are already present
if self.connected is False:
raise MsticpyAzureException("Please connect before continuing")
self._check_client("resource_client", sub_id) # type: ignore
# Normalise elements depending on user input type
if resource_id is not None:
try:
namespace = resource_id.split("/")[6]
service = resource_id.split("/")[7]
except IndexError:
raise MsticpyAzureException(
"Provided Resource ID isn't in the correct format.",
"It should look like:",
"/subscriptions/SUB_ID/resourceGroups/RESOURCE_GROUP/"
+ "providers/NAMESPACE/SERVICE_NAME/RESOURCE_NAME ",
)
elif resource_provider is not None:
try:
namespace = resource_provider.split("/")[0]
service = resource_provider.split("/")[1]
except IndexError:
raise MsticpyAzureException(
"""Provided Resource Provider isn't in the correct format. It should look like:
NAMESPACE/SERVICE_NAME"""
)
else:
Accepted inputs = 'hour' or 'minute'
start_time: int (Optional)
The number of days prior to today to collect metrics for, default is 30
Returns
-------
results: dict
A Dictionary of DataFrames containing the metrics details
"""
if sample_time.casefold().startswith("h"):
interval = "PT1H"
elif sample_time.casefold().startswith("m"):
interval = "PT1M"
else:
raise MsticpyAzureException(
"Select how often you want to sample data - 'hour', or 'minute'"
)
# Check if connection and client required are already present
if self.connected is False:
raise MsticpyAzureException("Please connect before continuing")
self._check_client("monitoring_client", sub_id)
# Get metrics in one hour chunks for the last 30 days
start = datetime.datetime.now().date()
end = start - datetime.timedelta(days=start_time)
mon_details = self.monitoring_client.metrics.list( # type: ignore
resource_id,
timespan=f"{end}/{start}",
def connect(self, client_id: str = None, tenant_id: str = None, secret: str = None):
"""Authenticate with the SDK."""
# Use details of msticpyyaml if not provided
if client_id is None and tenant_id is None and secret is None:
data_provs = get_provider_settings(config_section="DataProviders")
az_cli_config = data_provs.get("AzureCLI")
# az_cli_config = config.settings.get("AzureCLI")
if not az_cli_config:
raise MsticpyAzureException(
"No AzureCLI configuration found in configuration settings."
)
config_items = az_cli_config.args
client_id = config_items["clientId"]
tenant_id = config_items["tenantId"]
secret = config_items["clientSecret"]
# Create credentials and connect to the subscription client to validate
self.credentials = ServicePrincipalCredentials(
client_id=client_id, secret=secret, tenant=tenant_id
)
if not self.credentials:
raise CloudError("Could not obtain credentials.")
self.sub_client = SubscriptionClient(self.credentials)
if not self.sub_client:
raise CloudError("Could not create a Subscription client.")
provider = self.resource_client.providers.get(namespace) # type: ignore
resource_types = next(
(t for t in provider.resource_types if t.resource_type == service), None
)
# Get first API version that isn't in preview
if resource_types:
api_version = [
v for v in resource_types.api_versions if "preview" not in v.lower()
]
if api_version is None or not api_version:
api_ver = resource_types.api_versions[0]
else:
api_ver = api_version[0]
else:
raise MsticpyAzureException("Resource provider not found")
return str(api_ver)
def get_subscription_info(self, sub_id: str) -> dict:
"""
Get information on a specific subscription.
Parameters
----------
sub_id : str
The ID of the subscription to return details on.
"""
if self.connected is False:
raise MsticpyAzureException("Please connect before continuing")
sub = self.sub_client.subscriptions.get(sub_id) # type: ignore
sub_details = {
"Subscription ID": sub.subscription_id,
"Display Name": sub.display_name,
"State": str(sub.state),
"Subscription Location": sub.subscription_policies.location_placement_id,
"Subscription Quota": sub.subscription_policies.quota_id,
"Spending Limit": sub.subscription_policies.spending_limit,
}
return sub_details