404 lines
12 KiB
Python
404 lines
12 KiB
Python
from django.db import models
|
|
|
|
|
|
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):
|
|
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):
|
|
from django.db.models import Q
|
|
return ~Q(**{f'{self._field_expr._django_name}__startswith': value})
|
|
|
|
def contains(self, value):
|
|
from django.db.models import Q
|
|
return ~Q(**{f'{self._field_expr._django_name}__contains': value})
|
|
|
|
def like(self, pattern):
|
|
from django.db.models import Q
|
|
clean = pattern.strip('%')
|
|
return ~Q(**{f'{self._field_expr._django_name}__icontains': clean})
|
|
|
|
def in_(self, values):
|
|
from django.db.models import Q
|
|
return ~Q(**{f'{self._field_expr._django_name}__in': values})
|
|
|
|
def is_(self, other):
|
|
from django.db.models import Q
|
|
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
|
|
pascal = snake_to_pascal(f.name)
|
|
if pascal == 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):
|
|
from django.db.models import Q
|
|
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):
|
|
from django.db.models import Q
|
|
if isinstance(other, FieldExpression):
|
|
return NotImplemented
|
|
return ~Q(**{self._django_name: other})
|
|
|
|
def __gt__(self, other):
|
|
from django.db.models import Q
|
|
return Q(**{f'{self._django_name}__gt': other})
|
|
|
|
def __ge__(self, other):
|
|
from django.db.models import Q
|
|
return Q(**{f'{self._django_name}__gte': other})
|
|
|
|
def __lt__(self, other):
|
|
from django.db.models import Q
|
|
return Q(**{f'{self._django_name}__lt': other})
|
|
|
|
def __le__(self, other):
|
|
from django.db.models import Q
|
|
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):
|
|
from django.db.models import Q
|
|
if isinstance(other, Q):
|
|
return Q(self) & other
|
|
return NotImplemented
|
|
|
|
def __or__(self, other):
|
|
from django.db.models import Q
|
|
if isinstance(other, Q):
|
|
return Q(self) | other
|
|
return NotImplemented
|
|
|
|
def __invert__(self):
|
|
return _InvertedFieldExpression(self)
|
|
|
|
def is_(self, other):
|
|
from django.db.models import Q
|
|
if other is None:
|
|
return Q(**{f'{self._django_name}__isnull': True})
|
|
return Q(**{self._django_name: other})
|
|
|
|
def is_not(self, other):
|
|
from django.db.models import Q
|
|
if other is None:
|
|
return Q(**{f'{self._django_name}__isnull': False})
|
|
return ~Q(**{self._django_name: other})
|
|
|
|
def like(self, pattern):
|
|
from django.db.models import Q
|
|
clean = pattern.strip('%')
|
|
return Q(**{f'{self._django_name}__icontains': clean})
|
|
|
|
def contains(self, value):
|
|
from django.db.models import Q
|
|
return Q(**{f'{self._django_name}__contains': value})
|
|
|
|
def ilike(self, pattern):
|
|
return self.like(pattern)
|
|
|
|
def startswith(self, value):
|
|
from django.db.models import Q
|
|
return Q(**{f'{self._django_name}__startswith': value})
|
|
|
|
def istartswith(self, value):
|
|
from django.db.models import Q
|
|
return Q(**{f'{self._django_name}__istartswith': value})
|
|
|
|
def endswith(self, value):
|
|
from django.db.models import Q
|
|
return Q(**{f'{self._django_name}__endswith': value})
|
|
|
|
def iendswith(self, value):
|
|
from django.db.models import Q
|
|
return Q(**{f'{self._django_name}__iendswith': value})
|
|
|
|
def in_(self, values):
|
|
from django.db.models import Q
|
|
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):
|
|
from django.db.models import Q
|
|
return Q(**{f'{self._django_name}__gte': low}) & Q(**{f'{self._django_name}__lte': high})
|
|
|
|
def regex(self, pattern):
|
|
from django.db.models import Q
|
|
return Q(**{f'{self._django_name}__regex': pattern})
|
|
|
|
def iregex(self, pattern):
|
|
from django.db.models import Q
|
|
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):
|
|
from django.db.models import Q
|
|
return Q(**{self._lookup(): other})
|
|
|
|
def __ne__(self, other):
|
|
from django.db.models import Q
|
|
return ~Q(**{self._lookup(): other})
|
|
|
|
def __gt__(self, other):
|
|
from django.db.models import Q
|
|
return Q(**{self._lookup('gt'): other})
|
|
|
|
def __ge__(self, other):
|
|
from django.db.models import Q
|
|
return Q(**{self._lookup('gte'): other})
|
|
|
|
def __lt__(self, other):
|
|
from django.db.models import Q
|
|
return Q(**{self._lookup('lt'): other})
|
|
|
|
def __le__(self, other):
|
|
from django.db.models import Q
|
|
return Q(**{self._lookup('lte'): other})
|
|
|
|
def between(self, low, high):
|
|
from django.db.models import Q
|
|
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):
|
|
from django.db.models import Q
|
|
return Q(**{self._lookup('gte'): low}) & Q(**{self._lookup('lte'): high})
|
|
|
|
def __eq__(self, other):
|
|
from django.db.models import Q
|
|
return Q(**{self._lookup(): other})
|
|
|
|
def __ne__(self, other):
|
|
from django.db.models import Q
|
|
return ~Q(**{self._lookup(): other})
|
|
|
|
def __gt__(self, other):
|
|
from django.db.models import Q
|
|
return Q(**{self._lookup('gt'): other})
|
|
|
|
def __ge__(self, other):
|
|
from django.db.models import Q
|
|
return Q(**{self._lookup('gte'): other})
|
|
|
|
def __lt__(self, other):
|
|
from django.db.models import Q
|
|
return Q(**{self._lookup('lt'): other})
|
|
|
|
def __le__(self, other):
|
|
from django.db.models import Q
|
|
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 PascalCaseModelBase(models.base.ModelBase):
|
|
def __new__(mcs, name, bases, namespace, **kwargs):
|
|
from django.db import models as dj_models
|
|
fk_fields = {}
|
|
for attr_name, attr_val in list(namespace.items()):
|
|
if isinstance(attr_val, dj_models.Field) and not attr_val.db_column:
|
|
attr_val.db_column = snake_to_pascal(attr_name)
|
|
if isinstance(attr_val, dj_models.ForeignKey):
|
|
fk_fields[attr_name] = attr_val
|
|
cls = super().__new__(mcs, name, bases, namespace, **kwargs)
|
|
|
|
fk_field_names = set()
|
|
for fk_name in fk_fields:
|
|
fk_field_names.add(fk_name)
|
|
|
|
for field in cls._meta.local_fields:
|
|
if field.name in fk_field_names:
|
|
continue
|
|
pascal_name = snake_to_pascal(field.name)
|
|
if pascal_name != field.name and not hasattr(cls, pascal_name):
|
|
try:
|
|
desc = FieldExpressionDescriptor(field.name)
|
|
setattr(cls, pascal_name, desc)
|
|
except Exception:
|
|
pass
|
|
|
|
for fk_name, fk_field in fk_fields.items():
|
|
pascal_col = snake_to_pascal(fk_name)
|
|
if not hasattr(cls, pascal_col):
|
|
try:
|
|
desc = ForeignKeyIdDescriptor(fk_name)
|
|
setattr(cls, pascal_col, desc)
|
|
except Exception:
|
|
pass
|
|
|
|
return cls
|
|
|
|
|
|
class PascalCaseModel(models.Model, metaclass=PascalCaseModelBase):
|
|
class Meta:
|
|
abstract = True
|
|
|
|
@classproperty
|
|
def query(cls):
|
|
from gvsds.fluent import FluentQuery
|
|
return FluentQuery(cls)
|