461 lines
14 KiB
Python
461 lines
14 KiB
Python
from __future__ import annotations
|
||
|
||
import typing
|
||
from typing import Any, Dict, List, Optional, Tuple, Type, Union
|
||
|
||
from django.db import models
|
||
from django.db.models import Q
|
||
from .fluent import FluentQuery
|
||
|
||
|
||
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:
|
||
"""FK 字段描述符：类级别访问返回 FieldExpression，实例级别访问/设置委托给原始 Django 字段。"""
|
||
def __init__(self, fk_name):
|
||
self.fk_name = fk_name
|
||
self.id_attr = fk_name + '_id'
|
||
self._original_field = None # 延迟绑定原始 Django 字段
|
||
|
||
def _get_original_field(self, owner):
|
||
if self._original_field is None:
|
||
try:
|
||
self._original_field = owner._meta.get_field(self.fk_name)
|
||
except Exception:
|
||
pass
|
||
return self._original_field
|
||
|
||
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):
|
||
# 直接写入实例 __dict__，避免触发描述符递归
|
||
obj.__dict__[self.id_attr] = value
|
||
|
||
|
||
class FieldExpressionDescriptor:
|
||
"""普通字段描述符：类级别访问返回 FieldExpression，实例级别访问/设置委托给原始 Django 字段。"""
|
||
def __init__(self, field_name):
|
||
self.field_name = field_name
|
||
self._original_field = None # 延迟绑定原始 Django 字段
|
||
|
||
def _get_original_field(self, owner):
|
||
if self._original_field is None:
|
||
try:
|
||
self._original_field = owner._meta.get_field(self.field_name)
|
||
except Exception:
|
||
pass
|
||
return self._original_field
|
||
|
||
def __get__(self, obj, objtype=None):
|
||
if obj is None:
|
||
return FieldExpression(objtype, self.field_name)
|
||
# 实例级别：从 __dict__ 获取，或让 Django 字段描述符处理
|
||
if self.field_name in obj.__dict__:
|
||
return obj.__dict__[self.field_name]
|
||
# 委托给原始 Django 字段
|
||
orig = self._get_original_field(objtype)
|
||
if orig is not None:
|
||
return orig.__get__(obj, objtype)
|
||
return getattr(obj, self.field_name)
|
||
|
||
def __set__(self, obj, value):
|
||
# 直接写入实例 __dict__，避免触发描述符递归
|
||
obj.__dict__[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
|
||
|
||
cls = super().__new__(mcs, name, bases, namespace, **kwargs)
|
||
|
||
# 模型创建后，自动为所有字段添加描述符，支持 Model.field_name 链式 OOP
|
||
if not cls._meta.abstract:
|
||
_attach_field_descriptors(cls)
|
||
|
||
return cls
|
||
|
||
|
||
def _attach_field_descriptors(cls: Type) -> None:
|
||
"""为 QModel 子类的所有字段自动添加 FieldExpressionDescriptor / ForeignKeyIdDescriptor，
|
||
使 Model.field_name 在类级别访问时返回 FieldExpression，支持链式 OOP 查询。"""
|
||
from django.db import models as dj_models
|
||
|
||
for field in cls._meta.local_fields:
|
||
fname = field.name
|
||
# 跳过已有描述符或内部属性
|
||
if fname.startswith('_'):
|
||
continue
|
||
# 检查是否已有同名描述符（避免覆盖手动定义的）
|
||
existing = cls.__dict__.get(fname)
|
||
if isinstance(existing, (FieldExpressionDescriptor, ForeignKeyIdDescriptor)):
|
||
continue
|
||
# 跳过 Django 自动创建的反向关系和 id 主键（如果用户未显式定义）
|
||
if fname == 'id' and field.primary_key and not isinstance(
|
||
cls.__dict__.get(fname), (FieldExpressionDescriptor, ForeignKeyIdDescriptor)
|
||
):
|
||
# id 字段也添加描述符
|
||
setattr(cls, fname, FieldExpressionDescriptor(fname))
|
||
continue
|
||
|
||
if isinstance(field, (dj_models.ForeignKey, dj_models.OneToOneField)):
|
||
# FK 字段：类级别访问返回 FieldExpression，实例级别访问返回关联对象
|
||
setattr(cls, fname, ForeignKeyIdDescriptor(fname))
|
||
else:
|
||
setattr(cls, fname, FieldExpressionDescriptor(fname))
|
||
|
||
|
||
class QModel(models.Model, metaclass=QModelBase):
|
||
"""QModel 兼容层基类，提供 SQLAlchemy 风格的链式 OOP 查询接口。
|
||
|
||
所有继承 QModel 的模型自动获得：
|
||
- ``Model.query`` : 返回 FluentQuery，支持链式查询
|
||
- ``Model.field_name`` : 类级别访问返回 FieldExpression，支持表达式查询
|
||
- ``Model.field_name`` : 实例级别访问返回字段值（与普通 Django Model 一致）
|
||
|
||
用法示例::
|
||
|
||
# 链式查询
|
||
orders = Dingdan.query.filter(zhuangtai=8).order_by('-create_time').to_list()
|
||
|
||
# 字段表达式
|
||
Dingdan.query.filter(Dingdan.jine > 100, Dingdan.zhuangtai.in_([1,2,3]))
|
||
|
||
# 聚合
|
||
Dingdan.query.aggregate(total=Sum('jine'))
|
||
|
||
# values / annotate
|
||
Dingdan.query.values('zhuangtai').annotate(cnt=Count('id'))
|
||
|
||
# 分页
|
||
Dingdan.query.paginate(page=1, per_page=20)
|
||
"""
|
||
|
||
class Meta:
|
||
abstract = True
|
||
|
||
@classproperty
|
||
def query(cls) -> FluentQuery:
|
||
"""返回 FluentQuery 兼容层查询构建器，支持链式 OOP 查询。"""
|
||
return FluentQuery(cls)
|
||
|
||
@classmethod
|
||
def __class_getitem__(cls, item) -> FluentQuery:
|
||
"""支持 Model[condition] 语法，返回 filter 后的 FluentQuery。
|
||
|
||
用法::
|
||
|
||
Dingdan[Dingdan.zhuangtai == 8].order_by('-create_time').to_list()
|
||
"""
|
||
if isinstance(item, Q):
|
||
return cls.query.filter(item)
|
||
if isinstance(item, dict):
|
||
return cls.query.filter(**item)
|
||
return cls.query.filter(item)
|
||
|
||
|