1from base64 import b64decode
 2from collections.abc import Callable, Sequence
 3from typing import TYPE_CHECKING, Any
 4
 5import psycopg
 6from plain.validators import MaxLengthValidator
 7
 8from .base import NOT_PROVIDED, DefaultableField
 9
10if TYPE_CHECKING:
11    from plain.postgres.connection import DatabaseConnection
12    from plain.postgres.sql.compiler import SQLCompiler
13
14
15class BinaryField[
16    T: (bytes | memoryview, bytes | memoryview | None) = bytes | memoryview
17](DefaultableField[T]):
18    db_type_sql = "bytea"
19    empty_values = (None, b"")
20    _default_empty_value = b""
21    only_empty_default = True
22
23    def __init__(
24        self,
25        *,
26        max_length: int | None = None,
27        required: bool = True,
28        allow_null: bool = False,
29        default: Any = NOT_PROVIDED,
30        validators: Sequence[Callable[..., Any]] = (),
31    ):
32        self.max_length = max_length
33        super().__init__(
34            required=required,
35            allow_null=allow_null,
36            default=default,
37            validators=validators,
38        )
39        if self.max_length is not None:
40            self.validators.append(MaxLengthValidator(self.max_length))
41
42    def deconstruct(self) -> tuple[str, str, list[Any], dict[str, Any]]:
43        name, path, args, kwargs = super().deconstruct()
44        if self.max_length is not None:
45            kwargs["max_length"] = self.max_length
46        return name, path, args, kwargs
47
48    def get_placeholder(
49        self, value: Any, compiler: SQLCompiler, connection: DatabaseConnection
50    ) -> Any:
51        return "%s"
52
53    def get_db_prep_value(
54        self, value: Any, connection: DatabaseConnection, prepared: bool = False
55    ) -> Any:
56        value = super().get_db_prep_value(value, connection, prepared)
57        if value is not None:
58            return psycopg.Binary(value)
59        return value
60
61    def to_python(self, value: Any) -> bytes | memoryview | None:
62        # If it's a string, it should be base64-encoded data
63        if isinstance(value, str):
64            return memoryview(b64decode(value.encode("ascii")))
65        return value