Source code for azure.agrifood.farming.operations._seasonal_fields_operations

# 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.
# --------------------------------------------------------------------------
import datetime
from typing import TYPE_CHECKING
import warnings

from azure.core.exceptions import ClientAuthenticationError, HttpResponseError, ResourceExistsError, ResourceNotFoundError, map_error
from azure.core.paging import ItemPaged
from azure.core.pipeline import PipelineResponse
from azure.core.pipeline.transport import HttpRequest, HttpResponse
from azure.core.polling import LROPoller, NoPolling, PollingMethod
from azure.core.polling.base_polling import LROBasePolling

from .. import models as _models

if TYPE_CHECKING:
    # pylint: disable=unused-import,ungrouped-imports
    from typing import Any, Callable, Dict, Generic, Iterable, List, Optional, TypeVar, Union

    T = TypeVar('T')
    ClsType = Optional[Callable[[PipelineResponse[HttpRequest, HttpResponse], T, Dict[str, Any]], Any]]

[docs]class SeasonalFieldsOperations(object): """SeasonalFieldsOperations operations. You should not instantiate this class directly. Instead, you should create a Client instance that instantiates it for you and attaches it as an attribute. :ivar models: Alias to model classes used in this operation group. :type models: ~azure.agrifood.farming.models :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 list_by_farmer_id( self, farmer_id, # type: str farm_ids=None, # type: Optional[List[str]] field_ids=None, # type: Optional[List[str]] season_ids=None, # type: Optional[List[str]] crop_variety_ids=None, # type: Optional[List[str]] crop_ids=None, # type: Optional[List[str]] min_avg_yield_value=None, # type: Optional[float] max_avg_yield_value=None, # type: Optional[float] avg_yield_unit=None, # type: Optional[str] min_avg_seed_population_value=None, # type: Optional[float] max_avg_seed_population_value=None, # type: Optional[float] avg_seed_population_unit=None, # type: Optional[str] min_planting_date_time=None, # type: Optional[datetime.datetime] max_planting_date_time=None, # type: Optional[datetime.datetime] ids=None, # type: Optional[List[str]] names=None, # type: Optional[List[str]] property_filters=None, # type: Optional[List[str]] statuses=None, # type: Optional[List[str]] min_created_date_time=None, # type: Optional[datetime.datetime] max_created_date_time=None, # type: Optional[datetime.datetime] min_last_modified_date_time=None, # type: Optional[datetime.datetime] max_last_modified_date_time=None, # type: Optional[datetime.datetime] max_page_size=50, # type: Optional[int] skip_token=None, # type: Optional[str] **kwargs # type: Any ): # type: (...) -> Iterable["_models.SeasonalFieldListResponse"] """Returns a paginated list of seasonal field resources under a particular farmer. :param farmer_id: ID of the associated farmer. :type farmer_id: str :param farm_ids: Farm Ids of the resource. :type farm_ids: list[str] :param field_ids: Field Ids of the resource. :type field_ids: list[str] :param season_ids: Season Ids of the resource. :type season_ids: list[str] :param crop_variety_ids: CropVarietyIds of the resource. :type crop_variety_ids: list[str] :param crop_ids: Ids of the crop it belongs to. :type crop_ids: list[str] :param min_avg_yield_value: Minimum average yield value of the seasonal field(inclusive). :type min_avg_yield_value: float :param max_avg_yield_value: Maximum average yield value of the seasonal field(inclusive). :type max_avg_yield_value: float :param avg_yield_unit: Unit of the average yield value attribute. :type avg_yield_unit: str :param min_avg_seed_population_value: Minimum average seed population value of the seasonal field(inclusive). :type min_avg_seed_population_value: float :param max_avg_seed_population_value: Maximum average seed population value of the seasonal field(inclusive). :type max_avg_seed_population_value: float :param avg_seed_population_unit: Unit of average seed population value attribute. :type avg_seed_population_unit: str :param min_planting_date_time: Minimum planting datetime, sample format: yyyy-MM-ddTHH:mm:ssZ. :type min_planting_date_time: ~datetime.datetime :param max_planting_date_time: Maximum planting datetime, sample format: yyyy-MM-ddTHH:mm:ssZ. :type max_planting_date_time: ~datetime.datetime :param ids: Ids of the resource. :type ids: list[str] :param names: Names of the resource. :type names: list[str] :param property_filters: Filters on key-value pairs within the Properties object. eg. "{testKey} eq {testValue}". :type property_filters: list[str] :param statuses: Statuses of the resource. :type statuses: list[str] :param min_created_date_time: Minimum creation date of resource (inclusive). :type min_created_date_time: ~datetime.datetime :param max_created_date_time: Maximum creation date of resource (inclusive). :type max_created_date_time: ~datetime.datetime :param min_last_modified_date_time: Minimum last modified date of resource (inclusive). :type min_last_modified_date_time: ~datetime.datetime :param max_last_modified_date_time: Maximum last modified date of resource (inclusive). :type max_last_modified_date_time: ~datetime.datetime :param max_page_size: Maximum number of items needed (inclusive). Minimum = 10, Maximum = 1000, Default value = 50. :type max_page_size: int :param skip_token: Skip token for getting next set of results. :type skip_token: str :keyword callable cls: A custom type or function that will be passed the direct response :return: An iterator like instance of either SeasonalFieldListResponse or the result of cls(response) :rtype: ~azure.core.paging.ItemPaged[~azure.agrifood.farming.models.SeasonalFieldListResponse] :raises: ~azure.core.exceptions.HttpResponseError """ cls = kwargs.pop('cls', None) # type: ClsType["_models.SeasonalFieldListResponse"] error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {})) api_version = "2021-03-31-preview" accept = "application/json" def prepare_request(next_link=None): # Construct headers header_parameters = {} # type: Dict[str, Any] header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') if not next_link: # Construct URL url = self.list_by_farmer_id.metadata['url'] # type: ignore path_format_arguments = { 'Endpoint': self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), 'farmerId': self._serialize.url("farmer_id", farmer_id, 'str'), } url = self._client.format_url(url, **path_format_arguments) # Construct parameters query_parameters = {} # type: Dict[str, Any] if farm_ids is not None: query_parameters['farmIds'] = [self._serialize.query("farm_ids", q, 'str') if q is not None else '' for q in farm_ids] if field_ids is not None: query_parameters['fieldIds'] = [self._serialize.query("field_ids", q, 'str') if q is not None else '' for q in field_ids] if season_ids is not None: query_parameters['seasonIds'] = [self._serialize.query("season_ids", q, 'str') if q is not None else '' for q in season_ids] if crop_variety_ids is not None: query_parameters['cropVarietyIds'] = [self._serialize.query("crop_variety_ids", q, 'str') if q is not None else '' for q in crop_variety_ids] if crop_ids is not None: query_parameters['cropIds'] = [self._serialize.query("crop_ids", q, 'str') if q is not None else '' for q in crop_ids] if min_avg_yield_value is not None: query_parameters['minAvgYieldValue'] = self._serialize.query("min_avg_yield_value", min_avg_yield_value, 'float') if max_avg_yield_value is not None: query_parameters['maxAvgYieldValue'] = self._serialize.query("max_avg_yield_value", max_avg_yield_value, 'float') if avg_yield_unit is not None: query_parameters['avgYieldUnit'] = self._serialize.query("avg_yield_unit", avg_yield_unit, 'str') if min_avg_seed_population_value is not None: query_parameters['minAvgSeedPopulationValue'] = self._serialize.query("min_avg_seed_population_value", min_avg_seed_population_value, 'float') if max_avg_seed_population_value is not None: query_parameters['maxAvgSeedPopulationValue'] = self._serialize.query("max_avg_seed_population_value", max_avg_seed_population_value, 'float') if avg_seed_population_unit is not None: query_parameters['avgSeedPopulationUnit'] = self._serialize.query("avg_seed_population_unit", avg_seed_population_unit, 'str') if min_planting_date_time is not None: query_parameters['minPlantingDateTime'] = self._serialize.query("min_planting_date_time", min_planting_date_time, 'iso-8601') if max_planting_date_time is not None: query_parameters['maxPlantingDateTime'] = self._serialize.query("max_planting_date_time", max_planting_date_time, 'iso-8601') if ids is not None: query_parameters['ids'] = [self._serialize.query("ids", q, 'str') if q is not None else '' for q in ids] if names is not None: query_parameters['names'] = [self._serialize.query("names", q, 'str') if q is not None else '' for q in names] if property_filters is not None: query_parameters['propertyFilters'] = [self._serialize.query("property_filters", q, 'str') if q is not None else '' for q in property_filters] if statuses is not None: query_parameters['statuses'] = [self._serialize.query("statuses", q, 'str') if q is not None else '' for q in statuses] if min_created_date_time is not None: query_parameters['minCreatedDateTime'] = self._serialize.query("min_created_date_time", min_created_date_time, 'iso-8601') if max_created_date_time is not None: query_parameters['maxCreatedDateTime'] = self._serialize.query("max_created_date_time", max_created_date_time, 'iso-8601') if min_last_modified_date_time is not None: query_parameters['minLastModifiedDateTime'] = self._serialize.query("min_last_modified_date_time", min_last_modified_date_time, 'iso-8601') if max_last_modified_date_time is not None: query_parameters['maxLastModifiedDateTime'] = self._serialize.query("max_last_modified_date_time", max_last_modified_date_time, 'iso-8601') if max_page_size is not None: query_parameters['$maxPageSize'] = self._serialize.query("max_page_size", max_page_size, 'int', maximum=1000, minimum=10) if skip_token is not None: query_parameters['$skipToken'] = self._serialize.query("skip_token", skip_token, 'str') query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') request = self._client.get(url, query_parameters, header_parameters) else: url = next_link query_parameters = {} # type: Dict[str, Any] path_format_arguments = { 'Endpoint': self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), 'farmerId': self._serialize.url("farmer_id", farmer_id, 'str'), } url = self._client.format_url(url, **path_format_arguments) request = self._client.get(url, query_parameters, header_parameters) return request def extract_data(pipeline_response): deserialized = self._deserialize('SeasonalFieldListResponse', pipeline_response) list_of_elem = deserialized.value if cls: list_of_elem = cls(list_of_elem) return deserialized.next_link or None, iter(list_of_elem) def get_next(next_link=None): request = prepare_request(next_link) pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) response = pipeline_response.http_response if response.status_code not in [200]: error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, response) map_error(status_code=response.status_code, response=response, error_map=error_map) raise HttpResponseError(response=response, model=error) return pipeline_response return ItemPaged( get_next, extract_data )
list_by_farmer_id.metadata = {'url': '/farmers/{farmerId}/seasonal-fields'} # type: ignore
[docs] def list( self, farm_ids=None, # type: Optional[List[str]] field_ids=None, # type: Optional[List[str]] season_ids=None, # type: Optional[List[str]] crop_variety_ids=None, # type: Optional[List[str]] crop_ids=None, # type: Optional[List[str]] min_avg_yield_value=None, # type: Optional[float] max_avg_yield_value=None, # type: Optional[float] avg_yield_unit=None, # type: Optional[str] min_avg_seed_population_value=None, # type: Optional[float] max_avg_seed_population_value=None, # type: Optional[float] avg_seed_population_unit=None, # type: Optional[str] min_planting_date_time=None, # type: Optional[datetime.datetime] max_planting_date_time=None, # type: Optional[datetime.datetime] ids=None, # type: Optional[List[str]] names=None, # type: Optional[List[str]] property_filters=None, # type: Optional[List[str]] statuses=None, # type: Optional[List[str]] min_created_date_time=None, # type: Optional[datetime.datetime] max_created_date_time=None, # type: Optional[datetime.datetime] min_last_modified_date_time=None, # type: Optional[datetime.datetime] max_last_modified_date_time=None, # type: Optional[datetime.datetime] max_page_size=50, # type: Optional[int] skip_token=None, # type: Optional[str] **kwargs # type: Any ): # type: (...) -> Iterable["_models.SeasonalFieldListResponse"] """Returns a paginated list of seasonal field resources across all farmers. :param farm_ids: Farm Ids of the resource. :type farm_ids: list[str] :param field_ids: Field Ids of the resource. :type field_ids: list[str] :param season_ids: Season Ids of the resource. :type season_ids: list[str] :param crop_variety_ids: CropVarietyIds of the resource. :type crop_variety_ids: list[str] :param crop_ids: Ids of the crop it belongs to. :type crop_ids: list[str] :param min_avg_yield_value: Minimum average yield value of the seasonal field(inclusive). :type min_avg_yield_value: float :param max_avg_yield_value: Maximum average yield value of the seasonal field(inclusive). :type max_avg_yield_value: float :param avg_yield_unit: Unit of the average yield value attribute. :type avg_yield_unit: str :param min_avg_seed_population_value: Minimum average seed population value of the seasonal field(inclusive). :type min_avg_seed_population_value: float :param max_avg_seed_population_value: Maximum average seed population value of the seasonal field(inclusive). :type max_avg_seed_population_value: float :param avg_seed_population_unit: Unit of average seed population value attribute. :type avg_seed_population_unit: str :param min_planting_date_time: Minimum planting datetime, sample format: yyyy-MM-ddTHH:mm:ssZ. :type min_planting_date_time: ~datetime.datetime :param max_planting_date_time: Maximum planting datetime, sample format: yyyy-MM-ddTHH:mm:ssZ. :type max_planting_date_time: ~datetime.datetime :param ids: Ids of the resource. :type ids: list[str] :param names: Names of the resource. :type names: list[str] :param property_filters: Filters on key-value pairs within the Properties object. eg. "{testKey} eq {testValue}". :type property_filters: list[str] :param statuses: Statuses of the resource. :type statuses: list[str] :param min_created_date_time: Minimum creation date of resource (inclusive). :type min_created_date_time: ~datetime.datetime :param max_created_date_time: Maximum creation date of resource (inclusive). :type max_created_date_time: ~datetime.datetime :param min_last_modified_date_time: Minimum last modified date of resource (inclusive). :type min_last_modified_date_time: ~datetime.datetime :param max_last_modified_date_time: Maximum last modified date of resource (inclusive). :type max_last_modified_date_time: ~datetime.datetime :param max_page_size: Maximum number of items needed (inclusive). Minimum = 10, Maximum = 1000, Default value = 50. :type max_page_size: int :param skip_token: Skip token for getting next set of results. :type skip_token: str :keyword callable cls: A custom type or function that will be passed the direct response :return: An iterator like instance of either SeasonalFieldListResponse or the result of cls(response) :rtype: ~azure.core.paging.ItemPaged[~azure.agrifood.farming.models.SeasonalFieldListResponse] :raises: ~azure.core.exceptions.HttpResponseError """ cls = kwargs.pop('cls', None) # type: ClsType["_models.SeasonalFieldListResponse"] error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {})) api_version = "2021-03-31-preview" accept = "application/json" def prepare_request(next_link=None): # Construct headers header_parameters = {} # type: Dict[str, Any] header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') if not next_link: # Construct URL url = self.list.metadata['url'] # type: ignore 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 = {} # type: Dict[str, Any] if farm_ids is not None: query_parameters['farmIds'] = [self._serialize.query("farm_ids", q, 'str') if q is not None else '' for q in farm_ids] if field_ids is not None: query_parameters['fieldIds'] = [self._serialize.query("field_ids", q, 'str') if q is not None else '' for q in field_ids] if season_ids is not None: query_parameters['seasonIds'] = [self._serialize.query("season_ids", q, 'str') if q is not None else '' for q in season_ids] if crop_variety_ids is not None: query_parameters['cropVarietyIds'] = [self._serialize.query("crop_variety_ids", q, 'str') if q is not None else '' for q in crop_variety_ids] if crop_ids is not None: query_parameters['cropIds'] = [self._serialize.query("crop_ids", q, 'str') if q is not None else '' for q in crop_ids] if min_avg_yield_value is not None: query_parameters['minAvgYieldValue'] = self._serialize.query("min_avg_yield_value", min_avg_yield_value, 'float') if max_avg_yield_value is not None: query_parameters['maxAvgYieldValue'] = self._serialize.query("max_avg_yield_value", max_avg_yield_value, 'float') if avg_yield_unit is not None: query_parameters['avgYieldUnit'] = self._serialize.query("avg_yield_unit", avg_yield_unit, 'str') if min_avg_seed_population_value is not None: query_parameters['minAvgSeedPopulationValue'] = self._serialize.query("min_avg_seed_population_value", min_avg_seed_population_value, 'float') if max_avg_seed_population_value is not None: query_parameters['maxAvgSeedPopulationValue'] = self._serialize.query("max_avg_seed_population_value", max_avg_seed_population_value, 'float') if avg_seed_population_unit is not None: query_parameters['avgSeedPopulationUnit'] = self._serialize.query("avg_seed_population_unit", avg_seed_population_unit, 'str') if min_planting_date_time is not None: query_parameters['minPlantingDateTime'] = self._serialize.query("min_planting_date_time", min_planting_date_time, 'iso-8601') if max_planting_date_time is not None: query_parameters['maxPlantingDateTime'] = self._serialize.query("max_planting_date_time", max_planting_date_time, 'iso-8601') if ids is not None: query_parameters['ids'] = [self._serialize.query("ids", q, 'str') if q is not None else '' for q in ids] if names is not None: query_parameters['names'] = [self._serialize.query("names", q, 'str') if q is not None else '' for q in names] if property_filters is not None: query_parameters['propertyFilters'] = [self._serialize.query("property_filters", q, 'str') if q is not None else '' for q in property_filters] if statuses is not None: query_parameters['statuses'] = [self._serialize.query("statuses", q, 'str') if q is not None else '' for q in statuses] if min_created_date_time is not None: query_parameters['minCreatedDateTime'] = self._serialize.query("min_created_date_time", min_created_date_time, 'iso-8601') if max_created_date_time is not None: query_parameters['maxCreatedDateTime'] = self._serialize.query("max_created_date_time", max_created_date_time, 'iso-8601') if min_last_modified_date_time is not None: query_parameters['minLastModifiedDateTime'] = self._serialize.query("min_last_modified_date_time", min_last_modified_date_time, 'iso-8601') if max_last_modified_date_time is not None: query_parameters['maxLastModifiedDateTime'] = self._serialize.query("max_last_modified_date_time", max_last_modified_date_time, 'iso-8601') if max_page_size is not None: query_parameters['$maxPageSize'] = self._serialize.query("max_page_size", max_page_size, 'int', maximum=1000, minimum=10) if skip_token is not None: query_parameters['$skipToken'] = self._serialize.query("skip_token", skip_token, 'str') query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') request = self._client.get(url, query_parameters, header_parameters) else: url = next_link query_parameters = {} # type: Dict[str, Any] 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) request = self._client.get(url, query_parameters, header_parameters) return request def extract_data(pipeline_response): deserialized = self._deserialize('SeasonalFieldListResponse', pipeline_response) list_of_elem = deserialized.value if cls: list_of_elem = cls(list_of_elem) return deserialized.next_link or None, iter(list_of_elem) def get_next(next_link=None): request = prepare_request(next_link) pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) response = pipeline_response.http_response if response.status_code not in [200]: error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, response) map_error(status_code=response.status_code, response=response, error_map=error_map) raise HttpResponseError(response=response, model=error) return pipeline_response return ItemPaged( get_next, extract_data )
list.metadata = {'url': '/seasonal-fields'} # type: ignore
[docs] def get( self, farmer_id, # type: str seasonal_field_id, # type: str **kwargs # type: Any ): # type: (...) -> "_models.SeasonalField" """Gets a specified seasonal field resource under a particular farmer. :param farmer_id: ID of the associated farmer. :type farmer_id: str :param seasonal_field_id: ID of the seasonal field. :type seasonal_field_id: str :keyword callable cls: A custom type or function that will be passed the direct response :return: SeasonalField, or the result of cls(response) :rtype: ~azure.agrifood.farming.models.SeasonalField :raises: ~azure.core.exceptions.HttpResponseError """ cls = kwargs.pop('cls', None) # type: ClsType["_models.SeasonalField"] error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {})) api_version = "2021-03-31-preview" accept = "application/json" # Construct URL url = self.get.metadata['url'] # type: ignore path_format_arguments = { 'Endpoint': self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), 'farmerId': self._serialize.url("farmer_id", farmer_id, 'str'), 'seasonalFieldId': self._serialize.url("seasonal_field_id", seasonal_field_id, 'str'), } url = self._client.format_url(url, **path_format_arguments) # Construct parameters query_parameters = {} # type: Dict[str, Any] query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') # Construct headers header_parameters = {} # type: Dict[str, Any] header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') request = self._client.get(url, query_parameters, header_parameters) pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, response) raise HttpResponseError(response=response, model=error) deserialized = self._deserialize('SeasonalField', pipeline_response) if cls: return cls(pipeline_response, deserialized, {}) return deserialized
get.metadata = {'url': '/farmers/{farmerId}/seasonal-fields/{seasonalFieldId}'} # type: ignore
[docs] def create_or_update( self, farmer_id, # type: str seasonal_field_id, # type: str seasonal_field=None, # type: Optional["_models.SeasonalField"] **kwargs # type: Any ): # type: (...) -> "_models.SeasonalField" """Creates or Updates a seasonal field resource under a particular farmer. :param farmer_id: ID of the associated farmer resource. :type farmer_id: str :param seasonal_field_id: ID of the seasonal field resource. :type seasonal_field_id: str :param seasonal_field: Seasonal field resource payload to create or update. :type seasonal_field: ~azure.agrifood.farming.models.SeasonalField :keyword callable cls: A custom type or function that will be passed the direct response :return: SeasonalField, or the result of cls(response) :rtype: ~azure.agrifood.farming.models.SeasonalField :raises: ~azure.core.exceptions.HttpResponseError """ cls = kwargs.pop('cls', None) # type: ClsType["_models.SeasonalField"] error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {})) api_version = "2021-03-31-preview" content_type = kwargs.pop("content_type", "application/merge-patch+json") accept = "application/json" # Construct URL url = self.create_or_update.metadata['url'] # type: ignore path_format_arguments = { 'Endpoint': self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), 'farmerId': self._serialize.url("farmer_id", farmer_id, 'str'), 'seasonalFieldId': self._serialize.url("seasonal_field_id", seasonal_field_id, 'str'), } url = self._client.format_url(url, **path_format_arguments) # Construct parameters query_parameters = {} # type: Dict[str, Any] query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') # Construct headers header_parameters = {} # type: Dict[str, Any] header_parameters['Content-Type'] = self._serialize.header("content_type", content_type, 'str') header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') body_content_kwargs = {} # type: Dict[str, Any] if seasonal_field is not None: body_content = self._serialize.body(seasonal_field, 'SeasonalField') else: body_content = None body_content_kwargs['content'] = body_content request = self._client.patch(url, query_parameters, header_parameters, **body_content_kwargs) pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) response = pipeline_response.http_response if response.status_code not in [200, 201]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, response) raise HttpResponseError(response=response, model=error) if response.status_code == 200: deserialized = self._deserialize('SeasonalField', pipeline_response) if response.status_code == 201: deserialized = self._deserialize('SeasonalField', pipeline_response) if cls: return cls(pipeline_response, deserialized, {}) return deserialized
create_or_update.metadata = {'url': '/farmers/{farmerId}/seasonal-fields/{seasonalFieldId}'} # type: ignore
[docs] def delete( self, farmer_id, # type: str seasonal_field_id, # type: str **kwargs # type: Any ): # type: (...) -> None """Deletes a specified seasonal-field resource under a particular farmer. :param farmer_id: ID of the farmer. :type farmer_id: str :param seasonal_field_id: ID of the seasonal field. :type seasonal_field_id: str :keyword callable cls: A custom type or function that will be passed the direct response :return: None, or the result of cls(response) :rtype: None :raises: ~azure.core.exceptions.HttpResponseError """ cls = kwargs.pop('cls', None) # type: ClsType[None] error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {})) api_version = "2021-03-31-preview" accept = "application/json" # Construct URL url = self.delete.metadata['url'] # type: ignore path_format_arguments = { 'Endpoint': self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), 'farmerId': self._serialize.url("farmer_id", farmer_id, 'str'), 'seasonalFieldId': self._serialize.url("seasonal_field_id", seasonal_field_id, 'str'), } url = self._client.format_url(url, **path_format_arguments) # Construct parameters query_parameters = {} # type: Dict[str, Any] query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') # Construct headers header_parameters = {} # type: Dict[str, Any] header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') request = self._client.delete(url, query_parameters, header_parameters) pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) response = pipeline_response.http_response if response.status_code not in [204]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, response) raise HttpResponseError(response=response, model=error) if cls: return cls(pipeline_response, None, {})
delete.metadata = {'url': '/farmers/{farmerId}/seasonal-fields/{seasonalFieldId}'} # type: ignore
[docs] def get_cascade_delete_job_details( self, job_id, # type: str **kwargs # type: Any ): # type: (...) -> "_models.CascadeDeleteJob" """Get cascade delete job for specified seasonal field. :param job_id: ID of the job. :type job_id: str :keyword callable cls: A custom type or function that will be passed the direct response :return: CascadeDeleteJob, or the result of cls(response) :rtype: ~azure.agrifood.farming.models.CascadeDeleteJob :raises: ~azure.core.exceptions.HttpResponseError """ cls = kwargs.pop('cls', None) # type: ClsType["_models.CascadeDeleteJob"] error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {})) api_version = "2021-03-31-preview" accept = "application/json" # Construct URL url = self.get_cascade_delete_job_details.metadata['url'] # type: ignore path_format_arguments = { 'Endpoint': self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), 'jobId': self._serialize.url("job_id", job_id, 'str'), } url = self._client.format_url(url, **path_format_arguments) # Construct parameters query_parameters = {} # type: Dict[str, Any] query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') # Construct headers header_parameters = {} # type: Dict[str, Any] header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') request = self._client.get(url, query_parameters, header_parameters) pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) response = pipeline_response.http_response if response.status_code not in [200]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, response) raise HttpResponseError(response=response, model=error) deserialized = self._deserialize('CascadeDeleteJob', pipeline_response) if cls: return cls(pipeline_response, deserialized, {}) return deserialized
get_cascade_delete_job_details.metadata = {'url': '/seasonal-fields/cascade-delete/{jobId}'} # type: ignore def _create_cascade_delete_job_initial( self, job_id, # type: str farmer_id, # type: str seasonal_field_id, # type: str **kwargs # type: Any ): # type: (...) -> "_models.CascadeDeleteJob" cls = kwargs.pop('cls', None) # type: ClsType["_models.CascadeDeleteJob"] error_map = { 401: ClientAuthenticationError, 404: ResourceNotFoundError, 409: ResourceExistsError } error_map.update(kwargs.pop('error_map', {})) api_version = "2021-03-31-preview" accept = "application/json" # Construct URL url = self._create_cascade_delete_job_initial.metadata['url'] # type: ignore path_format_arguments = { 'Endpoint': self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), 'jobId': self._serialize.url("job_id", job_id, 'str'), } url = self._client.format_url(url, **path_format_arguments) # Construct parameters query_parameters = {} # type: Dict[str, Any] query_parameters['farmerId'] = self._serialize.query("farmer_id", farmer_id, 'str') query_parameters['seasonalFieldId'] = self._serialize.query("seasonal_field_id", seasonal_field_id, 'str') query_parameters['api-version'] = self._serialize.query("api_version", api_version, 'str') # Construct headers header_parameters = {} # type: Dict[str, Any] header_parameters['Accept'] = self._serialize.header("accept", accept, 'str') request = self._client.put(url, query_parameters, header_parameters) pipeline_response = self._client._pipeline.run(request, stream=False, **kwargs) response = pipeline_response.http_response if response.status_code not in [202]: map_error(status_code=response.status_code, response=response, error_map=error_map) error = self._deserialize.failsafe_deserialize(_models.ErrorResponse, response) raise HttpResponseError(response=response, model=error) deserialized = self._deserialize('CascadeDeleteJob', pipeline_response) if cls: return cls(pipeline_response, deserialized, {}) return deserialized _create_cascade_delete_job_initial.metadata = {'url': '/seasonal-fields/cascade-delete/{jobId}'} # type: ignore
[docs] def begin_create_cascade_delete_job( self, job_id, # type: str farmer_id, # type: str seasonal_field_id, # type: str **kwargs # type: Any ): # type: (...) -> LROPoller["_models.CascadeDeleteJob"] """Create a cascade delete job for specified seasonal field. :param job_id: Job ID supplied by end user. :type job_id: str :param farmer_id: ID of the associated farmer. :type farmer_id: str :param seasonal_field_id: ID of the seasonalField to be deleted. :type seasonal_field_id: str :keyword callable cls: A custom type or function that will be passed the direct response :keyword str continuation_token: A continuation token to restart a poller from a saved state. :keyword polling: By default, your polling method will be LROBasePolling. Pass in False for this operation to not poll, or pass in your own initialized polling object for a personal polling strategy. :paramtype polling: bool or ~azure.core.polling.PollingMethod :keyword int polling_interval: Default waiting time between two polls for LRO operations if no Retry-After header is present. :return: An instance of LROPoller that returns either CascadeDeleteJob or the result of cls(response) :rtype: ~azure.core.polling.LROPoller[~azure.agrifood.farming.models.CascadeDeleteJob] :raises ~azure.core.exceptions.HttpResponseError: """ polling = kwargs.pop('polling', True) # type: Union[bool, PollingMethod] cls = kwargs.pop('cls', None) # type: ClsType["_models.CascadeDeleteJob"] lro_delay = kwargs.pop( 'polling_interval', self._config.polling_interval ) cont_token = kwargs.pop('continuation_token', None) # type: Optional[str] if cont_token is None: raw_result = self._create_cascade_delete_job_initial( job_id=job_id, farmer_id=farmer_id, seasonal_field_id=seasonal_field_id, cls=lambda x,y,z: x, **kwargs ) kwargs.pop('error_map', None) kwargs.pop('content_type', None) def get_long_running_output(pipeline_response): deserialized = self._deserialize('CascadeDeleteJob', pipeline_response) if cls: return cls(pipeline_response, deserialized, {}) return deserialized path_format_arguments = { 'Endpoint': self._serialize.url("self._config.endpoint", self._config.endpoint, 'str', skip_quote=True), 'jobId': self._serialize.url("job_id", job_id, 'str'), } if polling is True: polling_method = LROBasePolling(lro_delay, lro_options={'final-state-via': 'location'}, path_format_arguments=path_format_arguments, **kwargs) elif polling is False: polling_method = NoPolling() else: polling_method = polling if cont_token: return LROPoller.from_continuation_token( polling_method=polling_method, continuation_token=cont_token, client=self._client, deserialization_callback=get_long_running_output ) else: return LROPoller(self._client, raw_result, get_long_running_output, polling_method)
begin_create_cascade_delete_job.metadata = {'url': '/seasonal-fields/cascade-delete/{jobId}'} # type: ignore