|
1 | 1 | from __future__ import annotations |
2 | 2 |
|
3 | | -from json import loads, dumps |
4 | | -from sqlalchemy.types import TypeDecorator, TEXT |
| 3 | +from sqlalchemy.ext.mutable import MutableDict |
| 4 | +from sqlalchemy.types import TypeDecorator |
| 5 | +from sqlalchemy.dialects.postgresql import JSONB |
5 | 6 |
|
6 | 7 |
|
7 | 8 | from typing import Any |
8 | 9 | from typing import TYPE_CHECKING |
9 | 10 | if TYPE_CHECKING: |
10 | 11 | from sqlalchemy.engine import Dialect |
11 | 12 |
|
12 | | - _Base = TypeDecorator[Any] |
| 13 | + _Base = TypeDecorator[dict[str, Any]] |
13 | 14 | else: |
14 | 15 | _Base = TypeDecorator |
15 | 16 |
|
16 | 17 |
|
17 | 18 | class JSON(_Base): |
18 | | - """Like the default JSON, but using the json serializer from the dialect |
19 | | - (postgres) each time the value is read, even if it never left the ORM. The |
20 | | - default json type will only do it when the record is read from the |
21 | | - database. |
| 19 | + """ A JSONB based type that coerces None's to empty dictionaries. |
| 20 | +
|
| 21 | + That is, this JSONB column cannot be `'null'::jsonb`. It could |
| 22 | + still be `NULL` though, if it's nullable and never explicitly |
| 23 | + set. But on the Python end you should always see a dictionary. |
22 | 24 |
|
23 | 25 | """ |
24 | 26 |
|
25 | | - # FIXME: We should switch to JSONB now, but we will need to |
26 | | - # add a database migration to OneGov at the same time |
27 | | - impl = TEXT |
| 27 | + impl = JSONB |
28 | 28 |
|
29 | | - def process_bind_param( |
| 29 | + def process_bind_param( # type:ignore[override] |
30 | 30 | self, |
31 | | - value: Any, |
| 31 | + value: dict[str, Any] | None, |
32 | 32 | dialect: Dialect |
33 | | - ) -> str | None: |
34 | | - |
35 | | - if value is not None: |
36 | | - value = (dialect._json_serializer or dumps)(value) # type:ignore |
| 33 | + ) -> dict[str, Any]: |
37 | 34 |
|
38 | | - return value # type: ignore[no-any-return] |
| 35 | + return {} if value is None else value |
39 | 36 |
|
40 | 37 | def process_result_value( |
41 | 38 | self, |
42 | | - value: str | None, |
| 39 | + value: dict[str, Any] | None, |
43 | 40 | dialect: Dialect |
44 | | - ) -> Any | None: |
| 41 | + ) -> dict[str, Any]: |
| 42 | + |
| 43 | + return {} if value is None else value |
45 | 44 |
|
46 | | - if value is not None: |
47 | | - value = (dialect._json_deserializer or loads)(value) # type:ignore |
48 | 45 |
|
49 | | - return value |
| 46 | +MutableDict.associate_with(JSON) # type:ignore[no-untyped-call] |
0 commit comments