1from collections.abc import Callable, Sequence
  2from typing import TYPE_CHECKING, Any
  3
  4from plain.postgres.dialect import adapt_ipaddressfield_value
  5from plain.preflight import PreflightResult
  6from plain.utils.ipv6 import clean_ipv6_address
  7from plain.validators import ip_address_validators
  8
  9from plain import exceptions
 10
 11from .base import NOT_PROVIDED, DefaultableField
 12
 13if TYPE_CHECKING:
 14    from plain.postgres.connection import DatabaseConnection
 15
 16
 17class GenericIPAddressField[T: (str, str | None) = str](DefaultableField[T]):
 18    db_type_sql = "inet"
 19    empty_strings_allowed = False
 20
 21    def __init__(
 22        self,
 23        *,
 24        protocol: str = "both",
 25        unpack_ipv4: bool = False,
 26        required: bool = True,
 27        allow_null: bool = False,
 28        default: Any = NOT_PROVIDED,
 29        validators: Sequence[Callable[..., Any]] = (),
 30    ):
 31        self.unpack_ipv4 = unpack_ipv4
 32        self.protocol = protocol
 33        (
 34            self.default_validators,
 35            self.invalid_error_message,
 36        ) = ip_address_validators(protocol, unpack_ipv4)
 37        super().__init__(
 38            required=required,
 39            allow_null=allow_null,
 40            default=default,
 41            validators=validators,
 42        )
 43
 44    def preflight(self, **kwargs: Any) -> list[PreflightResult]:
 45        return [
 46            *super().preflight(**kwargs),
 47            *self._check_required_and_null_values(),
 48        ]
 49
 50    def _check_required_and_null_values(self) -> list[PreflightResult]:
 51        if not getattr(self, "allow_null", False) and not getattr(
 52            self, "required", True
 53        ):
 54            return [
 55                PreflightResult(
 56                    fix="GenericIPAddressFields cannot have required=False if allow_null=False, "
 57                    "as blank values are stored as nulls.",
 58                    obj=self,
 59                    id="fields.generic_ip_field_null_blank_config",
 60                )
 61            ]
 62        return []
 63
 64    def deconstruct(self) -> tuple[str, str, list[Any], dict[str, Any]]:
 65        name, path, args, kwargs = super().deconstruct()
 66        if self.unpack_ipv4 is not False:
 67            kwargs["unpack_ipv4"] = self.unpack_ipv4
 68        if self.protocol != "both":
 69            kwargs["protocol"] = self.protocol
 70        return name, path, args, kwargs
 71
 72    def to_python(self, value: Any) -> str | None:
 73        if value is None:
 74            return None
 75        if not isinstance(value, str):
 76            value = str(value)
 77        value = value.strip()
 78        if ":" in value:
 79            return clean_ipv6_address(
 80                value, self.unpack_ipv4, self.invalid_error_message
 81            )
 82        return value
 83
 84    def get_db_prep_value(
 85        self, value: Any, connection: DatabaseConnection, prepared: bool = False
 86    ) -> Any:
 87        if not prepared:
 88            value = self.get_prep_value(value)
 89        return adapt_ipaddressfield_value(value)
 90
 91    def get_prep_value(self, value: Any) -> Any:
 92        value = super().get_prep_value(value)
 93        if value is None:
 94            return None
 95        if value and ":" in value:
 96            try:
 97                return clean_ipv6_address(value, self.unpack_ipv4)
 98            except exceptions.ValidationError:
 99                pass
100        return str(value)