Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2026-03-27 02:58
📄 __init__.py # Copyright 2014 Amazon.com, Inc. or its affiliates. All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"). You # may not use this file except in compliance with the License. A copy of # the License is located at # # https://aws.amazon.com/apache2.0/ # # or in the "license" file accompanying this file. This file is # distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF # ANY KIND, either express or implied. See the License for the specific # language governing permissions and limitations under the License. import logging from boto3.compat import _warn_deprecated_python from boto3.session import Session __author__ = 'Amazon Web Services' __version__ = '1.38.23' # The default Boto3 session; autoloaded when needed. DEFAULT_SESSION = None def setup_default_session(kwargs): """ Set up a default session, passing through any parameters to the session constructor. There is no need to call this unless you wish to pass custom parameters, because a default session will be created for you. """ global DEFAULT_SESSION DEFAULT_SESSION = Session(kwargs) def set_stream_logger(name='boto3', level=logging.DEBUG, format_string=None): """ Add a stream handler for the given name and level to the logging module. By default, this logs all boto3 messages to ``stdout``. >>> import boto3 >>> boto3.set_stream_logger('boto3.resources', logging.INFO) For debugging purposes a good choice is to set the stream logger to ``''`` which is equivalent to saying "log everything". .. WARNING:: Be aware that when logging anything from ``'botocore'`` the full wire trace will appear in your logs. If your payloads contain sensitive data this should not be used in production. :type name: string :param name: Log name :type level: int :param level: Logging level, e.g. ``logging.INFO`` :type format_string: str :param format_string: Log message format """ if format_string is None: format_string = "%(asctime)s %(name)s [%(levelname)s] %(message)s" logger = logging.getLogger(name) logger.setLevel(level) handler = logging.StreamHandler() handler.setLevel(level) formatter = logging.Formatter(format_string) handler.setFormatter(formatter) logger.addHandler(handler) def _get_default_session(): """ Get the default session, creating one if needed. :rtype: :py:class:`~boto3.session.Session` :return: The default session """ if DEFAULT_SESSION is None: setup_default_session() _warn_deprecated_python() return DEFAULT_SESSION def client(args, kwargs): """ Create a low-level service client by name using the default session. See :py:meth:`boto3.session.Session.client`. """ return _get_default_session().client(args, *kwargs) def resource(args, *kwargs): """ Create a resource service client by name using the default session. See :py:meth:`boto3.session.Session.resource`. """ return _get_default_session().resource(args, **kwargs) # Set up logging to ``/dev/null`` like a library is supposed to. # https://docs.python.org/3.3/howto/logging.html#configuring-logging-for-a-library class NullHandler(logging.Handler): def emit(self, record): pass logging.getLogger('boto3').addHandler(NullHandler())	3.34 KB	2025-05-25 12:26
📁 __pycache__	-	2025-05-25 12:26
📄 compat.py # Copyright 2015 Amazon.com, Inc. or its affiliates. All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"). You # may not use this file except in compliance with the License. A copy of # the License is located at # # https://aws.amazon.com/apache2.0/ # # or in the "license" file accompanying this file. This file is # distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF # ANY KIND, either express or implied. See the License for the specific # language governing permissions and limitations under the License. import sys import os import errno import socket import warnings from boto3.exceptions import PythonDeprecationWarning # In python3, socket.error is OSError, which is too general # for what we want (i.e FileNotFoundError is a subclass of OSError). # In py3 all the socket related errors are in a newly created # ConnectionError SOCKET_ERROR = ConnectionError _APPEND_MODE_CHAR = 'a' import collections.abc as collections_abc if sys.platform.startswith('win'): def rename_file(current_filename, new_filename): try: os.remove(new_filename) except OSError as e: if not e.errno == errno.ENOENT: # We only want to a ignore trying to remove # a file that does not exist. If it fails # for any other reason we should be propagating # that exception. raise os.rename(current_filename, new_filename) else: rename_file = os.rename def filter_python_deprecation_warnings(): """ Invoking this filter acknowledges your runtime will soon be deprecated at which time you will stop receiving all updates to your client. """ warnings.filterwarnings( 'ignore', message=".Boto3 will no longer support Python.", category=PythonDeprecationWarning, module=r".*boto3\.compat" ) def _warn_deprecated_python(): """Use this template for future deprecation campaigns as needed.""" py_37_params = { 'date': 'December 13, 2023', 'blog_link': ( 'https://aws.amazon.com/blogs/developer/' 'python-support-policy-updates-for-aws-sdks-and-tools/' ) } deprecated_versions = { # Example template for future deprecations (3, 7): py_37_params, } py_version = sys.version_info[:2] if py_version in deprecated_versions: params = deprecated_versions[py_version] warning = ( "Boto3 will no longer support Python {}.{} " "starting {}. To continue receiving service updates, " "bug fixes, and security updates please upgrade to Python 3.8 or " "later. More information can be found here: {}" ).format(py_version[0], py_version[1], params['date'], params['blog_link']) warnings.warn(warning, PythonDeprecationWarning) def is_append_mode(fileobj): return ( hasattr(fileobj, 'mode') and isinstance(fileobj.mode, str) and _APPEND_MODE_CHAR in fileobj.mode )	3.01 KB	2025-05-25 12:26
📄 crt.py # Copyright 2023 Amazon.com, Inc. or its affiliates. All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"). You # may not use this file except in compliance with the License. A copy of # the License is located at # # https://aws.amazon.com/apache2.0/ # # or in the "license" file accompanying this file. This file is # distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF # ANY KIND, either express or implied. See the License for the specific # language governing permissions and limitations under the License. """ This file contains private functionality for interacting with the AWS Common Runtime library (awscrt) in boto3. All code contained within this file is for internal usage within this project and is not intended for external consumption. All interfaces contained within are subject to abrupt breaking changes. """ import threading import botocore.exceptions from botocore.session import Session from s3transfer.crt import ( BotocoreCRTCredentialsWrapper, BotocoreCRTRequestSerializer, CRTTransferManager, acquire_crt_s3_process_lock, create_s3_crt_client, ) # Singletons for CRT-backed transfers CRT_S3_CLIENT = None BOTOCORE_CRT_SERIALIZER = None CLIENT_CREATION_LOCK = threading.Lock() PROCESS_LOCK_NAME = 'boto3' def _create_crt_client(session, config, region_name, cred_provider): """Create a CRT S3 Client for file transfer. Instantiating many of these may lead to degraded performance or system resource exhaustion. """ create_crt_client_kwargs = { 'region': region_name, 'use_ssl': True, 'crt_credentials_provider': cred_provider, } return create_s3_crt_client(create_crt_client_kwargs) def _create_crt_request_serializer(session, region_name): return BotocoreCRTRequestSerializer( session, {'region_name': region_name, 'endpoint_url': None} ) def _create_crt_s3_client( session, config, region_name, credentials, lock, kwargs ): """Create boto3 wrapper class to manage crt lock reference and S3 client.""" cred_wrapper = BotocoreCRTCredentialsWrapper(credentials) cred_provider = cred_wrapper.to_crt_credentials_provider() return CRTS3Client( _create_crt_client(session, config, region_name, cred_provider), lock, region_name, cred_wrapper, ) def _initialize_crt_transfer_primatives(client, config): lock = acquire_crt_s3_process_lock(PROCESS_LOCK_NAME) if lock is None: # If we're unable to acquire the lock, we cannot # use the CRT in this process and should default to # the classic s3transfer manager. return None, None session = Session() region_name = client.meta.region_name credentials = client._get_credentials() serializer = _create_crt_request_serializer(session, region_name) s3_client = _create_crt_s3_client( session, config, region_name, credentials, lock ) return serializer, s3_client def get_crt_s3_client(client, config): global CRT_S3_CLIENT global BOTOCORE_CRT_SERIALIZER with CLIENT_CREATION_LOCK: if CRT_S3_CLIENT is None: serializer, s3_client = _initialize_crt_transfer_primatives( client, config ) BOTOCORE_CRT_SERIALIZER = serializer CRT_S3_CLIENT = s3_client return CRT_S3_CLIENT class CRTS3Client: """ This wrapper keeps track of our underlying CRT client, the lock used to acquire it and the region we've used to instantiate the client. Due to limitations in the existing CRT interfaces, we can only make calls in a single region and does not support redirects. We track the region to ensure we don't use the CRT client when a successful request cannot be made. """ def __init__(self, crt_client, process_lock, region, cred_provider): self.crt_client = crt_client self.process_lock = process_lock self.region = region self.cred_provider = cred_provider def is_crt_compatible_request(client, crt_s3_client): """ Boto3 client must use same signing region and credentials as the CRT_S3_CLIENT singleton. Otherwise fallback to classic. """ if crt_s3_client is None: return False boto3_creds = client._get_credentials() if boto3_creds is None: return False is_same_identity = compare_identity( boto3_creds.get_frozen_credentials(), crt_s3_client.cred_provider ) is_same_region = client.meta.region_name == crt_s3_client.region return is_same_region and is_same_identity def compare_identity(boto3_creds, crt_s3_creds): try: crt_creds = crt_s3_creds() except botocore.exceptions.NoCredentialsError: return False is_matching_identity = ( boto3_creds.access_key == crt_creds.access_key_id and boto3_creds.secret_key == crt_creds.secret_access_key and boto3_creds.token == crt_creds.session_token ) return is_matching_identity def create_crt_transfer_manager(client, config): """Create a CRTTransferManager for optimized data transfer.""" crt_s3_client = get_crt_s3_client(client, config) if is_crt_compatible_request(client, crt_s3_client): return CRTTransferManager( crt_s3_client.crt_client, BOTOCORE_CRT_SERIALIZER ) return None	5.28 KB	2025-05-25 12:26
📁 data	-	2025-05-25 12:26
📁 docs	-	2025-05-25 12:26
📁 dynamodb	-	2025-05-25 12:26
📁 ec2	-	2025-05-25 12:26
📁 examples	-	2025-05-25 12:26
📄 exceptions.py # Copyright 2014 Amazon.com, Inc. or its affiliates. All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"). You # may not use this file except in compliance with the License. A copy of # the License is located at # # https://aws.amazon.com/apache2.0/ # # or in the "license" file accompanying this file. This file is # distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF # ANY KIND, either express or implied. See the License for the specific # language governing permissions and limitations under the License. # All exceptions in this class should subclass from Boto3Error. import botocore.exceptions # All exceptions should subclass from Boto3Error in this module. class Boto3Error(Exception): """Base class for all Boto3 errors.""" class ResourceLoadException(Boto3Error): pass # NOTE: This doesn't appear to be used anywhere. # It's probably safe to remove this. class NoVersionFound(Boto3Error): pass # We're subclassing from botocore.exceptions.DataNotFoundError # to keep backwards compatibility with anyone that was catching # this low level Botocore error before this exception was # introduced in boto3. # Same thing for ResourceNotExistsError below. class UnknownAPIVersionError( Boto3Error, botocore.exceptions.DataNotFoundError ): def __init__(self, service_name, bad_api_version, available_api_versions): msg = ( f"The '{service_name}' resource does not support an API version of: {bad_api_version}\n" f"Valid API versions are: {available_api_versions}" ) # Not using super because we don't want the DataNotFoundError # to be called, it has a different __init__ signature. Boto3Error.__init__(self, msg) class ResourceNotExistsError( Boto3Error, botocore.exceptions.DataNotFoundError ): """Raised when you attempt to create a resource that does not exist.""" def __init__(self, service_name, available_services, has_low_level_client): msg = ( "The '{}' resource does not exist.\n" "The available resources are:\n" " - {}\n".format( service_name, '\n - '.join(available_services) ) ) if has_low_level_client: msg = ( f"{msg}\nConsider using a boto3.client('{service_name}') " f"instead of a resource for '{service_name}'" ) # Not using super because we don't want the DataNotFoundError # to be called, it has a different __init__ signature. Boto3Error.__init__(self, msg) class RetriesExceededError(Boto3Error): def __init__(self, last_exception, msg='Max Retries Exceeded'): super().__init__(msg) self.last_exception = last_exception class S3TransferFailedError(Boto3Error): pass class S3UploadFailedError(Boto3Error): pass class DynamoDBOperationNotSupportedError(Boto3Error): """Raised for operations that are not supported for an operand.""" def __init__(self, operation, value): msg = ( f'{operation} operation cannot be applied to value {value} of type ' f'{type(value)} directly. Must use AttributeBase object methods ' f'(i.e. Attr().eq()). to generate ConditionBase instances first.' ) Exception.__init__(self, msg) # FIXME: Backward compatibility DynanmoDBOperationNotSupportedError = DynamoDBOperationNotSupportedError class DynamoDBNeedsConditionError(Boto3Error): """Raised when input is not a condition""" def __init__(self, value): msg = ( f'Expecting a ConditionBase object. Got {value} of type {type(value)}. ' f'Use AttributeBase object methods (i.e. Attr().eq()). to ' f'generate ConditionBase instances.' ) Exception.__init__(self, msg) class DynamoDBNeedsKeyConditionError(Boto3Error): pass class PythonDeprecationWarning(Warning): """ Python version being used is scheduled to become unsupported in an future release. See warning for specifics. """ pass	4.03 KB	2025-05-25 12:26
📁 resources	-	2025-05-25 12:26
📁 s3	-	2025-05-25 12:26
📄 session.py # Copyright 2014 Amazon.com, Inc. or its affiliates. All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"). You # may not use this file except in compliance with the License. A copy of # the License is located at # # https://aws.amazon.com/apache2.0/ # # or in the "license" file accompanying this file. This file is # distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF # ANY KIND, either express or implied. See the License for the specific # language governing permissions and limitations under the License. import copy import os import botocore.session from botocore.client import Config from botocore.exceptions import ( DataNotFoundError, NoCredentialsError, UnknownServiceError, ) import boto3 import boto3.utils from boto3.exceptions import ResourceNotExistsError, UnknownAPIVersionError from .resources.factory import ResourceFactory class Session: """ A session stores configuration state and allows you to create service clients and resources. :type aws_access_key_id: string :param aws_access_key_id: AWS access key ID :type aws_secret_access_key: string :param aws_secret_access_key: AWS secret access key :type aws_session_token: string :param aws_session_token: AWS temporary session token :type region_name: string :param region_name: Default region when creating new connections :type botocore_session: botocore.session.Session :param botocore_session: Use this Botocore session instead of creating a new default one. :type profile_name: string :param profile_name: The name of a profile to use. If not given, then the default profile is used. :type aws_account_id: string :param aws_account_id: AWS account ID """ def __init__( self, aws_access_key_id=None, aws_secret_access_key=None, aws_session_token=None, region_name=None, botocore_session=None, profile_name=None, aws_account_id=None, ): if botocore_session is not None: self._session = botocore_session else: # Create a new default session self._session = botocore.session.get_session() # Setup custom user-agent string if it isn't already customized if self._session.user_agent_name == 'Botocore': botocore_info = f'Botocore/{self._session.user_agent_version}' if self._session.user_agent_extra: self._session.user_agent_extra += ' ' + botocore_info else: self._session.user_agent_extra = botocore_info self._session.user_agent_name = 'Boto3' self._session.user_agent_version = boto3.__version__ if profile_name is not None: self._session.set_config_variable('profile', profile_name) creds = ( aws_access_key_id, aws_secret_access_key, aws_session_token, aws_account_id, ) if any(creds): if self._account_id_set_without_credentials( aws_account_id, aws_access_key_id, aws_secret_access_key ): raise NoCredentialsError() self._session.set_credentials( aws_access_key_id, aws_secret_access_key, aws_session_token, aws_account_id, ) if region_name is not None: self._session.set_config_variable('region', region_name) self.resource_factory = ResourceFactory( self._session.get_component('event_emitter') ) self._setup_loader() self._register_default_handlers() def __repr__(self): return '{}(region_name={})'.format( self.__class__.__name__, repr(self._session.get_config_variable('region')), ) @property def profile_name(self): """ The read-only profile name. """ return self._session.profile or 'default' @property def region_name(self): """ The read-only region name. """ return self._session.get_config_variable('region') @property def events(self): """ The event emitter for a session """ return self._session.get_component('event_emitter') @property def available_profiles(self): """ The profiles available to the session credentials """ return self._session.available_profiles def _setup_loader(self): """ Setup loader paths so that we can load resources. """ self._loader = self._session.get_component('data_loader') self._loader.search_paths.append( os.path.join(os.path.dirname(__file__), 'data') ) def get_available_services(self): """ Get a list of available services that can be loaded as low-level clients via :py:meth:`Session.client`. :rtype: list :return: List of service names """ return self._session.get_available_services() def get_available_resources(self): """ Get a list of available services that can be loaded as resource clients via :py:meth:`Session.resource`. :rtype: list :return: List of service names """ return self._loader.list_available_services(type_name='resources-1') def get_available_partitions(self): """Lists the available partitions :rtype: list :return: Returns a list of partition names (e.g., ["aws", "aws-cn"]) """ return self._session.get_available_partitions() def get_available_regions( self, service_name, partition_name='aws', allow_non_regional=False ): """Lists the region and endpoint names of a particular partition. The list of regions returned by this method are regions that are explicitly known by the client to exist and is not comprehensive. A region not returned in this list may still be available for the provided service. :type service_name: string :param service_name: Name of a service to list endpoint for (e.g., s3). :type partition_name: string :param partition_name: Name of the partition to limit endpoints to. (e.g., aws for the public AWS endpoints, aws-cn for AWS China endpoints, aws-us-gov for AWS GovCloud (US) Endpoints, etc.) :type allow_non_regional: bool :param allow_non_regional: Set to True to include endpoints that are not regional endpoints (e.g., s3-external-1, fips-us-gov-west-1, etc). :return: Returns a list of endpoint names (e.g., ["us-east-1"]). """ return self._session.get_available_regions( service_name=service_name, partition_name=partition_name, allow_non_regional=allow_non_regional, ) def get_credentials(self): """ Return the :class:`botocore.credentials.Credentials` object associated with this session. If the credentials have not yet been loaded, this will attempt to load them. If they have already been loaded, this will return the cached credentials. """ return self._session.get_credentials() def get_partition_for_region(self, region_name): """Lists the partition name of a particular region. :type region_name: string :param region_name: Name of the region to list partition for (e.g., us-east-1). :rtype: string :return: Returns the respective partition name (e.g., aws). """ return self._session.get_partition_for_region(region_name) def client( self, service_name, region_name=None, api_version=None, use_ssl=True, verify=None, endpoint_url=None, aws_access_key_id=None, aws_secret_access_key=None, aws_session_token=None, config=None, aws_account_id=None, ): """ Create a low-level service client by name. :type service_name: string :param service_name: The name of a service, e.g. 's3' or 'ec2'. You can get a list of available services via :py:meth:`get_available_services`. :type region_name: string :param region_name: The name of the region associated with the client. A client is associated with a single region. :type api_version: string :param api_version: The API version to use. By default, botocore will use the latest API version when creating a client. You only need to specify this parameter if you want to use a previous API version of the client. :type use_ssl: boolean :param use_ssl: Whether or not to use SSL. By default, SSL is used. Note that not all services support non-ssl connections. :type verify: boolean/string :param verify: Whether or not to verify SSL certificates. By default SSL certificates are verified. You can provide the following values: * False - do not validate SSL certificates. SSL will still be used (unless use_ssl is False), but SSL certificates will not be verified. * path/to/cert/bundle.pem - A filename of the CA cert bundle to uses. You can specify this argument if you want to use a different CA cert bundle than the one used by botocore. :type endpoint_url: string :param endpoint_url: The complete URL to use for the constructed client. Normally, botocore will automatically construct the appropriate URL to use when communicating with a service. You can specify a complete URL (including the "http/https" scheme) to override this behavior. If this value is provided, then ``use_ssl`` is ignored. :type aws_access_key_id: string :param aws_access_key_id: The access key to use when creating the client. This is entirely optional, and if not provided, the credentials configured for the session will automatically be used. You only need to provide this argument if you want to override the credentials used for this specific client. :type aws_secret_access_key: string :param aws_secret_access_key: The secret key to use when creating the client. Same semantics as aws_access_key_id above. :type aws_session_token: string :param aws_session_token: The session token to use when creating the client. Same semantics as aws_access_key_id above. :type config: botocore.client.Config :param config: Advanced client configuration options. If region_name is specified in the client config, its value will take precedence over environment variables and configuration values, but not over a region_name value passed explicitly to the method. See `botocore config documentation <https://botocore.amazonaws.com/v1/documentation/api/latest/reference/config.html>`_ for more details. :type aws_account_id: string :param aws_account_id: The account id to use when creating the client. Same semantics as aws_access_key_id above. :return: Service client instance """ return self._session.create_client( service_name, region_name=region_name, api_version=api_version, use_ssl=use_ssl, verify=verify, endpoint_url=endpoint_url, aws_access_key_id=aws_access_key_id, aws_secret_access_key=aws_secret_access_key, aws_session_token=aws_session_token, config=config, aws_account_id=aws_account_id, ) def resource( self, service_name, region_name=None, api_version=None, use_ssl=True, verify=None, endpoint_url=None, aws_access_key_id=None, aws_secret_access_key=None, aws_session_token=None, config=None, ): """ Create a resource service client by name. :type service_name: string :param service_name: The name of a service, e.g. 's3' or 'ec2'. You can get a list of available services via :py:meth:`get_available_resources`. :type region_name: string :param region_name: The name of the region associated with the client. A client is associated with a single region. :type api_version: string :param api_version: The API version to use. By default, botocore will use the latest API version when creating a client. You only need to specify this parameter if you want to use a previous API version of the client. :type use_ssl: boolean :param use_ssl: Whether or not to use SSL. By default, SSL is used. Note that not all services support non-ssl connections. :type verify: boolean/string :param verify: Whether or not to verify SSL certificates. By default SSL certificates are verified. You can provide the following values: * False - do not validate SSL certificates. SSL will still be used (unless use_ssl is False), but SSL certificates will not be verified. * path/to/cert/bundle.pem - A filename of the CA cert bundle to uses. You can specify this argument if you want to use a different CA cert bundle than the one used by botocore. :type endpoint_url: string :param endpoint_url: The complete URL to use for the constructed client. Normally, botocore will automatically construct the appropriate URL to use when communicating with a service. You can specify a complete URL (including the "http/https" scheme) to override this behavior. If this value is provided, then ``use_ssl`` is ignored. :type aws_access_key_id: string :param aws_access_key_id: The access key to use when creating the client. This is entirely optional, and if not provided, the credentials configured for the session will automatically be used. You only need to provide this argument if you want to override the credentials used for this specific client. :type aws_secret_access_key: string :param aws_secret_access_key: The secret key to use when creating the client. Same semantics as aws_access_key_id above. :type aws_session_token: string :param aws_session_token: The session token to use when creating the client. Same semantics as aws_access_key_id above. :type config: botocore.client.Config :param config: Advanced client configuration options. If region_name is specified in the client config, its value will take precedence over environment variables and configuration values, but not over a region_name value passed explicitly to the method. If user_agent_extra is specified in the client config, it overrides the default user_agent_extra provided by the resource API. See `botocore config documentation <https://botocore.amazonaws.com/v1/documentation/api/latest/reference/config.html>`_ for more details. :return: Subclass of :py:class:`~boto3.resources.base.ServiceResource` """ try: resource_model = self._loader.load_service_model( service_name, 'resources-1', api_version ) except UnknownServiceError: available = self.get_available_resources() has_low_level_client = ( service_name in self.get_available_services() ) raise ResourceNotExistsError( service_name, available, has_low_level_client ) except DataNotFoundError: # This is because we've provided an invalid API version. available_api_versions = self._loader.list_api_versions( service_name, 'resources-1' ) raise UnknownAPIVersionError( service_name, api_version, ', '.join(available_api_versions) ) if api_version is None: # Even though botocore's load_service_model() can handle # using the latest api_version if not provided, we need # to track this api_version in boto3 in order to ensure # we're pairing a resource model with a client model # of the same API version. It's possible for the latest # API version of a resource model in boto3 to not be # the same API version as a service model in botocore. # So we need to look up the api_version if one is not # provided to ensure we load the same API version of the # client. # # Note: This is relying on the fact that # loader.load_service_model(..., api_version=None) # and loader.determine_latest_version(..., 'resources-1') # both load the same api version of the file. api_version = self._loader.determine_latest_version( service_name, 'resources-1' ) # Creating a new resource instance requires the low-level client # and service model, the resource version and resource JSON data. # We pass these to the factory and get back a class, which is # instantiated on top of the low-level client. if config is not None: if config.user_agent_extra is None: config = copy.deepcopy(config) config.user_agent_extra = 'Resource' else: config = Config(user_agent_extra='Resource') client = self.client( service_name, region_name=region_name, api_version=api_version, use_ssl=use_ssl, verify=verify, endpoint_url=endpoint_url, aws_access_key_id=aws_access_key_id, aws_secret_access_key=aws_secret_access_key, aws_session_token=aws_session_token, config=config, ) service_model = client.meta.service_model # Create a ServiceContext object to serve as a reference to # important read-only information about the general service. service_context = boto3.utils.ServiceContext( service_name=service_name, service_model=service_model, resource_json_definitions=resource_model['resources'], service_waiter_model=boto3.utils.LazyLoadedWaiterModel( self._session, service_name, api_version ), ) # Create the service resource class. cls = self.resource_factory.load_from_definition( resource_name=service_name, single_resource_json_definition=resource_model['service'], service_context=service_context, ) return cls(client=client) def _register_default_handlers(self): # S3 customizations self._session.register( 'creating-client-class.s3', boto3.utils.lazy_call( 'boto3.s3.inject.inject_s3_transfer_methods' ), ) self._session.register( 'creating-resource-class.s3.Bucket', boto3.utils.lazy_call('boto3.s3.inject.inject_bucket_methods'), ) self._session.register( 'creating-resource-class.s3.Object', boto3.utils.lazy_call('boto3.s3.inject.inject_object_methods'), ) self._session.register( 'creating-resource-class.s3.ObjectSummary', boto3.utils.lazy_call( 'boto3.s3.inject.inject_object_summary_methods' ), ) # DynamoDb customizations self._session.register( 'creating-resource-class.dynamodb', boto3.utils.lazy_call( 'boto3.dynamodb.transform.register_high_level_interface' ), unique_id='high-level-dynamodb', ) self._session.register( 'creating-resource-class.dynamodb.Table', boto3.utils.lazy_call( 'boto3.dynamodb.table.register_table_methods' ), unique_id='high-level-dynamodb-table', ) # EC2 Customizations self._session.register( 'creating-resource-class.ec2.ServiceResource', boto3.utils.lazy_call('boto3.ec2.createtags.inject_create_tags'), ) self._session.register( 'creating-resource-class.ec2.Instance', boto3.utils.lazy_call( 'boto3.ec2.deletetags.inject_delete_tags', event_emitter=self.events, ), ) def _account_id_set_without_credentials( self, account_id, access_key, secret_key ): if account_id is None: return False elif access_key is None or secret_key is None: return True return False	21.21 KB	2025-05-25 12:26
📄 utils.py # Copyright 2015 Amazon.com, Inc. or its affiliates. All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"). You # may not use this file except in compliance with the License. A copy of # the License is located at # # https://aws.amazon.com/apache2.0/ # # or in the "license" file accompanying this file. This file is # distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF # ANY KIND, either express or implied. See the License for the specific # language governing permissions and limitations under the License. import sys from collections import namedtuple _ServiceContext = namedtuple( 'ServiceContext', [ 'service_name', 'service_model', 'service_waiter_model', 'resource_json_definitions', ], ) class ServiceContext(_ServiceContext): """Provides important service-wide, read-only information about a service :type service_name: str :param service_name: The name of the service :type service_model: :py:class:`botocore.model.ServiceModel` :param service_model: The model of the service. :type service_waiter_model: :py:class:`botocore.waiter.WaiterModel` or a waiter model-like object such as :py:class:`boto3.utils.LazyLoadedWaiterModel` :param service_waiter_model: The waiter model of the service. :type resource_json_definitions: dict :param resource_json_definitions: The loaded json models of all resource shapes for a service. It is equivalient of loading a ``resource-1.json`` and retrieving the value at the key "resources". """ pass def import_module(name): """Import module given a name. Does not support relative imports. """ __import__(name) return sys.modules[name] def lazy_call(full_name, kwargs): parent_kwargs = kwargs def _handler(kwargs): module, function_name = full_name.rsplit('.', 1) module = import_module(module) kwargs.update(parent_kwargs) return getattr(module, function_name)(**kwargs) return _handler def inject_attribute(class_attributes, name, value): if name in class_attributes: raise RuntimeError( f'Cannot inject class attribute "{name}", attribute ' f'already exists in class dict.' ) else: class_attributes[name] = value class LazyLoadedWaiterModel: """A lazily loaded waiter model This does not load the service waiter model until an attempt is made to retrieve the waiter model for a specific waiter. This is helpful in docstring generation where we do not need to actually need to grab the waiter-2.json until it is accessed through a ``get_waiter`` call when the docstring is generated/accessed. """ def __init__(self, bc_session, service_name, api_version): self._session = bc_session self._service_name = service_name self._api_version = api_version def get_waiter(self, waiter_name): return self._session.get_waiter_model( self._service_name, self._api_version ).get_waiter(waiter_name)	3.07 KB	2025-05-25 12:26

Terminal