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"