1666511475
- services/kb-postgres: pgvector/pgvector:pg16 na SOLARIA (:5433), named volume, init/001_envelope.sql (CREATE EXTENSION vector + zamrożona tabela envelope: id/source/ts/geo/raw_ref/entities), service.yaml, healthcheck, README z poprawnym mechanizmem deploy (deploy-node.sh składa dwa -f) - hosts/solaria/runtime/kb-postgres/docker-compose.override.yml: mem_limit 4g - inventory/topology.yaml + hosts/solaria/services.yaml: kb-postgres wpisany - packages/kb-mail: nowa konwencja shared lib (pip install /repo/packages/<lib>/) envelope.py — @dataclass Envelope, walidacja tz-aware ts db.py — insert_envelope / get_envelope (asyncpg, ON CONFLICT DO NOTHING) archive.py — save_eml append-only (asyncio.to_thread, FileExistsError na dup) tests: 15 unit pass + 5 integration (@pytest.mark.integration, wymaga KB_TEST_DSN) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
68 lines
1.9 KiB
Python
68 lines
1.9 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import timezone
|
|
from typing import Optional
|
|
|
|
import asyncpg
|
|
import structlog
|
|
|
|
from .envelope import Envelope
|
|
|
|
_log = structlog.get_logger(__name__)
|
|
|
|
|
|
async def insert_envelope(conn: asyncpg.Connection, env: Envelope) -> None:
|
|
"""Insert envelope into kb-postgres. Silently ignores duplicate ids (ON CONFLICT DO NOTHING)."""
|
|
geo_param = json.dumps(env.geo) if env.geo is not None else None
|
|
await conn.execute(
|
|
"""
|
|
INSERT INTO envelope (id, source, ts, geo, raw_ref, entities)
|
|
VALUES ($1, $2, $3, $4::jsonb, $5, $6::jsonb)
|
|
ON CONFLICT (id) DO NOTHING
|
|
""",
|
|
env.id,
|
|
env.source,
|
|
env.ts,
|
|
geo_param,
|
|
env.raw_ref,
|
|
json.dumps(env.entities),
|
|
)
|
|
_log.info("envelope.inserted", id=env.id, source=env.source)
|
|
|
|
|
|
async def get_envelope(conn: asyncpg.Connection, envelope_id: str) -> Optional[Envelope]:
|
|
"""Fetch a single envelope by id. Returns None if not found."""
|
|
row = await conn.fetchrow(
|
|
"SELECT id, source, ts, geo, raw_ref, entities FROM envelope WHERE id = $1",
|
|
envelope_id,
|
|
)
|
|
if row is None:
|
|
return None
|
|
|
|
ts = row["ts"]
|
|
if ts.tzinfo is None:
|
|
ts = ts.replace(tzinfo=timezone.utc)
|
|
|
|
# asyncpg decodes JSONB columns to Python objects; handle both str and object
|
|
geo = _decode_jsonb(row["geo"])
|
|
entities = _decode_jsonb(row["entities"]) or []
|
|
|
|
return Envelope(
|
|
id=row["id"],
|
|
source=row["source"],
|
|
ts=ts,
|
|
geo=geo,
|
|
raw_ref=row["raw_ref"],
|
|
entities=entities,
|
|
)
|
|
|
|
|
|
def _decode_jsonb(value: object) -> object:
|
|
"""Decode a JSONB value that may arrive as a string or already-decoded object."""
|
|
if value is None:
|
|
return None
|
|
if isinstance(value, str):
|
|
return json.loads(value)
|
|
return value
|