1from __future__ import annotations
 2
 3from typing import TYPE_CHECKING, Any, cast
 4
 5from plain import exceptions
 6from plain.preflight import PreflightResult
 7
 8from .base import ColumnField
 9
10if TYPE_CHECKING:
11    from plain.postgres.base import Model
12    from plain.postgres.connection import DatabaseConnection
13
14
15class PrimaryKeyField(ColumnField[int]):
16    """Identity primary key โ€” Postgres generates the value on INSERT.
17
18    Accepts no constructor arguments: the column is always ``bigint GENERATED
19    BY DEFAULT AS IDENTITY``, NOT NULL, with RETURNING populating the value
20    back onto the instance after save.
21    """
22
23    db_type_sql = "bigint"
24    cast_db_type_sql = "bigint"
25    db_type_suffix_sql = "GENERATED BY DEFAULT AS IDENTITY"
26    empty_strings_allowed = False
27
28    def __init__(self) -> None:
29        super().__init__(required=False)
30        self.primary_key = True
31        self.auto_created = True
32
33    @property
34    def db_returning(self) -> bool:
35        return True
36
37    def preflight(self, **kwargs: Any) -> list[PreflightResult]:
38        errors = super().preflight(**kwargs)
39        # `id` is the reserved PK name โ€” Field._check_field_name rejects it
40        # for every other field type.
41        return [e for e in errors if e.id != "fields.reserved_field_name_id"]
42
43    def deconstruct(self) -> tuple[str | None, str, list[Any], dict[str, Any]]:
44        return (
45            self.name,
46            "plain.postgres.PrimaryKeyField",
47            cast(list[Any], []),
48            cast(dict[str, Any], {}),
49        )
50
51    def validate(self, value: Any, model_instance: Model) -> None:
52        # Identity columns aren't user-validated โ€” Postgres owns the value.
53        pass
54
55    def to_python(self, value: Any) -> int | None:
56        if value is None:
57            return value
58        try:
59            return int(value)
60        except (TypeError, ValueError):
61            raise exceptions.ValidationError(
62                '"%(value)s" value must be an integer.',
63                code="invalid",
64                params={"value": value},
65            )
66
67    def get_prep_value(self, value: Any) -> Any:
68        value = super().get_prep_value(value)
69        if value is None:
70            return None
71        try:
72            return int(value)
73        except (TypeError, ValueError) as e:
74            raise e.__class__(
75                f"Field '{self.name}' expected a number but got {value!r}.",
76            ) from e
77
78    def get_db_prep_value(
79        self, value: Any, connection: DatabaseConnection, prepared: bool = False
80    ) -> Any:
81        if not prepared:
82            value = self.get_prep_value(value)
83        return value
84
85    def rel_db_type(self) -> str:
86        return "bigint"