def pip_execute(*args, **kwargs):
"""Overriden pip_execute() to stop sys.path being changed.
The act of importing main from the pip module seems to cause add wheels
from the /usr/share/python-wheels which are installed by various tools.
This function ensures that sys.path remains the same after the call is
executed.
"""
try:
_path = sys.path
try:
from pip import main as _pip_execute
except ImportError:
apt_update()
if six.PY2:
apt_install('python-pip')
else:
apt_install('python3-pip')
from pip import main as _pip_execute
_pip_execute(*args, **kwargs)
finally:
sys.path = _path
python类PY2的实例源码
def ns_query(address):
try:
import dns.resolver
except ImportError:
if six.PY2:
apt_install('python-dnspython', fatal=True)
else:
apt_install('python3-dnspython', fatal=True)
import dns.resolver
if isinstance(address, dns.name.Name):
rtype = 'PTR'
elif isinstance(address, six.string_types):
rtype = 'A'
else:
return None
try:
answers = dns.resolver.query(address, rtype)
except dns.resolver.NXDOMAIN:
return None
if answers:
return str(answers[0])
return None
def pip_execute(*args, **kwargs):
"""Overriden pip_execute() to stop sys.path being changed.
The act of importing main from the pip module seems to cause add wheels
from the /usr/share/python-wheels which are installed by various tools.
This function ensures that sys.path remains the same after the call is
executed.
"""
try:
_path = sys.path
try:
from pip import main as _pip_execute
except ImportError:
apt_update()
if six.PY2:
apt_install('python-pip')
else:
apt_install('python3-pip')
from pip import main as _pip_execute
_pip_execute(*args, **kwargs)
finally:
sys.path = _path
def ns_query(address):
try:
import dns.resolver
except ImportError:
if six.PY2:
apt_install('python-dnspython', fatal=True)
else:
apt_install('python3-dnspython', fatal=True)
import dns.resolver
if isinstance(address, dns.name.Name):
rtype = 'PTR'
elif isinstance(address, six.string_types):
rtype = 'A'
else:
return None
try:
answers = dns.resolver.query(address, rtype)
except dns.resolver.NXDOMAIN:
return None
if answers:
return str(answers[0])
return None
def test_keymap_new_from_buffer(self):
ctx = xkb.Context()
if six.PY2:
typecode = b'b'
else:
typecode = 'b'
test_data = array.array(typecode, sample_keymap_bytes)
km = ctx.keymap_new_from_buffer(test_data)
self.assertIsNotNone(km)
self.assertEqual(km.load_method, "buffer")
test_data.extend([0] * 10)
length = len(sample_keymap_bytes)
km = ctx.keymap_new_from_buffer(test_data, length=length)
self.assertIsNotNone(km)
# This class makes use of the details of the sample keymap in
# sample_keymap_string.
def main(args=None):
args = parse_args(args)
if not os.path.exists(args.aws_credentials):
print("%s does not exist. Please run 'aws configure' or specify an "
"alternate credentials file with --aws-credentials."
% args.aws_credentials, file=sys.stderr)
return USER_RECOVERABLE_ERROR
if PY2:
credentials = configparser.ConfigParser()
else:
credentials = configparser.ConfigParser(default_section=None)
credentials.read(args.aws_credentials)
err = one_mfa(args, credentials)
if err != OK:
return err
if args.rotate_identity_keys:
err = rotate(args, credentials)
if err != OK:
return err
if args.env:
print_env_vars(credentials, args.target_profile)
return OK
def _execute(self, idx, statement, params):
self._exec_depth += 1
try:
future = self.session.execute_async(statement, params, timeout=None)
args = (future, idx)
future.add_callbacks(
callback=self._on_success, callback_args=args,
errback=self._on_error, errback_args=args)
except Exception as exc:
# exc_info with fail_fast to preserve stack trace info when raising on the client thread
# (matches previous behavior -- not sure why we wouldn't want stack trace in the other case)
e = sys.exc_info() if self._fail_fast and six.PY2 else exc
# If we're not failing fast and all executions are raising, there is a chance of recursing
# here as subsequent requests are attempted. If we hit this threshold, schedule this result/retry
# and let the event loop thread return.
if self._exec_depth < self.max_error_recursion:
self._put_result(e, idx, False)
else:
self.session.submit(self._put_result, e, idx, False)
self._exec_depth -= 1
def apply_parameters(cls, subtypes, names=None):
"""
Given a set of other CassandraTypes, create a new subtype of this type
using them as parameters. This is how composite types are constructed.
>>> MapType.apply_parameters([DateType, BooleanType])
<class 'cassandra.cqltypes.MapType(DateType, BooleanType)'>
`subtypes` will be a sequence of CassandraTypes. If provided, `names`
will be an equally long sequence of column names or Nones.
"""
if cls.num_subtypes != 'UNKNOWN' and len(subtypes) != cls.num_subtypes:
raise ValueError("%s types require %d subtypes (%d given)"
% (cls.typename, cls.num_subtypes, len(subtypes)))
newname = cls.cass_parameterized_type_with(subtypes)
if six.PY2 and isinstance(newname, unicode):
newname = newname.encode('utf-8')
return type(newname, (cls,), {'subtypes': subtypes, 'cassname': cls.cassname, 'fieldnames': names})
def make_udt_class(cls, keyspace, udt_name, field_names, field_types):
assert len(field_names) == len(field_types)
if six.PY2 and isinstance(udt_name, unicode):
udt_name = udt_name.encode('utf-8')
instance = cls._cache.get((keyspace, udt_name))
if not instance or instance.fieldnames != field_names or instance.subtypes != field_types:
instance = type(udt_name, (cls,), {'subtypes': field_types,
'cassname': cls.cassname,
'typename': udt_name,
'fieldnames': field_names,
'keyspace': keyspace,
'mapped_class': None,
'tuple_type': cls._make_registered_udt_namedtuple(keyspace, udt_name, field_names)})
cls._cache[(keyspace, udt_name)] = instance
return instance
def pip_execute(*args, **kwargs):
"""Overriden pip_execute() to stop sys.path being changed.
The act of importing main from the pip module seems to cause add wheels
from the /usr/share/python-wheels which are installed by various tools.
This function ensures that sys.path remains the same after the call is
executed.
"""
try:
_path = sys.path
try:
from pip import main as _pip_execute
except ImportError:
apt_update()
if six.PY2:
apt_install('python-pip')
else:
apt_install('python3-pip')
from pip import main as _pip_execute
_pip_execute(*args, **kwargs)
finally:
sys.path = _path
def ns_query(address):
try:
import dns.resolver
except ImportError:
if six.PY2:
apt_install('python-dnspython', fatal=True)
else:
apt_install('python3-dnspython', fatal=True)
import dns.resolver
if isinstance(address, dns.name.Name):
rtype = 'PTR'
elif isinstance(address, six.string_types):
rtype = 'A'
else:
return None
try:
answers = dns.resolver.query(address, rtype)
except dns.resolver.NXDOMAIN:
return None
if answers:
return str(answers[0])
return None
def __init__(self, templates_dir, openstack_release):
if not os.path.isdir(templates_dir):
log('Could not locate templates dir %s' % templates_dir,
level=ERROR)
raise OSConfigException
self.templates_dir = templates_dir
self.openstack_release = openstack_release
self.templates = {}
self._tmpl_env = None
if None in [Environment, ChoiceLoader, FileSystemLoader]:
# if this code is running, the object is created pre-install hook.
# jinja2 shouldn't get touched until the module is reloaded on next
# hook execution, with proper jinja2 bits successfully imported.
if six.PY2:
apt_install('python-jinja2')
else:
apt_install('python3-jinja2')
def pip_execute(*args, **kwargs):
"""Overriden pip_execute() to stop sys.path being changed.
The act of importing main from the pip module seems to cause add wheels
from the /usr/share/python-wheels which are installed by various tools.
This function ensures that sys.path remains the same after the call is
executed.
"""
try:
_path = sys.path
try:
from pip import main as _pip_execute
except ImportError:
apt_update()
if six.PY2:
apt_install('python-pip')
else:
apt_install('python3-pip')
from pip import main as _pip_execute
_pip_execute(*args, **kwargs)
finally:
sys.path = _path
def ns_query(address):
try:
import dns.resolver
except ImportError:
if six.PY2:
apt_install('python-dnspython', fatal=True)
else:
apt_install('python3-dnspython', fatal=True)
import dns.resolver
if isinstance(address, dns.name.Name):
rtype = 'PTR'
elif isinstance(address, six.string_types):
rtype = 'A'
else:
return None
try:
answers = dns.resolver.query(address, rtype)
except dns.resolver.NXDOMAIN:
return None
if answers:
return str(answers[0])
return None
def __init__(self, templates_dir, openstack_release):
if not os.path.isdir(templates_dir):
log('Could not locate templates dir %s' % templates_dir,
level=ERROR)
raise OSConfigException
self.templates_dir = templates_dir
self.openstack_release = openstack_release
self.templates = {}
self._tmpl_env = None
if None in [Environment, ChoiceLoader, FileSystemLoader]:
# if this code is running, the object is created pre-install hook.
# jinja2 shouldn't get touched until the module is reloaded on next
# hook execution, with proper jinja2 bits successfully imported.
if six.PY2:
apt_install('python-jinja2')
else:
apt_install('python3-jinja2')
def pip_execute(*args, **kwargs):
"""Overriden pip_execute() to stop sys.path being changed.
The act of importing main from the pip module seems to cause add wheels
from the /usr/share/python-wheels which are installed by various tools.
This function ensures that sys.path remains the same after the call is
executed.
"""
try:
_path = sys.path
try:
from pip import main as _pip_execute
except ImportError:
apt_update()
if six.PY2:
apt_install('python-pip')
else:
apt_install('python3-pip')
from pip import main as _pip_execute
_pip_execute(*args, **kwargs)
finally:
sys.path = _path
def ns_query(address):
try:
import dns.resolver
except ImportError:
if six.PY2:
apt_install('python-dnspython', fatal=True)
else:
apt_install('python3-dnspython', fatal=True)
import dns.resolver
if isinstance(address, dns.name.Name):
rtype = 'PTR'
elif isinstance(address, six.string_types):
rtype = 'A'
else:
return None
try:
answers = dns.resolver.query(address, rtype)
except dns.resolver.NXDOMAIN:
return None
if answers:
return str(answers[0])
return None
def pip_execute(*args, **kwargs):
"""Overriden pip_execute() to stop sys.path being changed.
The act of importing main from the pip module seems to cause add wheels
from the /usr/share/python-wheels which are installed by various tools.
This function ensures that sys.path remains the same after the call is
executed.
"""
try:
_path = sys.path
try:
from pip import main as _pip_execute
except ImportError:
apt_update()
if six.PY2:
apt_install('python-pip')
else:
apt_install('python3-pip')
from pip import main as _pip_execute
_pip_execute(*args, **kwargs)
finally:
sys.path = _path
def ns_query(address):
try:
import dns.resolver
except ImportError:
if six.PY2:
apt_install('python-dnspython', fatal=True)
else:
apt_install('python3-dnspython', fatal=True)
import dns.resolver
if isinstance(address, dns.name.Name):
rtype = 'PTR'
elif isinstance(address, six.string_types):
rtype = 'A'
else:
return None
try:
answers = dns.resolver.query(address, rtype)
except dns.resolver.NXDOMAIN:
return None
if answers:
return str(answers[0])
return None
def __init__(self, templates_dir, openstack_release):
if not os.path.isdir(templates_dir):
log('Could not locate templates dir %s' % templates_dir,
level=ERROR)
raise OSConfigException
self.templates_dir = templates_dir
self.openstack_release = openstack_release
self.templates = {}
self._tmpl_env = None
if None in [Environment, ChoiceLoader, FileSystemLoader]:
# if this code is running, the object is created pre-install hook.
# jinja2 shouldn't get touched until the module is reloaded on next
# hook execution, with proper jinja2 bits successfully imported.
if six.PY2:
apt_install('python-jinja2')
else:
apt_install('python3-jinja2')
def pip_execute(*args, **kwargs):
"""Overriden pip_execute() to stop sys.path being changed.
The act of importing main from the pip module seems to cause add wheels
from the /usr/share/python-wheels which are installed by various tools.
This function ensures that sys.path remains the same after the call is
executed.
"""
try:
_path = sys.path
try:
from pip import main as _pip_execute
except ImportError:
apt_update()
if six.PY2:
apt_install('python-pip')
else:
apt_install('python3-pip')
from pip import main as _pip_execute
_pip_execute(*args, **kwargs)
finally:
sys.path = _path
def __init__(self, templates_dir, openstack_release):
if not os.path.isdir(templates_dir):
log('Could not locate templates dir %s' % templates_dir,
level=ERROR)
raise OSConfigException
self.templates_dir = templates_dir
self.openstack_release = openstack_release
self.templates = {}
self._tmpl_env = None
if None in [Environment, ChoiceLoader, FileSystemLoader]:
# if this code is running, the object is created pre-install hook.
# jinja2 shouldn't get touched until the module is reloaded on next
# hook execution, with proper jinja2 bits successfully imported.
if six.PY2:
apt_install('python-jinja2')
else:
apt_install('python3-jinja2')
def pip_execute(*args, **kwargs):
"""Overriden pip_execute() to stop sys.path being changed.
The act of importing main from the pip module seems to cause add wheels
from the /usr/share/python-wheels which are installed by various tools.
This function ensures that sys.path remains the same after the call is
executed.
"""
try:
_path = sys.path
try:
from pip import main as _pip_execute
except ImportError:
apt_update()
if six.PY2:
apt_install('python-pip')
else:
apt_install('python3-pip')
from pip import main as _pip_execute
_pip_execute(*args, **kwargs)
finally:
sys.path = _path
def ns_query(address):
try:
import dns.resolver
except ImportError:
if six.PY2:
apt_install('python-dnspython', fatal=True)
else:
apt_install('python3-dnspython', fatal=True)
import dns.resolver
if isinstance(address, dns.name.Name):
rtype = 'PTR'
elif isinstance(address, six.string_types):
rtype = 'A'
else:
return None
try:
answers = dns.resolver.query(address, rtype)
except dns.resolver.NXDOMAIN:
return None
if answers:
return str(answers[0])
return None
def __init__(self, templates_dir, openstack_release):
if not os.path.isdir(templates_dir):
log('Could not locate templates dir %s' % templates_dir,
level=ERROR)
raise OSConfigException
self.templates_dir = templates_dir
self.openstack_release = openstack_release
self.templates = {}
self._tmpl_env = None
if None in [Environment, ChoiceLoader, FileSystemLoader]:
# if this code is running, the object is created pre-install hook.
# jinja2 shouldn't get touched until the module is reloaded on next
# hook execution, with proper jinja2 bits successfully imported.
if six.PY2:
apt_install('python-jinja2')
else:
apt_install('python3-jinja2')
def get(self, request, container, object_name):
"""Get the object contents.
"""
obj = api.swift.swift_get_object(
request,
container,
object_name
)
# Add the original file extension back on if it wasn't preserved in the
# name given to the object.
filename = object_name.rsplit(api.swift.FOLDER_DELIMITER)[-1]
if not os.path.splitext(obj.name)[1] and obj.orig_name:
name, ext = os.path.splitext(obj.orig_name)
filename = "%s%s" % (filename, ext)
response = StreamingHttpResponse(obj.data)
safe = filename.replace(",", "")
if six.PY2:
safe = safe.encode('utf-8')
response['Content-Disposition'] = 'attachment; filename="%s"' % safe
response['Content-Type'] = 'application/octet-stream'
response['Content-Length'] = obj.bytes
return response
def object_download(request, container_name, object_path):
try:
obj = api.swift.swift_get_object(request, container_name, object_path,
resp_chunk_size=swift.CHUNK_SIZE)
except Exception:
redirect = reverse("horizon:project:containers:index")
exceptions.handle(request,
_("Unable to retrieve object."),
redirect=redirect)
# Add the original file extension back on if it wasn't preserved in the
# name given to the object.
filename = object_path.rsplit(swift.FOLDER_DELIMITER)[-1]
if not os.path.splitext(obj.name)[1] and obj.orig_name:
name, ext = os.path.splitext(obj.orig_name)
filename = "%s%s" % (filename, ext)
response = http.StreamingHttpResponse(obj.data)
safe_name = filename.replace(",", "")
if six.PY2:
safe_name = safe_name.encode('utf-8')
response['Content-Disposition'] = 'attachment; filename="%s"' % safe_name
response['Content-Type'] = 'application/octet-stream'
response['Content-Length'] = obj.bytes
return response
def _passthrough_interactive_check(self, method_name, mode):
# type: (str, str) -> bool
"""Attempt to call the specified method on the wrapped stream and return the result.
If the method is not found on the wrapped stream, returns False.
.. note::
Special Case: If wrapped stream is a Python 2 file, inspect the file mode.
:param str method_name: Name of method to call
:param str mode: Python 2 mode character
:rtype: bool
"""
try:
method = getattr(self.__wrapped, method_name)
except AttributeError:
if six.PY2 and isinstance(self.__wrapped, file): # noqa pylint: disable=undefined-variable
if mode in self.__wrapped.mode:
return True
return False
else:
return method()
def convert_into_with_meta(self, item, resp):
if isinstance(item, six.string_types):
if six.PY2 and isinstance(item, six.text_type):
return UnicodeWithMeta(item, resp)
else:
return StrWithMeta(item, resp)
elif isinstance(item, six.binary_type):
return BytesWithMeta(item, resp)
elif isinstance(item, list):
return ListWithMeta(item, resp)
elif isinstance(item, tuple):
return TupleWithMeta(item, resp)
elif item is None:
return TupleWithMeta((), resp)
else:
return DictWithMeta(item, resp)
def preprocess(self, x):
"""Load a single example using this field, tokenizing if necessary.
If the input is a Python 2 `str`, it will be converted to Unicode
first. If `sequential=True`, it will be tokenized. Then the input
will be optionally lowercased and passed to the user-provided
`preprocessing` Pipeline."""
if (six.PY2 and isinstance(x, six.string_types) and not
isinstance(x, six.text_type)):
x = Pipeline(lambda s: six.text_type(s, encoding='utf-8'))(x)
if self.sequential and isinstance(x, six.text_type):
x = self.tokenize(x.rstrip('\n'))
if self.lower:
x = Pipeline(six.text_type.lower)(x)
if self.preprocessing is not None:
return self.preprocessing(x)
else:
return x
