lico.core.contrib.models.py

  2. from typing import Callable, Dict, Iterable, Optional
  4. from django.db.models import Model as BaseModel
  5. from django.db.models.manager import BaseManager
  6. from django.db.models.query import QuerySet as BaseQuerySet
  9. class ToDictMixin:
  10.     as_dict_exclude = ()
  12.     """
  13.         # workspace\lenovo_git\lico-core\lico-core-license\lico\core\license\views.py
  15.         license = License.objects.get()
  16.         result = license.as_dict(
  17.             include=[
  18.                 'license_key', 'customer_name',
  19.                 'cluster_type', 'license_code', 'expire_time',
  20.                 'license_numbers', 'status'
  21.             ]
  22.         )
  24.         result: {
  25.             "license_key": "b\"l\\xaa\\x",
  26.             "cluster_type": "host",
  27.             "license_code": "Lenovo HPC AI LiCO Software",
  28.             "expire_time": "2032-06-30",
  29.             "license_numbers": 64,
  30.             "status": "valid",
  31.             "rest_days": 4146
  32.         }
  35.         >>> from lico.core.accounting.models import BillGroup, Deposit
  36.         >>> query = Deposit.objects.all()
  37.         >>> query[0].as_dict()
  38.         {'id': 1, 'user': 'test', 'credits': -240.0, 'apply_time': 1615432804, 'approved_time': 1615432804, 'billing_type': '', 'billing_id': None, 'balance': -107.0, 'bill_group': {'id': 1, 'name': 'test', 'balance': 500.0, 'charged': 0.0, 'used_time': 0, 'used_credits': 0.0, 'description': '', 'charge_rate': 2.0, 'cr_minute': 0.0, 'cr_display_type': 'hour', 'last_operation_time': 1615442820, 'gres_charge_rate': {}, 'gcr_minute': {}, 'gcr_display_type': {}, 'memory_charge_rate': 4.0, 'mcr_minute': 0.0, 'mcr_display_type': 'hour', 'storage_charge_rate': 1.0}}
  39.         >>> query[0].as_dict(include = ['bill_group'])
  40.         {'bill_group': {'id': 1, 'name': 'test', 'balance': 500.0, 'charged': 0.0, 'used_time': 0, 'used_credits': 0.0, 'description': '', 'charge_rate': 2.0, 'cr_minute': 0.0, 'cr_display_type': 'hour', 'last_operation_time': 1615442820, 'gres_charge_rate': {}, 'gcr_minute': {}, 'gcr_display_type': {}, 'memory_charge_rate': 4.0, 'mcr_minute': 0.0, 'mcr_display_type': 'hour', 'storage_charge_rate': 1.0}}
  41.         >>> query[0].as_dict(exclude = ['bill_group'])
  42.         {'id': 1, 'user': 'test', 'credits': -240.0, 'apply_time': 1615432804, 'approved_time': 1615432804, 'billing_type': '', 'billing_id': None, 'balance': -107.0}
  44.     """
  47.     def as_dict(
  48.         self, inspect_related=True,
  49.         include: Optional[Iterable[str]] = None,
  50.         exclude: Optional[Iterable[str]] = None,
  51.         related_field_options: Optional[Dict] = None, ## https://sikasjc.github.io/2018/07/14/type-hint-in-python/
  52.         **on_finished_options
  53.     ):
  54.         if include is not None:
  55.             def is_excluded(field):
  56.                 return field not in include
  57.         else:
  58.             if exclude is None:
  59.                 exclude = set(self.as_dict_exclude)
  60.             else:
  61.                 exclude = set(exclude) | set(self.as_dict_exclude)
  63.             def is_excluded(field):
  64.                 return field in exclude
  66.         if related_field_options is None:
  67.             related_field_options = {}
  69.         result = {}
  70.         for field in self._meta.concrete_fields:
  71.             if is_excluded(field.name) or field.is_relation:
  72.                 continue
  73.             if hasattr(field, 'dict_from_object'):
  74.                 result[field.name] = field.dict_from_object(self)
  75.             else:
  76.                 result[field.name] = field.value_from_object(self)
  78.         if inspect_related:
  79.             self.inspect_related_fields(
  80.                 result, is_excluded, related_field_options
  81.             )
  83.         self.as_dict_on_finished(result, is_excluded, **on_finished_options)
  85.         return result
  87.     def as_dict_on_finished(
  88.         self, result: Dict, is_exlucded: Callable, **kwargs
  89.     ):
  90.         pass
  92.     def inspect_related_fields(
  93.         self, result: Dict, is_excluded: Callable,
  94.         related_field_options: Dict
  95.     ):
  96.         related_fields = (
  97.             rf for rf in self._meta.get_fields()
  98.             if rf.is_relation and not is_excluded(rf.get_cache_name())
  99.         )
  100.         for rf in related_fields:
  101.             option = related_field_options.get(
  102.                 rf.get_cache_name(), dict(inspect_related=False)
  103.             )
  104.             if rf.one_to_many:
  105.                 self._on_inspect_one_to_many_field(rf, result, option)
  106.             if rf.many_to_one:
  107.                 self._on_inspect_many_to_one_field(rf, result, option)
  108.             if rf.one_to_one:
  109.                 self._on_inspect_one_to_one_field(rf, result, option)
  110.             if rf.many_to_many:
  111.                 self._on_inspect_many_to_many_field(rf, result, option)
  113.     def _on_inspect_one_to_many_field(self, rf, result: Dict, option: Dict):
  114.         cache_name = rf.get_cache_name()
  115.         result[cache_name] = [
  116.             related_object.as_dict(**option)
  117.             for related_object in getattr(self, cache_name).iterator()
  118.             if hasattr(related_object, 'as_dict')
  119.         ]
  121.     def _on_inspect_one_to_one_field(self, rf, result: Dict, option: Dict):
  122.         cache_name = rf.get_cache_name()
  123.         related_object = getattr(self, cache_name, None)
  124.         if related_object is None:
  125.             result[cache_name] = None
  126.         if hasattr(related_object, 'as_dict'):
  127.             result[cache_name] = related_object.as_dict(**option)
  129.     def _on_inspect_many_to_one_field(self, rf, result: Dict, option: Dict):
  130.         cache_name = rf.get_cache_name()
  131.         related_object = getattr(self, cache_name, None)
  132.         if related_object is None:
  133.             result[cache_name] = None
  134.         if hasattr(related_object, 'as_dict'):
  135.             result[cache_name] = related_object.as_dict(**option)
  137.     def _on_inspect_many_to_many_field(self, rf, result: Dict, option: Dict):
  138.         cache_name = rf.get_cache_name()
  139.         result[cache_name] = [
  140.             related_object.as_dict(**option)
  141.             for related_object in getattr(self, cache_name).iterator()
  142.             if hasattr(related_object, 'as_dict')
  143.         ]
  146. class QuerySet(BaseQuerySet):
  147.     def as_dict(self, *args, **kwargs):
  148.         return [
  149.             obj.as_dict(*args, **kwargs)
  150.             for obj in self.iterator()
  151.         ]
  153.     def ci_search(self, **kwargs):  # pragma: no cover
  154.         # case insensitive search
  155.         db_table_name = self.model._meta.db_table
  156.         where_clause = ""
  157.         for prop, keyword in kwargs.items():
  158.             if where_clause:
  159.                 where_clause += " OR "
  160.             where_clause += f"{db_table_name}.{prop} " \
  161.                             f"LIKE '%%{keyword}%%' COLLATE utf8_general_ci"
  163.         return self.extra(where=[where_clause])
  165. """
  166.     @classmethod
  167.     def from_queryset(cls, queryset_class, class_name=None):
  168.         if class_name is None:
  169.             class_name = '%sFrom%s' % (cls.__name__, queryset_class.__name__)
  170.         return type(class_name, (cls,), {
  171.             '_queryset_class': queryset_class,
  172.             **cls._get_queryset_methods(queryset_class),
  173.         })
  174. ## return  a new_class
  175. """
  176. class Manager(BaseManager.from_queryset(QuerySet)):
  177.     pass
  180. class Model(BaseModel, ToDictMixin):
  181.     objects = Manager()
  183.     class Meta:
  184.         abstract = True