36 lines
1015 B
Python
36 lines
1015 B
Python
"""asyncpg Record → JSON-serializovatelný dict."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import date, datetime, timezone
|
|
from decimal import Decimal
|
|
from typing import Any
|
|
from uuid import UUID
|
|
|
|
import asyncpg
|
|
|
|
|
|
def record_to_dict(r: asyncpg.Record) -> dict[str, Any]:
|
|
out: dict[str, Any] = {}
|
|
for k in r.keys():
|
|
v = r[k]
|
|
if v is None:
|
|
out[k] = None
|
|
elif isinstance(v, datetime):
|
|
if v.tzinfo is None:
|
|
v = v.replace(tzinfo=timezone.utc)
|
|
out[k] = v.isoformat()
|
|
elif isinstance(v, date):
|
|
out[k] = v.isoformat()
|
|
elif isinstance(v, Decimal):
|
|
out[k] = float(v)
|
|
elif isinstance(v, UUID):
|
|
out[k] = str(v)
|
|
elif isinstance(v, (dict, list, str, int, float, bool)):
|
|
out[k] = v
|
|
elif isinstance(v, (bytes, memoryview)):
|
|
out[k] = bytes(v).decode("utf-8", errors="replace")
|
|
else:
|
|
out[k] = str(v)
|
|
return out
|