51 lines
1.5 KiB
Python
51 lines
1.5 KiB
Python
"""asyncpg Record → JSON-serializovatelný dict + helper pro jsonb z fn_*."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
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
|
|
|
|
|
|
async def fetch_json(conn: asyncpg.Connection, query: str, *args: Any) -> Any:
|
|
"""fetchval pro dotazy vracející jsonb (např. select ems.fn_*(...))."""
|
|
v = await conn.fetchval(query, *args)
|
|
if v is None:
|
|
return None
|
|
if isinstance(v, (dict, list)):
|
|
return v
|
|
if isinstance(v, (bytes, memoryview)):
|
|
return json.loads(bytes(v).decode("utf-8"))
|
|
if isinstance(v, str):
|
|
return json.loads(v)
|
|
return v
|