Route method args and/or return value through the collection adapter.
(method, before, argument, after)
| 964 | |
| 965 | |
| 966 | def _instrument_membership_mutator(method, before, argument, after): |
| 967 | """Route method args and/or return value through the collection |
| 968 | adapter.""" |
| 969 | # This isn't smart enough to handle @adds(1) for 'def fn(self, (a, b))' |
| 970 | if before: |
| 971 | fn_args = list( |
| 972 | util.flatten_iterator(inspect_getfullargspec(method)[0]) |
| 973 | ) |
| 974 | if isinstance(argument, int): |
| 975 | pos_arg = argument |
| 976 | named_arg = len(fn_args) > argument and fn_args[argument] or None |
| 977 | else: |
| 978 | if argument in fn_args: |
| 979 | pos_arg = fn_args.index(argument) |
| 980 | else: |
| 981 | pos_arg = None |
| 982 | named_arg = argument |
| 983 | del fn_args |
| 984 | |
| 985 | def wrapper(*args, **kw): |
| 986 | if before: |
| 987 | if pos_arg is None: |
| 988 | if named_arg not in kw: |
| 989 | raise sa_exc.ArgumentError( |
| 990 | "Missing argument %s" % argument |
| 991 | ) |
| 992 | value = kw[named_arg] |
| 993 | else: |
| 994 | if len(args) > pos_arg: |
| 995 | value = args[pos_arg] |
| 996 | elif named_arg in kw: |
| 997 | value = kw[named_arg] |
| 998 | else: |
| 999 | raise sa_exc.ArgumentError( |
| 1000 | "Missing argument %s" % argument |
| 1001 | ) |
| 1002 | |
| 1003 | initiator = kw.pop("_sa_initiator", None) |
| 1004 | if initiator is False: |
| 1005 | executor = None |
| 1006 | else: |
| 1007 | executor = args[0]._sa_adapter |
| 1008 | |
| 1009 | if before and executor: |
| 1010 | getattr(executor, before)(value, initiator) |
| 1011 | |
| 1012 | if not after or not executor: |
| 1013 | return method(*args, **kw) |
| 1014 | else: |
| 1015 | res = method(*args, **kw) |
| 1016 | if res is not None: |
| 1017 | getattr(executor, after)(res, initiator) |
| 1018 | return res |
| 1019 | |
| 1020 | wrapper._sa_instrumented = True # type: ignore[attr-defined] |
| 1021 | if hasattr(method, "_sa_instrument_role"): |
| 1022 | wrapper._sa_instrument_role = method._sa_instrument_role # type: ignore[attr-defined] # noqa: E501 |
| 1023 | wrapper.__name__ = method.__name__ |
no test coverage detected