manager.py 2.6 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495
  1. from IPy import IP
  2. from django.core.exceptions import ValidationError
  3. from django.db import models, connection
  4. from django.db.models import sql, query
  5. NET_TERMS = {
  6. 'net_lt': '<',
  7. 'net_lte': '<=',
  8. 'net_exact': '=',
  9. 'net_gte': '>=',
  10. 'net_gt': '>',
  11. 'net_not': '<>',
  12. 'net_is_contained': '<<',
  13. 'net_is_contained_or_equal': '<<=',
  14. 'net_contains': '>>',
  15. 'net_contains_or_equals': '>>=',
  16. }
  17. class NetQuery(sql.Query):
  18. query_terms = sql.Query.query_terms.copy()
  19. query_terms.update(NET_TERMS)
  20. def add_filter(self, (filter_string, value), *args, **kwargs):
  21. if isinstance(value, IP):
  22. value = unicode(value)
  23. return super(NetQuery, self).add_filter((filter_string, value), *args, **kwargs)
  24. class NetWhere(sql.where.WhereNode):
  25. def make_atom(self, child, qn):
  26. table_alias, name, db_type, lookup_type, value_annot, params = child
  27. if db_type in ['cidr', 'inet'] and lookup_type in NET_TERMS:
  28. return ('%s.%s %s inet %%s' % (table_alias, name, NET_TERMS[lookup_type]), params)
  29. return super(NetWhere, self).make_atom(child, qn)
  30. class NetManger(models.Manager):
  31. use_for_related_fields = True
  32. def get_query_set(self):
  33. q = NetQuery(self.model, connection, NetWhere)
  34. return query.QuerySet(self.model, q)
  35. class _NetAddressField(models.Field):
  36. def to_python(self, value):
  37. if not value:
  38. return None
  39. try:
  40. return IP(value)
  41. except ValueError, e:
  42. raise ValidationError(e)
  43. def get_db_prep_value(self, value):
  44. return unicode(self.to_python(value))
  45. def get_db_prep_lookup(self, lookup_type, value):
  46. value = unicode(value)
  47. if lookup_type in INET_TERMS:
  48. return [value]
  49. return super(_NetAddressField, self).get_db_prep_lookup(lookup_type, value)
  50. class InetAddressField(_NetAddressField):
  51. description = "Postgresql inet field"
  52. __metaclass__ = models.SubfieldBase
  53. def db_type(self):
  54. return 'inet'
  55. class CidrAddressField(_NetAddressField):
  56. description = "Postgresql cidr field"
  57. __metaclass__ = models.SubfieldBase
  58. def db_type(self):
  59. return 'cidr'
  60. class MACAddressField(models.Field):
  61. description = "Postgresql macaddr field"
  62. def __init__(self, *args, **kwargs):
  63. kwargs['max_length'] = 17
  64. super(MACAddressField, self).__init__(*args, **kwargs)
  65. def db_type(self):
  66. return 'macaddr'
  67. class Foo(models.Model):
  68. inet = InetAddressField()
  69. test = CidrAddressField()
  70. mac = MACAddressField()
  71. objects = NetManger()