from django.db import models from django.db.models import Q class classproperty: def __init__(self, func): self.fget = func def __get__(self, obj, objtype=None): return self.fget(objtype if objtype is not None else type(obj)) ABBREVIATIONS = { 'uuid': 'UUID', 'id': 'ID', 'ip': 'IP', 'url': 'URL', 'uri': 'URI', 'api': 'API', 'html': 'HTML', 'css': 'CSS', 'js': 'JS', 'json': 'JSON', 'xml': 'XML', 'sql': 'SQL', 'http': 'HTTP', 'https': 'HTTPS', 'smtp': 'SMTP', 'ssl': 'SSL', 'tls': 'TLS', 'ssh': 'SSH', 'ftp': 'FTP', '2fa': '2FA', 'elt': 'ELT', 'sve': 'SVE', 't4': 'T4', 'mc': 'MC', 'crb': 'CRB', 'ccgo': 'CCGO', 'ugm': 'UGM', 'gerp': 'GERP', 'abac': 'ABAC', 'rbac': 'RBAC', 'smtm': 'SMTM', 'pwd': 'PWD', } def snake_to_pascal(name): if '_' not in name: return name[0].upper() + name[1:] if name else name parts = name.split('_') result = [] for p in parts: lower = p.lower() if lower in ABBREVIATIONS: result.append(ABBREVIATIONS[lower]) else: result.append(p.capitalize()) return ''.join(result) def pascal_to_snake(name): import re s1 = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', name) s2 = re.sub('([a-z0-9])([A-Z])', r'\1_\2', s1) parts = s2.lower().split('_') return '_'.join(parts) class _InvertedFieldExpression: def __init__(self, field_expr): self._field_expr = field_expr def startswith(self, value): return ~Q(**{f'{self._field_expr._django_name}__startswith': value}) def contains(self, value): return ~Q(**{f'{self._field_expr._django_name}__contains': value}) def like(self, pattern): clean = pattern.strip('%') return ~Q(**{f'{self._field_expr._django_name}__icontains': clean}) def in_(self, values): return ~Q(**{f'{self._field_expr._django_name}__in': values}) def is_(self, other): if other is None: return ~Q(**{f'{self._field_expr._django_name}__isnull': True}) return ~Q(**{self._field_expr._django_name: other}) class FieldExpression: _resolve_cache = {} def __init__(self, model_class, field_name): self.model_class = model_class self.field_name = field_name self._django_name = self._resolve_django_name() def _resolve_django_name(self): cache_key = (self.model_class.__name__, self.field_name) if cache_key in FieldExpression._resolve_cache: return FieldExpression._resolve_cache[cache_key] if hasattr(self.model_class, '_meta'): for f in self.model_class._meta.local_fields: if f.name == self.field_name: FieldExpression._resolve_cache[cache_key] = f.name return f.name FieldExpression._resolve_cache[cache_key] = self.field_name return self.field_name def __eq__(self, other): if isinstance(other, FieldExpression): return NotImplemented if other is None: return Q(**{f'{self._django_name}__isnull': True}) return Q(**{self._django_name: other}) def __ne__(self, other): if isinstance(other, FieldExpression): return NotImplemented return ~Q(**{self._django_name: other}) def __gt__(self, other): return Q(**{f'{self._django_name}__gt': other}) def __ge__(self, other): return Q(**{f'{self._django_name}__gte': other}) def __lt__(self, other): return Q(**{f'{self._django_name}__lt': other}) def __le__(self, other): return Q(**{f'{self._django_name}__lte': other}) def __getitem__(self, key): return _JsonKeyExpression(self._django_name, key) def __hash__(self): return hash((self.model_class.__name__, self.field_name)) def __and__(self, other): if isinstance(other, Q): return Q(self) & other return NotImplemented def __or__(self, other): if isinstance(other, Q): return Q(self) | other return NotImplemented def __invert__(self): return _InvertedFieldExpression(self) def is_(self, other): if other is None: return Q(**{f'{self._django_name}__isnull': True}) return Q(**{self._django_name: other}) def is_not(self, other): if other is None: return Q(**{f'{self._django_name}__isnull': False}) return ~Q(**{self._django_name: other}) def like(self, pattern): clean = pattern.strip('%') return Q(**{f'{self._django_name}__icontains': clean}) def contains(self, value): return Q(**{f'{self._django_name}__contains': value}) def ilike(self, pattern): return self.like(pattern) def startswith(self, value): return Q(**{f'{self._django_name}__startswith': value}) def istartswith(self, value): return Q(**{f'{self._django_name}__istartswith': value}) def endswith(self, value): return Q(**{f'{self._django_name}__endswith': value}) def iendswith(self, value): return Q(**{f'{self._django_name}__iendswith': value}) def in_(self, values): if hasattr(values, '_qs'): return Q(**{f'{self._django_name}__in': values._qs.values(self._django_name)}) return Q(**{f'{self._django_name}__in': values}) def between(self, low, high): return Q(**{f'{self._django_name}__gte': low}) & Q(**{f'{self._django_name}__lte': high}) def regex(self, pattern): return Q(**{f'{self._django_name}__regex': pattern}) def iregex(self, pattern): return Q(**{f'{self._django_name}__iregex': pattern}) def desc(self): return f'-{self._django_name}' def asc(self): return self._django_name def __repr__(self): return f'FieldExpr({self.model_class.__name__}.{self.field_name})' def __str__(self): return self._django_name class _JsonKeyExpression: def __init__(self, field_name, key): self.field_name = field_name self.key = key def _lookup(self, suffix=''): if suffix: return f'{self.field_name}__{self.key}__{suffix}' return f'{self.field_name}__{self.key}' def cast(self, type_hint): return _CastJsonExpression(self.field_name, self.key, type_hint) def __eq__(self, other): return Q(**{self._lookup(): other}) def __ne__(self, other): return ~Q(**{self._lookup(): other}) def __gt__(self, other): return Q(**{self._lookup('gt'): other}) def __ge__(self, other): return Q(**{self._lookup('gte'): other}) def __lt__(self, other): return Q(**{self._lookup('lt'): other}) def __le__(self, other): return Q(**{self._lookup('lte'): other}) def between(self, low, high): return Q(**{self._lookup('gte'): low}) & Q(**{self._lookup('lte'): high}) def __repr__(self): return f'JsonKeyExpr({self.field_name}[{self.key!r}])' class _CastJsonExpression: def __init__(self, field_name, key, type_hint): self.field_name = field_name self.key = key self.type_hint = type_hint def _lookup(self, suffix=''): if suffix: return f'{self.field_name}__{self.key}__{suffix}' return f'{self.field_name}__{self.key}' def between(self, low, high): return Q(**{self._lookup('gte'): low}) & Q(**{self._lookup('lte'): high}) def __eq__(self, other): return Q(**{self._lookup(): other}) def __ne__(self, other): return ~Q(**{self._lookup(): other}) def __gt__(self, other): return Q(**{self._lookup('gt'): other}) def __ge__(self, other): return Q(**{self._lookup('gte'): other}) def __lt__(self, other): return Q(**{self._lookup('lt'): other}) def __le__(self, other): return Q(**{self._lookup('lte'): other}) def __repr__(self): type_name = getattr(self.type_hint, '__name__', str(self.type_hint)) return f'CastJsonExpr({self.field_name}[{self.key!r}]->{type_name})' class ForeignKeyIdDescriptor: def __init__(self, fk_name): self.fk_name = fk_name self.id_attr = fk_name + '_id' def __get__(self, obj, objtype=None): if obj is None: return FieldExpression(objtype, self.id_attr) return getattr(obj, self.id_attr) def __set__(self, obj, value): setattr(obj, self.id_attr, value) class FieldExpressionDescriptor: def __init__(self, field_name): self.field_name = field_name def __get__(self, obj, objtype=None): if obj is None: return FieldExpression(objtype, self.field_name) return getattr(obj, self.field_name) def __set__(self, obj, value): setattr(obj, self.field_name, value) class QModelBase(models.base.ModelBase): def __new__(mcs, name, bases, namespace, **kwargs): from django.db import models as dj_models for attr_name, attr_val in list(namespace.items()): if isinstance(attr_val, dj_models.Field) and not attr_val.db_column: # ForeignKey/OneToOneField 的数据库列名是 attr_name + '_id' if isinstance(attr_val, (dj_models.ForeignKey, dj_models.OneToOneField)): attr_val.db_column = attr_name + '_id' else: attr_val.db_column = attr_name # 动态设置 managed:子服务器可设 GVSDSDK_MANAGE_TABLES=True 让 gvsdsdk 管理表 meta = namespace.get('Meta') if meta and getattr(meta, 'app_label', None) == 'gvsdsdk': try: from django.conf import settings if getattr(settings, 'GVSDSDK_MANAGE_TABLES', False): meta.managed = True except Exception: pass return super().__new__(mcs, name, bases, namespace, **kwargs) class QModel(models.Model, metaclass=QModelBase): class Meta: abstract = True @classproperty def query(cls): from .fluent import FluentQuery return FluentQuery(cls)