1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
|
from typing import Annotated
from odmantic import AIOEngine, Model, WithBsonSerializer
class ASCIISerializedAsBinaryBase(str):
@classmethod
def __get_validators__(cls):
yield cls.validate
@classmethod
def validate(cls, v):
if isinstance(v, bytes): # Handle data coming from MongoDB
return v.decode("ascii")
if not isinstance(v, str):
raise TypeError("string required")
if not v.isascii():
raise ValueError("Only ascii characters are allowed")
return v
def serialize_ascii_to_bytes(v: ASCIISerializedAsBinaryBase) -> bytes:
# We can encode this string as ascii since it contains
# only ascii characters
bytes_ = v.encode("ascii")
return bytes_
ASCIISerializedAsBinary = Annotated[
ASCIISerializedAsBinaryBase, WithBsonSerializer(serialize_ascii_to_bytes)
]
class Example(Model):
field: ASCIISerializedAsBinary
engine = AIOEngine()
await engine.save(Example(field="hello world"))
fetched = await engine.find_one(Example)
print(fetched.field)
#> hello world
|
