# 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 msrest.pipeline import ClientRawResponse
from .. import models
[docs]class SnapshotOperations(object):
"""SnapshotOperations operations.
You should not instantiate directly this class, but create a Client instance that will create it for you and attach it as attribute.
:param client: Client for service requests.
:param config: Configuration of service client.
:param serializer: An object model serializer.
:param deserializer: An object model deserializer.
"""
models = models
def __init__(self, client, config, serializer, deserializer):
self._client = client
self._serialize = serializer
self._deserialize = deserializer
self.config = config
[docs] def take(
self, type, object_id, apply_scope, user_data=None, custom_headers=None, raw=False, **operation_config):
"""Submit an operation to take a snapshot of face list, large face list,
person group or large person group, with user-specified snapshot type,
source object id, apply scope and an optional user data.<br />
The snapshot interfaces are for users to backup and restore their face
data from one face subscription to another, inside same region or
across regions. The workflow contains two phases, user first calls
Snapshot - Take to create a copy of the source object and store it as a
snapshot, then calls Snapshot - Apply to paste the snapshot to target
subscription. The snapshots are stored in a centralized location (per
Azure instance), so that they can be applied cross accounts and
regions.<br />
Taking snapshot is an asynchronous operation. An operation id can be
obtained from the "Operation-Location" field in response header, to be
used in OperationStatus - Get for tracking the progress of creating the
snapshot. The snapshot id will be included in the "resourceLocation"
field in OperationStatus - Get response when the operation status is
"succeeded".<br />
Snapshot taking time depends on the number of person and face entries
in the source object. It could be in seconds, or up to several hours
for 1,000,000 persons with multiple faces.<br />
Snapshots will be automatically expired and cleaned in 48 hours after
it is created by Snapshot - Take. User can delete the snapshot using
Snapshot - Delete by themselves any time before expiration.<br />
Taking snapshot for a certain object will not block any other
operations against the object. All read-only operations (Get/List and
Identify/FindSimilar/Verify) can be conducted as usual. For all
writable operations, including Add/Update/Delete the source object or
its persons/faces and Train, they are not blocked but not recommended
because writable updates may not be reflected on the snapshot during
its taking. After snapshot taking is completed, all readable and
writable operations can work as normal. Snapshot will also include the
training results of the source object, which means target subscription
the snapshot applied to does not need re-train the target object before
calling Identify/FindSimilar.<br />
* Free-tier subscription quota: 100 take operations per month.
* S0-tier subscription quota: 100 take operations per day.
:param type: User specified type for the source object to take
snapshot from. Currently FaceList, PersonGroup, LargeFaceList and
LargePersonGroup are supported. Possible values include: 'FaceList',
'LargeFaceList', 'LargePersonGroup', 'PersonGroup'
:type type: str or
~azure.cognitiveservices.vision.face.models.SnapshotObjectType
:param object_id: User specified source object id to take snapshot
from.
:type object_id: str
:param apply_scope: User specified array of target Face subscription
ids for the snapshot. For each snapshot, only subscriptions included
in the applyScope of Snapshot - Take can apply it.
:type apply_scope: list[str]
:param user_data: User specified data about the snapshot for any
purpose. Length should not exceed 16KB.
:type user_data: str
:param dict custom_headers: headers that will be added to the request
:param bool raw: returns the direct response alongside the
deserialized response
:param operation_config: :ref:`Operation configuration
overrides<msrest:optionsforoperations>`.
:return: None or ClientRawResponse if raw=true
:rtype: None or ~msrest.pipeline.ClientRawResponse
:raises:
:class:`APIErrorException<azure.cognitiveservices.vision.face.models.APIErrorException>`
"""
body = models.TakeSnapshotRequest(type=type, object_id=object_id, apply_scope=apply_scope, user_data=user_data)
# Construct URL
url = self.take.metadata['url']
path_format_arguments = {
'Endpoint': self._serialize.url("self.config.endpoint", self.config.endpoint, 'str', skip_quote=True)
}
url = self._client.format_url(url, **path_format_arguments)
# Construct parameters
query_parameters = {}
# Construct headers
header_parameters = {}
header_parameters['Content-Type'] = 'application/json; charset=utf-8'
if custom_headers:
header_parameters.update(custom_headers)
# Construct body
body_content = self._serialize.body(body, 'TakeSnapshotRequest')
# Construct and send request
request = self._client.post(url, query_parameters, header_parameters, body_content)
response = self._client.send(request, stream=False, **operation_config)
if response.status_code not in [202]:
raise models.APIErrorException(self._deserialize, response)
if raw:
client_raw_response = ClientRawResponse(None, response)
client_raw_response.add_headers({
'Operation-Location': 'str',
})
return client_raw_response
take.metadata = {'url': '/snapshots'}
[docs] def list(
self, type=None, apply_scope=None, custom_headers=None, raw=False, **operation_config):
"""List all accessible snapshots with related information, including
snapshots that were taken by the user, or snapshots to be applied to
the user (subscription id was included in the applyScope in Snapshot -
Take).
:param type: User specified object type as a search filter. Possible
values include: 'FaceList', 'LargeFaceList', 'LargePersonGroup',
'PersonGroup'
:type type: str or
~azure.cognitiveservices.vision.face.models.SnapshotObjectType
:param apply_scope: User specified snapshot apply scopes as a search
filter. ApplyScope is an array of the target Azure subscription ids
for the snapshot, specified by the user who created the snapshot by
Snapshot - Take.
:type apply_scope: list[str]
:param dict custom_headers: headers that will be added to the request
:param bool raw: returns the direct response alongside the
deserialized response
:param operation_config: :ref:`Operation configuration
overrides<msrest:optionsforoperations>`.
:return: list or ClientRawResponse if raw=true
:rtype: list[~azure.cognitiveservices.vision.face.models.Snapshot] or
~msrest.pipeline.ClientRawResponse
:raises:
:class:`APIErrorException<azure.cognitiveservices.vision.face.models.APIErrorException>`
"""
# Construct URL
url = self.list.metadata['url']
path_format_arguments = {
'Endpoint': self._serialize.url("self.config.endpoint", self.config.endpoint, 'str', skip_quote=True)
}
url = self._client.format_url(url, **path_format_arguments)
# Construct parameters
query_parameters = {}
if type is not None:
query_parameters['type'] = self._serialize.query("type", type, 'SnapshotObjectType')
if apply_scope is not None:
query_parameters['applyScope'] = self._serialize.query("apply_scope", apply_scope, '[str]', div=',')
# Construct headers
header_parameters = {}
header_parameters['Accept'] = 'application/json'
if custom_headers:
header_parameters.update(custom_headers)
# Construct and send request
request = self._client.get(url, query_parameters, header_parameters)
response = self._client.send(request, stream=False, **operation_config)
if response.status_code not in [200]:
raise models.APIErrorException(self._deserialize, response)
deserialized = None
if response.status_code == 200:
deserialized = self._deserialize('[Snapshot]', response)
if raw:
client_raw_response = ClientRawResponse(deserialized, response)
return client_raw_response
return deserialized
list.metadata = {'url': '/snapshots'}
[docs] def get(
self, snapshot_id, custom_headers=None, raw=False, **operation_config):
"""Retrieve information about a snapshot. Snapshot is only accessible to
the source subscription who took it, and target subscriptions included
in the applyScope in Snapshot - Take.
:param snapshot_id: Id referencing a particular snapshot.
:type snapshot_id: str
:param dict custom_headers: headers that will be added to the request
:param bool raw: returns the direct response alongside the
deserialized response
:param operation_config: :ref:`Operation configuration
overrides<msrest:optionsforoperations>`.
:return: Snapshot or ClientRawResponse if raw=true
:rtype: ~azure.cognitiveservices.vision.face.models.Snapshot or
~msrest.pipeline.ClientRawResponse
:raises:
:class:`APIErrorException<azure.cognitiveservices.vision.face.models.APIErrorException>`
"""
# Construct URL
url = self.get.metadata['url']
path_format_arguments = {
'Endpoint': self._serialize.url("self.config.endpoint", self.config.endpoint, 'str', skip_quote=True),
'snapshotId': self._serialize.url("snapshot_id", snapshot_id, 'str')
}
url = self._client.format_url(url, **path_format_arguments)
# Construct parameters
query_parameters = {}
# Construct headers
header_parameters = {}
header_parameters['Accept'] = 'application/json'
if custom_headers:
header_parameters.update(custom_headers)
# Construct and send request
request = self._client.get(url, query_parameters, header_parameters)
response = self._client.send(request, stream=False, **operation_config)
if response.status_code not in [200]:
raise models.APIErrorException(self._deserialize, response)
deserialized = None
if response.status_code == 200:
deserialized = self._deserialize('Snapshot', response)
if raw:
client_raw_response = ClientRawResponse(deserialized, response)
return client_raw_response
return deserialized
get.metadata = {'url': '/snapshots/{snapshotId}'}
[docs] def update(
self, snapshot_id, apply_scope=None, user_data=None, custom_headers=None, raw=False, **operation_config):
"""Update the information of a snapshot. Only the source subscription who
took the snapshot can update the snapshot.
:param snapshot_id: Id referencing a particular snapshot.
:type snapshot_id: str
:param apply_scope: Array of the target Face subscription ids for the
snapshot, specified by the user who created the snapshot when calling
Snapshot - Take. For each snapshot, only subscriptions included in the
applyScope of Snapshot - Take can apply it.
:type apply_scope: list[str]
:param user_data: User specified data about the snapshot for any
purpose. Length should not exceed 16KB.
:type user_data: str
:param dict custom_headers: headers that will be added to the request
:param bool raw: returns the direct response alongside the
deserialized response
:param operation_config: :ref:`Operation configuration
overrides<msrest:optionsforoperations>`.
:return: None or ClientRawResponse if raw=true
:rtype: None or ~msrest.pipeline.ClientRawResponse
:raises:
:class:`APIErrorException<azure.cognitiveservices.vision.face.models.APIErrorException>`
"""
body = models.UpdateSnapshotRequest(apply_scope=apply_scope, user_data=user_data)
# Construct URL
url = self.update.metadata['url']
path_format_arguments = {
'Endpoint': self._serialize.url("self.config.endpoint", self.config.endpoint, 'str', skip_quote=True),
'snapshotId': self._serialize.url("snapshot_id", snapshot_id, 'str')
}
url = self._client.format_url(url, **path_format_arguments)
# Construct parameters
query_parameters = {}
# Construct headers
header_parameters = {}
header_parameters['Content-Type'] = 'application/json; charset=utf-8'
if custom_headers:
header_parameters.update(custom_headers)
# Construct body
body_content = self._serialize.body(body, 'UpdateSnapshotRequest')
# Construct and send request
request = self._client.patch(url, query_parameters, header_parameters, body_content)
response = self._client.send(request, stream=False, **operation_config)
if response.status_code not in [200]:
raise models.APIErrorException(self._deserialize, response)
if raw:
client_raw_response = ClientRawResponse(None, response)
return client_raw_response
update.metadata = {'url': '/snapshots/{snapshotId}'}
[docs] def delete(
self, snapshot_id, custom_headers=None, raw=False, **operation_config):
"""Delete an existing snapshot according to the snapshotId. All object
data and information in the snapshot will also be deleted. Only the
source subscription who took the snapshot can delete the snapshot. If
the user does not delete a snapshot with this API, the snapshot will
still be automatically deleted in 48 hours after creation.
:param snapshot_id: Id referencing a particular snapshot.
:type snapshot_id: str
:param dict custom_headers: headers that will be added to the request
:param bool raw: returns the direct response alongside the
deserialized response
:param operation_config: :ref:`Operation configuration
overrides<msrest:optionsforoperations>`.
:return: None or ClientRawResponse if raw=true
:rtype: None or ~msrest.pipeline.ClientRawResponse
:raises:
:class:`APIErrorException<azure.cognitiveservices.vision.face.models.APIErrorException>`
"""
# Construct URL
url = self.delete.metadata['url']
path_format_arguments = {
'Endpoint': self._serialize.url("self.config.endpoint", self.config.endpoint, 'str', skip_quote=True),
'snapshotId': self._serialize.url("snapshot_id", snapshot_id, 'str')
}
url = self._client.format_url(url, **path_format_arguments)
# Construct parameters
query_parameters = {}
# Construct headers
header_parameters = {}
if custom_headers:
header_parameters.update(custom_headers)
# Construct and send request
request = self._client.delete(url, query_parameters, header_parameters)
response = self._client.send(request, stream=False, **operation_config)
if response.status_code not in [200]:
raise models.APIErrorException(self._deserialize, response)
if raw:
client_raw_response = ClientRawResponse(None, response)
return client_raw_response
delete.metadata = {'url': '/snapshots/{snapshotId}'}
[docs] def apply(
self, snapshot_id, object_id, mode="CreateNew", custom_headers=None, raw=False, **operation_config):
"""Submit an operation to apply a snapshot to current subscription. For
each snapshot, only subscriptions included in the applyScope of
Snapshot - Take can apply it.<br />
The snapshot interfaces are for users to backup and restore their face
data from one face subscription to another, inside same region or
across regions. The workflow contains two phases, user first calls
Snapshot - Take to create a copy of the source object and store it as a
snapshot, then calls Snapshot - Apply to paste the snapshot to target
subscription. The snapshots are stored in a centralized location (per
Azure instance), so that they can be applied cross accounts and
regions.<br />
Applying snapshot is an asynchronous operation. An operation id can be
obtained from the "Operation-Location" field in response header, to be
used in OperationStatus - Get for tracking the progress of applying the
snapshot. The target object id will be included in the
"resourceLocation" field in OperationStatus - Get response when the
operation status is "succeeded".<br />
Snapshot applying time depends on the number of person and face entries
in the snapshot object. It could be in seconds, or up to 1 hour for
1,000,000 persons with multiple faces.<br />
Snapshots will be automatically expired and cleaned in 48 hours after
it is created by Snapshot - Take. So the target subscription is
required to apply the snapshot in 48 hours since its creation.<br />
Applying a snapshot will not block any other operations against the
target object, however it is not recommended because the correctness
cannot be guaranteed during snapshot applying. After snapshot applying
is completed, all operations towards the target object can work as
normal. Snapshot also includes the training results of the source
object, which means target subscription the snapshot applied to does
not need re-train the target object before calling
Identify/FindSimilar.<br />
One snapshot can be applied multiple times in parallel, while currently
only CreateNew apply mode is supported, which means the apply operation
will fail if target subscription already contains an object of same
type and using the same objectId. Users can specify the "objectId" in
request body to avoid such conflicts.<br />
* Free-tier subscription quota: 100 apply operations per month.
* S0-tier subscription quota: 100 apply operations per day.
:param snapshot_id: Id referencing a particular snapshot.
:type snapshot_id: str
:param object_id: User specified target object id to be created from
the snapshot.
:type object_id: str
:param mode: Snapshot applying mode. Currently only CreateNew is
supported, which means the apply operation will fail if target
subscription already contains an object of same type and using the
same objectId. Users can specify the "objectId" in request body to
avoid such conflicts. Possible values include: 'CreateNew'
:type mode: str or
~azure.cognitiveservices.vision.face.models.SnapshotApplyMode
:param dict custom_headers: headers that will be added to the request
:param bool raw: returns the direct response alongside the
deserialized response
:param operation_config: :ref:`Operation configuration
overrides<msrest:optionsforoperations>`.
:return: None or ClientRawResponse if raw=true
:rtype: None or ~msrest.pipeline.ClientRawResponse
:raises:
:class:`APIErrorException<azure.cognitiveservices.vision.face.models.APIErrorException>`
"""
body = models.ApplySnapshotRequest(object_id=object_id, mode=mode)
# Construct URL
url = self.apply.metadata['url']
path_format_arguments = {
'Endpoint': self._serialize.url("self.config.endpoint", self.config.endpoint, 'str', skip_quote=True),
'snapshotId': self._serialize.url("snapshot_id", snapshot_id, 'str')
}
url = self._client.format_url(url, **path_format_arguments)
# Construct parameters
query_parameters = {}
# Construct headers
header_parameters = {}
header_parameters['Content-Type'] = 'application/json; charset=utf-8'
if custom_headers:
header_parameters.update(custom_headers)
# Construct body
body_content = self._serialize.body(body, 'ApplySnapshotRequest')
# Construct and send request
request = self._client.post(url, query_parameters, header_parameters, body_content)
response = self._client.send(request, stream=False, **operation_config)
if response.status_code not in [202]:
raise models.APIErrorException(self._deserialize, response)
if raw:
client_raw_response = ClientRawResponse(None, response)
client_raw_response.add_headers({
'Operation-Location': 'str',
})
return client_raw_response
apply.metadata = {'url': '/snapshots/{snapshotId}/apply'}
[docs] def get_operation_status(
self, operation_id, custom_headers=None, raw=False, **operation_config):
"""Retrieve the status of a take/apply snapshot operation.
:param operation_id: Id referencing a particular take/apply snapshot
operation.
:type operation_id: str
:param dict custom_headers: headers that will be added to the request
:param bool raw: returns the direct response alongside the
deserialized response
:param operation_config: :ref:`Operation configuration
overrides<msrest:optionsforoperations>`.
:return: OperationStatus or ClientRawResponse if raw=true
:rtype: ~azure.cognitiveservices.vision.face.models.OperationStatus or
~msrest.pipeline.ClientRawResponse
:raises:
:class:`APIErrorException<azure.cognitiveservices.vision.face.models.APIErrorException>`
"""
# Construct URL
url = self.get_operation_status.metadata['url']
path_format_arguments = {
'Endpoint': self._serialize.url("self.config.endpoint", self.config.endpoint, 'str', skip_quote=True),
'operationId': self._serialize.url("operation_id", operation_id, 'str')
}
url = self._client.format_url(url, **path_format_arguments)
# Construct parameters
query_parameters = {}
# Construct headers
header_parameters = {}
header_parameters['Accept'] = 'application/json'
if custom_headers:
header_parameters.update(custom_headers)
# Construct and send request
request = self._client.get(url, query_parameters, header_parameters)
response = self._client.send(request, stream=False, **operation_config)
if response.status_code not in [200]:
raise models.APIErrorException(self._deserialize, response)
deserialized = None
if response.status_code == 200:
deserialized = self._deserialize('OperationStatus', response)
if raw:
client_raw_response = ClientRawResponse(deserialized, response)
return client_raw_response
return deserialized
get_operation_status.metadata = {'url': '/operations/{operationId}'}