|
1 | 1 | import json |
| 2 | +from typing import Any |
2 | 3 |
|
3 | 4 | import pytest |
4 | 5 | from dirty_equals import IsStrictDict |
| 6 | +from typing_extensions import TypedDict |
5 | 7 |
|
6 | 8 | from pydantic_core import SchemaSerializer, core_schema |
7 | 9 |
|
@@ -159,3 +161,89 @@ def test_exclude_default(): |
159 | 161 |
|
160 | 162 | assert v.to_json({'foo': 1, 'bar': b'[default]'}) == b'{"foo":1,"bar":"[default]"}' |
161 | 163 | assert v.to_json({'foo': 1, 'bar': b'[default]'}, exclude_defaults=True) == b'{"foo":1}' |
| 164 | + |
| 165 | + |
| 166 | +def test_function_plain_field_serializer_to_python(): |
| 167 | + class Model(TypedDict): |
| 168 | + x: int |
| 169 | + |
| 170 | + def ser_x(data: Model, v: Any, _) -> str: |
| 171 | + assert data['x'] == 1_000 |
| 172 | + return f'{v:_}' |
| 173 | + |
| 174 | + s = SchemaSerializer( |
| 175 | + core_schema.typed_dict_schema( |
| 176 | + { |
| 177 | + 'x': core_schema.typed_dict_field( |
| 178 | + core_schema.int_schema(serialization=core_schema.field_function_plain_ser_schema(ser_x)) |
| 179 | + ) |
| 180 | + } |
| 181 | + ) |
| 182 | + ) |
| 183 | + assert s.to_python(Model(x=1000)) == {'x': '1_000'} |
| 184 | + |
| 185 | + |
| 186 | +def test_function_wrap_field_serializer_to_python(): |
| 187 | + class Model(TypedDict): |
| 188 | + x: int |
| 189 | + |
| 190 | + def ser_x(data: Model, v: Any, serializer: core_schema.SerializeWrapHandler, _) -> str: |
| 191 | + x = serializer(v) |
| 192 | + assert data['x'] == 1_000 |
| 193 | + return f'{x:_}' |
| 194 | + |
| 195 | + s = SchemaSerializer( |
| 196 | + core_schema.typed_dict_schema( |
| 197 | + { |
| 198 | + 'x': core_schema.typed_dict_field( |
| 199 | + core_schema.int_schema( |
| 200 | + serialization=core_schema.field_function_wrap_ser_schema(ser_x, schema=core_schema.any_schema()) |
| 201 | + ) |
| 202 | + ) |
| 203 | + } |
| 204 | + ) |
| 205 | + ) |
| 206 | + assert s.to_python(Model(x=1000)) == {'x': '1_000'} |
| 207 | + |
| 208 | + |
| 209 | +def test_function_plain_field_serializer_to_json(): |
| 210 | + class Model(TypedDict): |
| 211 | + x: int |
| 212 | + |
| 213 | + def ser_x(data: Model, v: Any, _) -> str: |
| 214 | + assert data['x'] == 1_000 |
| 215 | + return f'{v:_}' |
| 216 | + |
| 217 | + s = SchemaSerializer( |
| 218 | + core_schema.typed_dict_schema( |
| 219 | + { |
| 220 | + 'x': core_schema.typed_dict_field( |
| 221 | + core_schema.int_schema(serialization=core_schema.field_function_plain_ser_schema(ser_x)) |
| 222 | + ) |
| 223 | + } |
| 224 | + ) |
| 225 | + ) |
| 226 | + assert json.loads(s.to_json(Model(x=1000))) == {'x': '1_000'} |
| 227 | + |
| 228 | + |
| 229 | +def test_function_wrap_field_serializer_to_json(): |
| 230 | + class Model(TypedDict): |
| 231 | + x: int |
| 232 | + |
| 233 | + def ser_x(data: Model, v: Any, serializer: core_schema.SerializeWrapHandler, _) -> str: |
| 234 | + assert data['x'] == 1_000 |
| 235 | + x = serializer(v) |
| 236 | + return f'{x:_}' |
| 237 | + |
| 238 | + s = SchemaSerializer( |
| 239 | + core_schema.typed_dict_schema( |
| 240 | + { |
| 241 | + 'x': core_schema.typed_dict_field( |
| 242 | + core_schema.int_schema( |
| 243 | + serialization=core_schema.field_function_wrap_ser_schema(ser_x, schema=core_schema.any_schema()) |
| 244 | + ) |
| 245 | + ) |
| 246 | + } |
| 247 | + ) |
| 248 | + ) |
| 249 | + assert json.loads(s.to_json(Model(x=1000))) == {'x': '1_000'} |
0 commit comments