Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def bind_to_model(self, model_cls: Type[Model]) -> ModelObserver:
observer = ModelObserver(func=self.func, model_cls=model_cls)
observer.groups(self._group_names)
observer.serializer(self._serializer)
return observer
def groups(self, func):
self._group_names = func
return self
def serializer(self, func):
self._serializer = func
return self
class ObserverAPIConsumerMetaclass(APIConsumerMetaclass):
def __new__(mcs, name, bases, body) -> Type[GenericAsyncAPIConsumer]:
queryset = body.get("queryset", None)
if queryset is not None:
for attr_name, attr in body.items():
if isinstance(attr, _GenericModelObserver):
body[attr_name] = attr.bind_to_model(model_cls=queryset.model)
for base in bases:
for attr_name in dir(base):
attr = getattr(base, attr_name)
if isinstance(attr, _GenericModelObserver):
body[attr_name] = attr.bind_to_model(model_cls=queryset.model)
return super().__new__(mcs, name, bases, body)