Coverage for mongo/base.py: 100%

1from flask import current_app 

2from . import engine 

3from mongoengine.errors import * 

4import logging 

5 

6__all__ = ['MongoBase'] 

7 

8 

9class MongoBase: 

10 qs_filter = {} 

11 

12 def __init_subclass__(cls, engine, **kwargs): 

13 super().__init_subclass__(**kwargs) 

14 cls.engine = engine 

15 

16 def __new__(cls, pk, *args, **kwargs): 

17 if isinstance(pk, cls): 

18 return pk 

19 new = super().__new__(cls) 

20 # got a engine instance 

21 if isinstance(pk, new.engine): 

22 new.obj = pk 

23 else: 

24 try: 

25 new.obj = new.engine.objects(pk=pk).get() 

26 except engine.DoesNotExist: 

27 new.obj = new.engine(id=pk) 

28 return new 

29 

30 def __getattr__(self, name): 

31 return self.obj.__getattribute__(name) 

32 

33 def __setattr__(self, name, value): 

34 if name in self.engine._fields: 

35 self.obj.__setattr__(name, value) 

36 else: 

37 super().__setattr__(name, value) 

38 

39 def __eq__(self, other): 

40 return self and other is not None and self.pk == other.pk 

41 

42 def __bool__(self): 

43 try: 

44 return self._qs.filter(pk=self.pk, **self.qs_filter).__bool__() 

45 except ValidationError: 

46 return False 

47 

48 def __str__(self): 

49 return f'{self.__class__.__name__.lower()} [{self.pk}]' 

50 

51 def __repr__(self): 

52 return self.obj.to_json() if self else '{}' 

53 

54 def reload(self, *fields): 

55 if self: 

56 self.obj.reload(*fields) 

57 return self 

58 

59 @property 

60 def logger(self): 

61 try: 

62 return current_app.logger 

63 except RuntimeError: 

64 return logging.getLogger('gunicorn.error')