def authenticate(**credentials):
"""
If the given credentials are valid, return a User object.
"""
for backend, backend_path in _get_backends(return_tuples=True):
try:
inspect.getcallargs(backend.authenticate, **credentials)
except TypeError:
# This backend doesn't accept these credentials as arguments. Try the next one.
continue
try:
user = backend.authenticate(**credentials)
except PermissionDenied:
# This backend says to stop in our tracks - this user should not be allowed in at all.
return None
if user is None:
continue
# Annotate the user object with the path of the backend.
user.backend = backend_path
return user
# The credentials supplied are invalid to all backends, fire signal
user_login_failed.send(sender=__name__,
credentials=_clean_credentials(credentials))
python类getcallargs()的实例源码
def __new__(cls, name, bases, attrs):
get_form = attrs.get('get_form')
if get_form and inspect.isfunction(get_form):
try:
inspect.getcallargs(get_form, None)
except TypeError:
warnings.warn(
"`%s.%s.get_form` method must define a default value for "
"its `form_class` argument." % (attrs['__module__'], name),
RemovedInDjango110Warning, stacklevel=2
)
def get_form_with_form_class(self, form_class=None):
if form_class is None:
form_class = self.get_form_class()
return get_form(self, form_class=form_class)
attrs['get_form'] = get_form_with_form_class
return super(FormMixinBase, cls).__new__(cls, name, bases, attrs)
def authenticate(**credentials):
"""
If the given credentials are valid, return a User object.
"""
for backend, backend_path in _get_backends(return_tuples=True):
try:
inspect.getcallargs(backend.authenticate, **credentials)
except TypeError:
# This backend doesn't accept these credentials as arguments. Try the next one.
continue
try:
user = backend.authenticate(**credentials)
except PermissionDenied:
# This backend says to stop in our tracks - this user should not be allowed in at all.
break
if user is None:
continue
# Annotate the user object with the path of the backend.
user.backend = backend_path
return user
# The credentials supplied are invalid to all backends, fire signal
user_login_failed.send(sender=__name__, credentials=_clean_credentials(credentials))
def wrap_async(f, _inner_f, check_type_args, check_type_annotations):
@wraps(f)
async def _f(*args, **kwargs):
call_args = inspect.getcallargs(_inner_f, *args, **kwargs)
for k, v in list(call_args.items()):
if k in check_type_annotations:
call_args[k] = check_type(v, check_type_annotations[k])
# Create arguments
args = [call_args.pop(a) for a in check_type_args.args]
if check_type_args.varargs is not None:
args.extend(call_args.pop(check_type_args.varargs))
if check_type_args.varkw is not None:
kwargs = call_args.pop(check_type_args.varkw)
else:
kwargs = {}
kwargs.update(call_args)
_return = await f(*args, **kwargs)
if 'return' in check_type_annotations:
return check_type(_return, check_type_annotations['return'])
else:
return _return
return _f
def get_cache_key(self, *args, **kwargs):
callargs = inspect.getcallargs(self.fn, *args, **kwargs)
values = []
if isfunction(self.vary_on):
values = self.vary_on(*args, **kwargs)
else:
for arg_name, attrs in self.vary_on:
value = callargs[arg_name]
for attr in attrs:
value = getattr(value, attr)
values.append(value)
args_string = ','.join(self._serialize_for_key(value)
for value in values)
if len(args_string) > 150:
args_string = 'H' + self._hash(args_string)
return 'quickcache.{}/{}'.format(self.prefix, args_string)
def api_exc_patch(func):
if isinstance(func, FunctionType):
@wraps(func)
def deco(*args, **kwargs):
try:
return func(*args, **kwargs)
except RQInvalidArgument:
raise
except Exception as e:
if isinstance(e, TypeError):
exc_info = sys.exc_info()
try:
ret = inspect.getcallargs(unwrapper(func), *args, **kwargs)
except TypeError:
t, v, tb = exc_info
raise patch_user_exc(v.with_traceback(tb))
if getattr(e, EXC_EXT_NAME, EXC_TYPE.NOTSET) == EXC_TYPE.NOTSET:
patch_system_exc(e)
raise
return deco
return func
def api_exc_patch(func):
if isinstance(func, FunctionType):
@wraps(func)
def deco(*args, **kwargs):
try:
return func(*args, **kwargs)
except RQInvalidArgument:
raise
except Exception as e:
if isinstance(e, TypeError):
exc_info = sys.exc_info()
try:
ret = inspect.getcallargs(unwrapper(func), *args, **kwargs)
except TypeError:
t, v, tb = exc_info
raise patch_user_exc(v.with_traceback(tb))
if getattr(e, EXC_EXT_NAME, EXC_TYPE.NOTSET) == EXC_TYPE.NOTSET:
patch_system_exc(e)
raise
return deco
return func
def call(self, function, args=None, kw=None):
if args is None:
args = ()
if kw is None:
kw = {}
arguments = inspect.getcallargs(function, *args, **kw)
key = self.key_maker.key(function, arguments)
try:
value = self.cache.get(key)
except KeyError:
value = function(*args, **kw)
self.cache.set(key, value)
if self.return_copies:
value = copy.deepcopy(value)
return value
def authenticate(**credentials):
"""
If the given credentials are valid, return a User object.
"""
for backend, backend_path in _get_backends(return_tuples=True):
try:
inspect.getcallargs(backend.authenticate, **credentials)
except TypeError:
# This backend doesn't accept these credentials as arguments. Try the next one.
continue
try:
user = backend.authenticate(**credentials)
except PermissionDenied:
# This backend says to stop in our tracks - this user should not be allowed in at all.
break
if user is None:
continue
# Annotate the user object with the path of the backend.
user.backend = backend_path
return user
# The credentials supplied are invalid to all backends, fire signal
user_login_failed.send(sender=__name__, credentials=_clean_credentials(credentials))
def Resource(resource_name):
def real_dec(func):
@functools.wraps(func)
def fetch_resource(*args, **kwargs):
called_args = inspect.getcallargs(func, *args, **kwargs)
# We do not need `self` for building the params
self = called_args.pop('self')
url = self._BuildUrl('http://stats.nba.com/stats/',
resource_name,
called_args)
resp = self._FetchUrl(url)
resp_dict = resp.json()
if self._transform_json:
resp_dict = Api._TransformResponseDict(resp_dict)
return resp_dict
return fetch_resource
return real_dec
# noinspection PyPep8Naming
def _is_valid(cls, *args, **kwargs):
# fill our args
output_dir = args[0]
# boxlib datasets are always directories
if not os.path.isdir(output_dir): return False
header_filename = os.path.join(output_dir, "Header")
jobinfo_filename = os.path.join(output_dir, "job_info")
if not os.path.exists(header_filename):
# We *know* it's not boxlib if Header doesn't exist.
return False
args = inspect.getcallargs(cls.__init__, args, kwargs)
# This might need to be localized somehow
if args['cparam_filename'] is None:
return True # Treat as generic boxlib data
inputs_filename = os.path.join(
os.path.dirname(os.path.abspath(output_dir)),
args['cparam_filename'])
if not os.path.exists(inputs_filename) and \
not os.path.exists(jobinfo_filename):
return True # We have no parameters to go off of
# If we do have either inputs or jobinfo, we should be deferring to a
# different frontend.
return False
def authenticate(**credentials):
"""
If the given credentials are valid, return a User object.
"""
for backend, backend_path in _get_backends(return_tuples=True):
try:
inspect.getcallargs(backend.authenticate, **credentials)
except TypeError:
# This backend doesn't accept these credentials as arguments. Try the next one.
continue
try:
user = backend.authenticate(**credentials)
except PermissionDenied:
# This backend says to stop in our tracks - this user should not be allowed in at all.
return None
if user is None:
continue
# Annotate the user object with the path of the backend.
user.backend = backend_path
return user
# The credentials supplied are invalid to all backends, fire signal
user_login_failed.send(sender=__name__,
credentials=_clean_credentials(credentials))
def assertEqualException(self, func, call_param_string, locs=None):
locs = dict(locs or {}, func=func)
try:
eval('func(%s)' % call_param_string, None, locs)
except Exception, ex1:
pass
else:
self.fail('Exception not raised')
try:
eval('inspect.getcallargs(func, %s)' % call_param_string, None,
locs)
except Exception, ex2:
pass
else:
self.fail('Exception not raised')
self.assertIs(type(ex1), type(ex2))
self.assertEqual(str(ex1), str(ex2))
def assertEqualException(self, func, call_param_string, locs=None):
locs = dict(locs or {}, func=func)
try:
eval('func(%s)' % call_param_string, None, locs)
except Exception, ex1:
pass
else:
self.fail('Exception not raised')
try:
eval('inspect.getcallargs(func, %s)' % call_param_string, None,
locs)
except Exception, ex2:
pass
else:
self.fail('Exception not raised')
self.assertIs(type(ex1), type(ex2))
self.assertEqual(str(ex1), str(ex2))
def authenticate(**credentials):
"""
If the given credentials are valid, return a User object.
"""
for backend, backend_path in _get_backends(return_tuples=True):
try:
inspect.getcallargs(backend.authenticate, **credentials)
except TypeError:
# This backend doesn't accept these credentials as arguments. Try the next one.
continue
try:
user = backend.authenticate(**credentials)
except PermissionDenied:
# This backend says to stop in our tracks - this user should not be allowed in at all.
return None
if user is None:
continue
# Annotate the user object with the path of the backend.
user.backend = backend_path
return user
# The credentials supplied are invalid to all backends, fire signal
user_login_failed.send(sender=__name__,
credentials=_clean_credentials(credentials))
def authenticate(**credentials):
"""
If the given credentials are valid, return a User object.
"""
for backend, backend_path in _get_backends(return_tuples=True):
try:
inspect.getcallargs(backend.authenticate, **credentials)
except TypeError:
# This backend doesn't accept these credentials as arguments. Try the next one.
continue
try:
user = backend.authenticate(**credentials)
except PermissionDenied:
# This backend says to stop in our tracks - this user should not be allowed in at all.
return None
if user is None:
continue
# Annotate the user object with the path of the backend.
user.backend = backend_path
return user
# The credentials supplied are invalid to all backends, fire signal
user_login_failed.send(sender=__name__,
credentials=_clean_credentials(credentials))
def authenticate(**credentials):
"""
If the given credentials are valid, return a User object.
"""
for backend, backend_path in _get_backends(return_tuples=True):
try:
inspect.getcallargs(backend.authenticate, **credentials)
except TypeError:
# This backend doesn't accept these credentials as arguments. Try the next one.
continue
try:
user = backend.authenticate(**credentials)
except PermissionDenied:
# This backend says to stop in our tracks - this user should not be allowed in at all.
return None
if user is None:
continue
# Annotate the user object with the path of the backend.
user.backend = backend_path
return user
# The credentials supplied are invalid to all backends, fire signal
user_login_failed.send(sender=__name__,
credentials=_clean_credentials(credentials))
def authenticate(**credentials):
"""
If the given credentials are valid, return a User object.
"""
for backend, backend_path in _get_backends(return_tuples=True):
try:
inspect.getcallargs(backend.authenticate, **credentials)
except TypeError:
# This backend doesn't accept these credentials as arguments. Try the next one.
continue
try:
user = backend.authenticate(**credentials)
except PermissionDenied:
# This backend says to stop in our tracks - this user should not be allowed in at all.
return None
if user is None:
continue
# Annotate the user object with the path of the backend.
user.backend = backend_path
return user
# The credentials supplied are invalid to all backends, fire signal
user_login_failed.send(sender=__name__,
credentials=_clean_credentials(credentials))
def assertEqualException(self, func, call_param_string, locs=None):
locs = dict(locs or {}, func=func)
try:
eval('func(%s)' % call_param_string, None, locs)
except Exception as e:
ex1 = e
else:
self.fail('Exception not raised')
try:
eval('inspect.getcallargs(func, %s)' % call_param_string, None,
locs)
except Exception as e:
ex2 = e
else:
self.fail('Exception not raised')
self.assertIs(type(ex1), type(ex2))
self.assertEqual(str(ex1), str(ex2))
del ex1, ex2
def authenticate(**credentials):
"""
If the given credentials are valid, return a User object.
"""
for backend, backend_path in _get_backends(return_tuples=True):
try:
inspect.getcallargs(backend.authenticate, **credentials)
except TypeError:
# This backend doesn't accept these credentials as arguments. Try the next one.
continue
try:
user = backend.authenticate(**credentials)
except PermissionDenied:
# This backend says to stop in our tracks - this user should not be allowed in at all.
break
if user is None:
continue
# Annotate the user object with the path of the backend.
user.backend = backend_path
return user
# The credentials supplied are invalid to all backends, fire signal
user_login_failed.send(sender=__name__, credentials=_clean_credentials(credentials))
def check_required_args(parameters):
"""check parameters of action"""
def decorated(f):
"""decorator"""
@functools.wraps(f)
def wrapper(*args, **kwargs):
"""wrapper"""
func_args = inspect.getcallargs(f, *args, **kwargs)
kwargs = func_args.get('kwargs')
for item in parameters:
if kwargs.get(item) is None:
message = "check required args failed, `{0}` is not found in {1}".format(item, f.__name__)
LOG.error(message)
raise Exception(message)
return f(*args, **kwargs)
return wrapper
return decorated
def authenticate(**credentials):
"""
If the given credentials are valid, return a User object.
"""
for backend, backend_path in _get_backends(return_tuples=True):
try:
inspect.getcallargs(backend.authenticate, **credentials)
except TypeError:
# This backend doesn't accept these credentials as arguments. Try the next one.
continue
try:
user = backend.authenticate(**credentials)
except PermissionDenied:
# This backend says to stop in our tracks - this user should not be allowed in at all.
break
if user is None:
continue
# Annotate the user object with the path of the backend.
user.backend = backend_path
return user
# The credentials supplied are invalid to all backends, fire signal
user_login_failed.send(sender=__name__, credentials=_clean_credentials(credentials))
def dispatch_packet(self, packet):
if not getattr(packet, 'no_log', False):
self.log.debug('received %r', packet)
packet_type = packet.type
method = self._packet_handlers.get(packet_type, None)
if method is None:
self.on_missing_handler(packet)
return None
arg_spec = inspect.getargspec(method)
args, kwargs = packet.get_method_args(len(arg_spec[0]))
try:
inspect.getcallargs(method, packet_type, *args, **kwargs)
except TypeError as e:
raise PacketFormatError(text_type(e))
try:
ret = method(packet_type, *args, **kwargs)
except Exception:
raise
else:
return ret
def authenticate(**credentials):
"""
If the given credentials are valid, return a User object.
"""
for backend, backend_path in _get_backends(return_tuples=True):
try:
inspect.getcallargs(backend.authenticate, **credentials)
except TypeError:
# This backend doesn't accept these credentials as arguments. Try the next one.
continue
try:
user = backend.authenticate(**credentials)
except PermissionDenied:
# This backend says to stop in our tracks - this user should not be allowed in at all.
break
if user is None:
continue
# Annotate the user object with the path of the backend.
user.backend = backend_path
return user
# The credentials supplied are invalid to all backends, fire signal
user_login_failed.send(sender=__name__, credentials=_clean_credentials(credentials))
def is_closable_iterator(obj):
# Not an iterator.
if not is_iterator(obj):
return False
# A generator - the easiest thing to deal with.
import inspect
if inspect.isgenerator(obj):
return True
# A custom iterator. Look for a close method...
if not (hasattr(obj, 'close') and callable(obj.close)):
return False
# ... which doesn't require any arguments.
try:
inspect.getcallargs(obj.close)
except TypeError:
return False
else:
return True
def assertEqualException(self, func, call_param_string, locs=None):
locs = dict(locs or {}, func=func)
try:
eval('func(%s)' % call_param_string, None, locs)
except Exception, ex1:
pass
else:
self.fail('Exception not raised')
try:
eval('inspect.getcallargs(func, %s)' % call_param_string, None,
locs)
except Exception, ex2:
pass
else:
self.fail('Exception not raised')
self.assertIs(type(ex1), type(ex2))
if check_impl_detail():
self.assertEqual(str(ex1), str(ex2))
def assertEqualException(self, func, call_param_string, locs=None):
locs = dict(locs or {}, func=func)
try:
eval('func(%s)' % call_param_string, None, locs)
except Exception as e:
ex1 = e
else:
self.fail('Exception not raised')
try:
eval('inspect.getcallargs(func, %s)' % call_param_string, None,
locs)
except Exception as e:
ex2 = e
else:
self.fail('Exception not raised')
self.assertIs(type(ex1), type(ex2))
self.assertEqual(str(ex1), str(ex2))
del ex1, ex2
def assertEqualException(self, func, call_param_string, locs=None):
locs = dict(locs or {}, func=func)
try:
eval('func(%s)' % call_param_string, None, locs)
except Exception, ex1:
pass
else:
self.fail('Exception not raised')
try:
eval('inspect.getcallargs(func, %s)' % call_param_string, None,
locs)
except Exception, ex2:
pass
else:
self.fail('Exception not raised')
self.assertIs(type(ex1), type(ex2))
self.assertEqual(str(ex1), str(ex2))
def authenticate(**credentials):
"""
If the given credentials are valid, return a User object.
"""
for backend, backend_path in _get_backends(return_tuples=True):
try:
inspect.getcallargs(backend.authenticate, **credentials)
except TypeError:
# This backend doesn't accept these credentials as arguments. Try the next one.
continue
try:
user = backend.authenticate(**credentials)
except PermissionDenied:
# This backend says to stop in our tracks - this user should not be allowed in at all.
return None
if user is None:
continue
# Annotate the user object with the path of the backend.
user.backend = backend_path
return user
# The credentials supplied are invalid to all backends, fire signal
user_login_failed.send(sender=__name__,
credentials=_clean_credentials(credentials))
def authenticate(**credentials):
"""
If the given credentials are valid, return a User object.
"""
for backend, backend_path in _get_backends(return_tuples=True):
try:
inspect.getcallargs(backend.authenticate, **credentials)
except TypeError:
# This backend doesn't accept these credentials as arguments. Try the next one.
continue
try:
user = backend.authenticate(**credentials)
except PermissionDenied:
# This backend says to stop in our tracks - this user should not be allowed in at all.
return None
if user is None:
continue
# Annotate the user object with the path of the backend.
user.backend = backend_path
return user
# The credentials supplied are invalid to all backends, fire signal
user_login_failed.send(sender=__name__,
credentials=_clean_credentials(credentials))