async def list_registries(
self,
*,
config_overrides: Optional[AsyncschemasClientConfig] = None,
limit: Optional["capo_schemas.types.__integer.__integer"] = None,
next_token: Optional["capo_schemas.types.__string.__string"] = None,
registry_name_prefix: Optional["capo_schemas.types.__string.__string"] = None,
scope: Optional["capo_schemas.types.__string.__string"] = None,
) -> "capo_schemas.types.list_registries_response.ListRegistriesResponse":
"""<p>List the registries.</p>
Args:
next_token: <p>The token that specifies the next page of results to return. To request the first page, leave NextToken empty. The token will expire in 24 hours, and cannot be shared with other accounts.</p>
registry_name_prefix: <p>Specifying this limits the results to only those registry names that start with the specified prefix.</p>
scope: <p>Can be set to Local or AWS to limit responses to your custom registries, or the ones provided by AWS.</p>
Raises:
capo_schemas.errors.bad_request_exception.BadRequestException
capo_schemas.errors.forbidden_exception.ForbiddenException
capo_schemas.errors.internal_server_error_exception.InternalServerErrorException
capo_schemas.errors.service_unavailable_exception.ServiceUnavailableException
capo_schemas.errors.unauthorized_exception.UnauthorizedException
capo_schemas.errors.UnknownServiceError: The service returned an error code this client does not model.
"""
async def _handler(
req: "AsyncOperationRequest[capo_schemas.types.list_registries_request.ListRegistriesRequest]",
) -> AsyncOperationResponse[
"capo_schemas.types.list_registries_response.ListRegistriesResponse"
]:
import capo_schemas._operations.schemas.list_registries
(
output,
http_response,
) = await capo_schemas._operations.schemas.list_registries.async_list_registries(
req.options, req.input
)
return AsyncOperationResponse(output=output, response=http_response)
interceptors_, options_ = self.operation_options(
config_overrides, OperationLogging("ListRegistries")
)
input_: capo_schemas.types.list_registries_request.ListRegistriesRequest = {}
if limit is not None:
input_["limit"] = limit
if next_token is not None:
input_["next_token"] = next_token
if registry_name_prefix is not None:
input_["registry_name_prefix"] = registry_name_prefix
if scope is not None:
input_["scope"] = scope
response = await aexecute_pipeline(
AsyncOperationRequest(input=input_, options=options_),
handler=_handler,
interceptors=list(interceptors_),
)
await response.response.aclose()
return response.output