def can_paginate(self, operation_name):
"""Check if an operation can be paginated.
:type operation_name: string
:param operation_name: The operation name. This is the same name
as the method name on the client. For example, if the
method name is ``create_foo``, and you'd normally invoke the
operation as ``client.create_foo(**kwargs)``, if the
``create_foo`` operation can be paginated, you can use the
call ``client.get_paginator("create_foo")``.
:return: ``True`` if the operation can be paginated,
``False`` otherwise.
"""
if 'page_config' not in self._cache:
try:
page_config = self._loader.load_service_model(
self._service_model.service_name,
'paginators-1',
self._service_model.api_version)['pagination']
self._cache['page_config'] = page_config
except DataNotFoundError:
self._cache['page_config'] = {}
actual_operation_name = self._PY_TO_OP_NAME[operation_name]
return actual_operation_name in self._cache['page_config']
python类DataNotFoundError()的实例源码
def _get_waiter_config(self):
if 'waiter_config' not in self._cache:
try:
waiter_config = self._loader.load_service_model(
self._service_model.service_name,
'waiters-2',
self._service_model.api_version)
self._cache['waiter_config'] = waiter_config
except DataNotFoundError:
self._cache['waiter_config'] = {}
return self._cache['waiter_config']
def client_api(self, section):
examples = None
try:
examples = self.get_examples(self._service_name)
except DataNotFoundError:
pass
ClientDocumenter(self._client, examples).document_client(section)
def paginator_api(self, section):
try:
service_paginator_model = self._session.get_paginator_model(
self._service_name)
except DataNotFoundError:
return
paginator_documenter = PaginatorDocumenter(
self._client, service_paginator_model)
paginator_documenter.document_paginators(section)
def list_api_versions(self, service_name, type_name):
"""List all API versions available for a particular service type
:type service_name: str
:param service_name: The name of the service
:type type_name: str
:param type_name: The type name for the service (i.e service-2,
paginators-1, etc.)
:rtype: list
:return: A list of API version strings in sorted order.
"""
known_api_versions = set()
for possible_path in self._potential_locations(service_name,
must_exist=True,
is_dir=True):
for dirname in os.listdir(possible_path):
full_path = os.path.join(possible_path, dirname, type_name)
# Only add to the known_api_versions if the directory
# contains a service-2, paginators-1, etc. file corresponding
# to the type_name passed in.
if self.file_loader.exists(full_path):
known_api_versions.add(dirname)
if not known_api_versions:
raise DataNotFoundError(data_path=service_name)
return sorted(known_api_versions)
def _find_extras(self, service_name, type_name, api_version):
"""Creates an iterator over all the extras data."""
for extras_type in self.extras_types:
extras_name = '%s.%s-extras' % (type_name, extras_type)
full_path = os.path.join(service_name, api_version, extras_name)
try:
yield self.load_data(full_path)
except DataNotFoundError:
pass
def client_api(self, section):
examples = None
try:
examples = self.get_examples(self._service_name)
except DataNotFoundError:
pass
Boto3ClientDocumenter(self._client, examples).document_client(section)
def can_paginate(self, operation_name):
"""Check if an operation can be paginated.
:type operation_name: string
:param operation_name: The operation name. This is the same name
as the method name on the client. For example, if the
method name is ``create_foo``, and you'd normally invoke the
operation as ``client.create_foo(**kwargs)``, if the
``create_foo`` operation can be paginated, you can use the
call ``client.get_paginator("create_foo")``.
:return: ``True`` if the operation can be paginated,
``False`` otherwise.
"""
if 'page_config' not in self._cache:
try:
page_config = self._loader.load_service_model(
self._service_model.service_name,
'paginators-1',
self._service_model.api_version)['pagination']
self._cache['page_config'] = page_config
except DataNotFoundError:
self._cache['page_config'] = {}
actual_operation_name = self._PY_TO_OP_NAME[operation_name]
return actual_operation_name in self._cache['page_config']
def _get_waiter_config(self):
if 'waiter_config' not in self._cache:
try:
waiter_config = self._loader.load_service_model(
self._service_model.service_name,
'waiters-2',
self._service_model.api_version)
self._cache['waiter_config'] = waiter_config
except DataNotFoundError:
self._cache['waiter_config'] = {}
return self._cache['waiter_config']
def client_api(self, section):
examples = None
try:
examples = self.get_examples(self._service_name)
except DataNotFoundError:
pass
ClientDocumenter(self._client, examples).document_client(section)
def paginator_api(self, section):
try:
service_paginator_model = self._session.get_paginator_model(
self._service_name)
except DataNotFoundError:
return
paginator_documenter = PaginatorDocumenter(
self._client, service_paginator_model)
paginator_documenter.document_paginators(section)
def list_api_versions(self, service_name, type_name):
"""List all API versions available for a particular service type
:type service_name: str
:param service_name: The name of the service
:type type_name: str
:param type_name: The type name for the service (i.e service-2,
paginators-1, etc.)
:rtype: list
:return: A list of API version strings in sorted order.
"""
known_api_versions = set()
for possible_path in self._potential_locations(service_name,
must_exist=True,
is_dir=True):
for dirname in os.listdir(possible_path):
full_path = os.path.join(possible_path, dirname, type_name)
# Only add to the known_api_versions if the directory
# contains a service-2, paginators-1, etc. file corresponding
# to the type_name passed in.
if self.file_loader.exists(full_path):
known_api_versions.add(dirname)
if not known_api_versions:
raise DataNotFoundError(data_path=service_name)
return sorted(known_api_versions)
def _find_extras(self, service_name, type_name, api_version):
"""Creates an iterator over all the extras data."""
for extras_type in self.extras_types:
extras_name = '%s.%s-extras' % (type_name, extras_type)
full_path = os.path.join(service_name, api_version, extras_name)
try:
yield self.load_data(full_path)
except DataNotFoundError:
pass
service.py 文件源码
项目:tf_aws_ecs_instance_draining_on_scale_in
作者: terraform-community-modules
项目源码
文件源码
阅读 23
收藏 0
点赞 0
评论 0
def client_api(self, section):
examples = None
try:
examples = self.get_examples(self._service_name)
except DataNotFoundError:
pass
Boto3ClientDocumenter(self._client, examples).document_client(section)
client.py 文件源码
项目:tf_aws_ecs_instance_draining_on_scale_in
作者: terraform-community-modules
项目源码
文件源码
阅读 18
收藏 0
点赞 0
评论 0
def can_paginate(self, operation_name):
"""Check if an operation can be paginated.
:type operation_name: string
:param operation_name: The operation name. This is the same name
as the method name on the client. For example, if the
method name is ``create_foo``, and you'd normally invoke the
operation as ``client.create_foo(**kwargs)``, if the
``create_foo`` operation can be paginated, you can use the
call ``client.get_paginator("create_foo")``.
:return: ``True`` if the operation can be paginated,
``False`` otherwise.
"""
if 'page_config' not in self._cache:
try:
page_config = self._loader.load_service_model(
self._service_model.service_name,
'paginators-1',
self._service_model.api_version)['pagination']
self._cache['page_config'] = page_config
except DataNotFoundError:
self._cache['page_config'] = {}
actual_operation_name = self._PY_TO_OP_NAME[operation_name]
return actual_operation_name in self._cache['page_config']
client.py 文件源码
项目:tf_aws_ecs_instance_draining_on_scale_in
作者: terraform-community-modules
项目源码
文件源码
阅读 18
收藏 0
点赞 0
评论 0
def _get_waiter_config(self):
if 'waiter_config' not in self._cache:
try:
waiter_config = self._loader.load_service_model(
self._service_model.service_name,
'waiters-2',
self._service_model.api_version)
self._cache['waiter_config'] = waiter_config
except DataNotFoundError:
self._cache['waiter_config'] = {}
return self._cache['waiter_config']
service.py 文件源码
项目:tf_aws_ecs_instance_draining_on_scale_in
作者: terraform-community-modules
项目源码
文件源码
阅读 22
收藏 0
点赞 0
评论 0
def client_api(self, section):
examples = None
try:
examples = self.get_examples(self._service_name)
except DataNotFoundError:
pass
ClientDocumenter(self._client, examples).document_client(section)
service.py 文件源码
项目:tf_aws_ecs_instance_draining_on_scale_in
作者: terraform-community-modules
项目源码
文件源码
阅读 25
收藏 0
点赞 0
评论 0
def paginator_api(self, section):
try:
service_paginator_model = self._session.get_paginator_model(
self._service_name)
except DataNotFoundError:
return
paginator_documenter = PaginatorDocumenter(
self._client, service_paginator_model)
paginator_documenter.document_paginators(section)
loaders.py 文件源码
项目:tf_aws_ecs_instance_draining_on_scale_in
作者: terraform-community-modules
项目源码
文件源码
阅读 22
收藏 0
点赞 0
评论 0
def list_api_versions(self, service_name, type_name):
"""List all API versions available for a particular service type
:type service_name: str
:param service_name: The name of the service
:type type_name: str
:param type_name: The type name for the service (i.e service-2,
paginators-1, etc.)
:rtype: list
:return: A list of API version strings in sorted order.
"""
known_api_versions = set()
for possible_path in self._potential_locations(service_name,
must_exist=True,
is_dir=True):
for dirname in os.listdir(possible_path):
full_path = os.path.join(possible_path, dirname, type_name)
# Only add to the known_api_versions if the directory
# contains a service-2, paginators-1, etc. file corresponding
# to the type_name passed in.
if self.file_loader.exists(full_path):
known_api_versions.add(dirname)
if not known_api_versions:
raise DataNotFoundError(data_path=service_name)
return sorted(known_api_versions)
loaders.py 文件源码
项目:tf_aws_ecs_instance_draining_on_scale_in
作者: terraform-community-modules
项目源码
文件源码
阅读 36
收藏 0
点赞 0
评论 0
def _find_extras(self, service_name, type_name, api_version):
"""Creates an iterator over all the extras data."""
for extras_type in self.extras_types:
extras_name = '%s.%s-extras' % (type_name, extras_type)
full_path = os.path.join(service_name, api_version, extras_name)
try:
yield self.load_data(full_path)
except DataNotFoundError:
pass