search for _sa_instrument_role-decorated methods in method resolution order, assign to roles.
(cls)
| 845 | |
| 846 | |
| 847 | def _locate_roles_and_methods(cls): |
| 848 | """search for _sa_instrument_role-decorated methods in |
| 849 | method resolution order, assign to roles. |
| 850 | |
| 851 | """ |
| 852 | |
| 853 | roles: Dict[str, str] = {} |
| 854 | methods: Dict[str, Tuple[Optional[str], Optional[int], Optional[str]]] = {} |
| 855 | |
| 856 | for supercls in cls.__mro__: |
| 857 | for name, method in vars(supercls).items(): |
| 858 | if not callable(method): |
| 859 | continue |
| 860 | |
| 861 | # note role declarations |
| 862 | if hasattr(method, "_sa_instrument_role"): |
| 863 | role = method._sa_instrument_role |
| 864 | assert role in ("appender", "remover", "iterator") |
| 865 | roles.setdefault(role, name) |
| 866 | |
| 867 | # transfer instrumentation requests from decorated function |
| 868 | # to the combined queue |
| 869 | before: Optional[Tuple[str, int]] = None |
| 870 | after: Optional[str] = None |
| 871 | |
| 872 | if hasattr(method, "_sa_instrument_before"): |
| 873 | op, argument = method._sa_instrument_before |
| 874 | assert op in ("fire_append_event", "fire_remove_event") |
| 875 | before = op, argument |
| 876 | if hasattr(method, "_sa_instrument_after"): |
| 877 | op = method._sa_instrument_after |
| 878 | assert op in ("fire_append_event", "fire_remove_event") |
| 879 | after = op |
| 880 | if before: |
| 881 | methods[name] = before + (after,) |
| 882 | elif after: |
| 883 | methods[name] = None, None, after |
| 884 | return roles, methods |
| 885 | |
| 886 | |
| 887 | def _setup_canned_roles(cls, roles, methods): |
no test coverage detected