# coding=utf-8
# --------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for license information.
# Code generated by Microsoft (R) AutoRest Code Generator.
# Changes may cause incorrect behavior and will be lost if the code is regenerated.
# --------------------------------------------------------------------------
from typing import Any, Optional, TYPE_CHECKING, Union
from azure.core.pipeline.transport._base import _format_url_section
from azure.purview.scanning.core.rest import HttpRequest
from msrest import Serializer
if TYPE_CHECKING:
# pylint: disable=unused-import,ungrouped-imports
from typing import Any
_SERIALIZER = Serializer()
[docs]def build_create_or_update_request(
data_source_name: str,
scan_name: str,
*,
json: Any = None,
content: Any = None,
**kwargs: Any
) -> HttpRequest:
"""Creates an instance of a scan.
See https://aka.ms/azsdk/python/protocol/quickstart for how to incorporate this request builder into your code flow.
:param data_source_name:
:type data_source_name: str
:param scan_name:
:type scan_name: str
:keyword json:
:paramtype json: Any
:keyword content:
:paramtype content: Any
:return: Returns an :class:`~azure.purview.scanning.core.rest.HttpRequest` that you will pass to the client's `send_request` method.
See https://aka.ms/azsdk/python/protocol/quickstart for how to incorporate this response into your code flow.
:rtype: ~azure.purview.scanning.core.rest.HttpRequest
Example:
.. code-block:: python
# kind template as part of your input body
kind = 'AdlsGen1CredentialScan' or 'AdlsGen1MsiScan' or 'AdlsGen2CredentialScan' or 'AdlsGen2MsiScan' or 'AmazonAccountCredentialScan' or 'AmazonPostgreSqlCredentialScan' or 'AmazonS3CredentialScan' or 'AmazonS3RoleARNScan' or 'AmazonSqlCredentialScan' or 'AzureCosmosDbCredentialScan' or 'AzureDataExplorerCredentialScan' or 'AzureDataExplorerMsiScan' or 'AzureFileServiceCredentialScan' or 'AzureMySqlCredentialScan' or 'AzurePostgreSqlCredentialScan' or 'AzureResourceGroupCredentialScan' or 'AzureResourceGroupMsiScan' or 'AzureSqlDataWarehouseCredentialScan' or 'AzureSqlDataWarehouseMsiScan' or 'AzureSqlDatabaseCredentialScan' or 'AzureSqlDatabaseManagedInstanceCredentialScan' or 'AzureSqlDatabaseManagedInstanceMsiScan' or 'AzureSqlDatabaseMsiScan' or 'AzureStorageCredentialScan' or 'AzureStorageMsiScan' or 'AzureSubscriptionCredentialScan' or 'AzureSubscriptionMsiScan' or 'AzureSynapseCredentialScan' or 'AzureSynapseMsiScan' or 'AzureSynapseWorkspaceCredentialScan' or 'AzureSynapseWorkspaceMsiScan' or 'OracleCredentialScan' or 'OracleUserPassScan' or 'PowerBIDelegatedScan' or 'PowerBIMsiScan' or 'SapEccCredentialScan' or 'SapEccUserPassScan' or 'SapS4HanaSapS4HanaCredentialScan' or 'SapS4HanaSapS4HanaUserPassScan' or 'SqlServerDatabaseCredentialScan' or 'TeradataCredentialScan' or 'TeradataUserPassScanAutoGenerated' or 'TeradataUserPassScan'
# JSON input template you can fill out and use as your `json` input.
json = {
"kind": "Scan",
"scanResults": [
{
"assetsClassified": "long (optional)",
"assetsDiscovered": "long (optional)",
"dataSourceType": "str (optional)",
"diagnostics": {},
"endTime": "datetime (optional)",
"error": {},
"errorMessage": "str (optional)",
"id": "str (optional)",
"parentId": "str (optional)",
"pipelineStartTime": "datetime (optional)",
"queuedTime": "datetime (optional)",
"resourceId": "str (optional)",
"runType": "str (optional)",
"scanLevelType": "str (optional)",
"scanRulesetType": "str (optional)",
"scanRulesetVersion": "int (optional)",
"startTime": "datetime (optional)",
"status": "str (optional)"
}
]
}
# response body for status code(s): 200, 201
response_body == {
"kind": "Scan",
"scanResults": [
{
"assetsClassified": "long (optional)",
"assetsDiscovered": "long (optional)",
"dataSourceType": "str (optional)",
"diagnostics": {},
"endTime": "datetime (optional)",
"error": {},
"errorMessage": "str (optional)",
"id": "str (optional)",
"parentId": "str (optional)",
"pipelineStartTime": "datetime (optional)",
"queuedTime": "datetime (optional)",
"resourceId": "str (optional)",
"runType": "str (optional)",
"scanLevelType": "str (optional)",
"scanRulesetType": "str (optional)",
"scanRulesetVersion": "int (optional)",
"startTime": "datetime (optional)",
"status": "str (optional)"
}
]
}
"""
content_type = kwargs.pop("content_type", None)
api_version = "2018-12-01-preview"
accept = "application/json"
# Construct URL
url = kwargs.pop("template_url", '/datasources/{dataSourceName}/scans/{scanName}')
path_format_arguments = {
'dataSourceName': _SERIALIZER.url("data_source_name", data_source_name, 'str'),
'scanName': _SERIALIZER.url("scan_name", scan_name, 'str'),
}
url = _format_url_section(url, **path_format_arguments)
# Construct parameters
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
if content_type is not None:
header_parameters['Content-Type'] = _SERIALIZER.header("content_type", content_type, 'str')
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="PUT",
url=url,
params=query_parameters,
headers=header_parameters,
json=json,
content=content,
**kwargs
)
[docs]def build_get_request(
data_source_name: str,
scan_name: str,
**kwargs: Any
) -> HttpRequest:
"""Gets a scan information.
See https://aka.ms/azsdk/python/protocol/quickstart for how to incorporate this request builder into your code flow.
:param data_source_name:
:type data_source_name: str
:param scan_name:
:type scan_name: str
:return: Returns an :class:`~azure.purview.scanning.core.rest.HttpRequest` that you will pass to the client's `send_request` method.
See https://aka.ms/azsdk/python/protocol/quickstart for how to incorporate this response into your code flow.
:rtype: ~azure.purview.scanning.core.rest.HttpRequest
Example:
.. code-block:: python
# response body for status code(s): 200
response_body == {
"kind": "Scan",
"scanResults": [
{
"assetsClassified": "long (optional)",
"assetsDiscovered": "long (optional)",
"dataSourceType": "str (optional)",
"diagnostics": {},
"endTime": "datetime (optional)",
"error": {},
"errorMessage": "str (optional)",
"id": "str (optional)",
"parentId": "str (optional)",
"pipelineStartTime": "datetime (optional)",
"queuedTime": "datetime (optional)",
"resourceId": "str (optional)",
"runType": "str (optional)",
"scanLevelType": "str (optional)",
"scanRulesetType": "str (optional)",
"scanRulesetVersion": "int (optional)",
"startTime": "datetime (optional)",
"status": "str (optional)"
}
]
}
"""
api_version = "2018-12-01-preview"
accept = "application/json"
# Construct URL
url = kwargs.pop("template_url", '/datasources/{dataSourceName}/scans/{scanName}')
path_format_arguments = {
'dataSourceName': _SERIALIZER.url("data_source_name", data_source_name, 'str'),
'scanName': _SERIALIZER.url("scan_name", scan_name, 'str'),
}
url = _format_url_section(url, **path_format_arguments)
# Construct parameters
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="GET",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
[docs]def build_delete_request(
data_source_name: str,
scan_name: str,
**kwargs: Any
) -> HttpRequest:
"""Deletes the scan associated with the data source.
See https://aka.ms/azsdk/python/protocol/quickstart for how to incorporate this request builder into your code flow.
:param data_source_name:
:type data_source_name: str
:param scan_name:
:type scan_name: str
:return: Returns an :class:`~azure.purview.scanning.core.rest.HttpRequest` that you will pass to the client's `send_request` method.
See https://aka.ms/azsdk/python/protocol/quickstart for how to incorporate this response into your code flow.
:rtype: ~azure.purview.scanning.core.rest.HttpRequest
Example:
.. code-block:: python
# response body for status code(s): 200
response_body == {
"kind": "Scan",
"scanResults": [
{
"assetsClassified": "long (optional)",
"assetsDiscovered": "long (optional)",
"dataSourceType": "str (optional)",
"diagnostics": {},
"endTime": "datetime (optional)",
"error": {},
"errorMessage": "str (optional)",
"id": "str (optional)",
"parentId": "str (optional)",
"pipelineStartTime": "datetime (optional)",
"queuedTime": "datetime (optional)",
"resourceId": "str (optional)",
"runType": "str (optional)",
"scanLevelType": "str (optional)",
"scanRulesetType": "str (optional)",
"scanRulesetVersion": "int (optional)",
"startTime": "datetime (optional)",
"status": "str (optional)"
}
]
}
"""
api_version = "2018-12-01-preview"
accept = "application/json"
# Construct URL
url = kwargs.pop("template_url", '/datasources/{dataSourceName}/scans/{scanName}')
path_format_arguments = {
'dataSourceName': _SERIALIZER.url("data_source_name", data_source_name, 'str'),
'scanName': _SERIALIZER.url("scan_name", scan_name, 'str'),
}
url = _format_url_section(url, **path_format_arguments)
# Construct parameters
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="DELETE",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)
[docs]def build_list_by_data_source_request(
data_source_name: str,
**kwargs: Any
) -> HttpRequest:
"""List scans in data source.
See https://aka.ms/azsdk/python/protocol/quickstart for how to incorporate this request builder into your code flow.
:param data_source_name:
:type data_source_name: str
:return: Returns an :class:`~azure.purview.scanning.core.rest.HttpRequest` that you will pass to the client's `send_request` method.
See https://aka.ms/azsdk/python/protocol/quickstart for how to incorporate this response into your code flow.
:rtype: ~azure.purview.scanning.core.rest.HttpRequest
Example:
.. code-block:: python
# response body for status code(s): 200
response_body == {
"count": "long (optional)",
"nextLink": "str (optional)",
"value": [
{
"kind": "Scan",
"scanResults": [
{
"assetsClassified": "long (optional)",
"assetsDiscovered": "long (optional)",
"dataSourceType": "str (optional)",
"diagnostics": {},
"endTime": "datetime (optional)",
"error": {},
"errorMessage": "str (optional)",
"id": "str (optional)",
"parentId": "str (optional)",
"pipelineStartTime": "datetime (optional)",
"queuedTime": "datetime (optional)",
"resourceId": "str (optional)",
"runType": "str (optional)",
"scanLevelType": "str (optional)",
"scanRulesetType": "str (optional)",
"scanRulesetVersion": "int (optional)",
"startTime": "datetime (optional)",
"status": "str (optional)"
}
]
}
]
}
"""
api_version = "2018-12-01-preview"
accept = "application/json"
# Construct URL
url = kwargs.pop("template_url", '/datasources/{dataSourceName}/scans')
path_format_arguments = {
'dataSourceName': _SERIALIZER.url("data_source_name", data_source_name, 'str'),
}
url = _format_url_section(url, **path_format_arguments)
# Construct parameters
query_parameters = kwargs.pop("params", {}) # type: Dict[str, Any]
query_parameters['api-version'] = _SERIALIZER.query("api_version", api_version, 'str')
# Construct headers
header_parameters = kwargs.pop("headers", {}) # type: Dict[str, Any]
header_parameters['Accept'] = _SERIALIZER.header("accept", accept, 'str')
return HttpRequest(
method="GET",
url=url,
params=query_parameters,
headers=header_parameters,
**kwargs
)