« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
1"""SQL-backed durable relay channel store.
3Persists channel rows through
4:class:`~lexigram.contracts.data.DatabaseProviderProtocol` using only
5generic ``execute``/``execute_query`` SQL so the store works on any
6backend. Every mutation is compare-and-set on ``revision``: stale
7writers (a mismatched ``expected_revision``) are rejected instead of
8overwriting newer state, and ``delete`` only removes the row matching
9the expected revision.
10"""
12from __future__ import annotations
14from typing import TYPE_CHECKING, Any
16from lexigram.contracts.ai.relay.gateway import RelayChannel
17from lexigram.contracts.ai.relay.store import (
18 RelayChannelSnapshot,
19 RelayChannelStoreProtocol,
20)
21from lexigram.contracts.ai.relay.types import RelayFormat
22from lexigram.primitives import clock
23from lexigram.serialization import dumps_str, loads_str
25if TYPE_CHECKING:
26 from lexigram.contracts.data import DatabaseProviderProtocol
28__all__ = ["SqlRelayChannelStore"]
30_CREATE_TABLE = """
31CREATE TABLE IF NOT EXISTS ai_relay_channels (
32 name TEXT NOT NULL PRIMARY KEY,
33 payload TEXT NOT NULL,
34 revision INTEGER NOT NULL,
35 created_at TEXT NOT NULL,
36 updated_at TEXT NOT NULL
37)
38"""
40_SELECT_ROW = "SELECT revision FROM ai_relay_channels WHERE name = ?"
42_SELECT_ALL = (
43 "SELECT name, payload, revision, created_at, updated_at "
44 "FROM ai_relay_channels ORDER BY name"
45)
47_INSERT_ROW = (
48 "INSERT INTO ai_relay_channels "
49 "(name, payload, revision, created_at, updated_at) VALUES (?, ?, 1, ?, ?)"
50)
52_UPDATE_ROW = (
53 "UPDATE ai_relay_channels SET payload = ?, updated_at = ?, revision = ? "
54 "WHERE name = ? AND revision = ?"
55)
57_DELETE_ROW = "DELETE FROM ai_relay_channels WHERE name = ? AND revision = ?"
60def _payload(channel: RelayChannel) -> str:
61 """Encode a channel as canonical JSON (sets sorted, enum values)."""
62 return dumps_str(
63 {
64 "name": channel.name,
65 "upstream_base_url": channel.upstream_base_url,
66 "target_format": channel.target_format.value,
67 "models": list(channel.models),
68 "capabilities": sorted(channel.capabilities),
69 "endpoint_kinds": sorted(channel.endpoint_kinds),
70 "priority": channel.priority,
71 "weight": channel.weight,
72 "enabled": channel.enabled,
73 "timeout_seconds": channel.timeout_seconds,
74 }
75 )
78def _decode_channel(data: dict[str, Any]) -> RelayChannel:
79 """Rebuild a ``RelayChannel`` from a canonical payload dict."""
80 return RelayChannel(
81 name=data["name"],
82 upstream_base_url=data["upstream_base_url"],
83 target_format=RelayFormat(data["target_format"]),
84 models=tuple(data["models"]),
85 capabilities=frozenset(data["capabilities"]),
86 endpoint_kinds=frozenset(data["endpoint_kinds"]),
87 priority=data["priority"],
88 weight=data["weight"],
89 enabled=data["enabled"],
90 timeout_seconds=data["timeout_seconds"],
91 )
94class SqlRelayChannelStore(RelayChannelStoreProtocol):
95 """SQL-backed CRUD over durable channel rows.
97 Creates the ``ai_relay_channels`` table lazily on first use.
98 ``upsert`` matches on ``channel.name`` and returns the new revision;
99 ``delete`` returns whether a row was removed.
101 Args:
102 db: A connected
103 :class:`~lexigram.contracts.data.DatabaseProviderProtocol`
104 resolved from the DI container.
105 """
107 def __init__(self, db: DatabaseProviderProtocol) -> None:
108 self._db = db
109 self._initialised = False
111 async def _ensure_tables(self) -> None:
112 """Create the storage schema once, on first use."""
113 if not self._initialised:
114 await self._db.execute(_CREATE_TABLE)
115 self._initialised = True
117 async def list_channels(self) -> list[RelayChannelSnapshot]:
118 """Return all durable channels ordered by name.
120 Returns:
121 One snapshot per stored channel.
122 """
123 await self._ensure_tables()
124 result = await self._db.execute_query(_SELECT_ALL)
125 return [
126 RelayChannelSnapshot(
127 channel=_decode_channel(loads_str(row["payload"])),
128 revision=int(row["revision"]),
129 created_at=row["created_at"],
130 updated_at=row["updated_at"],
131 )
132 for row in result.rows
133 ]
135 async def upsert(
136 self, channel: RelayChannel, *, expected_revision: int | None = None
137 ) -> int | None:
138 """Insert or update *channel* under compare-and-set.
140 A new name inserts at revision 1 (any ``expected_revision`` on an
141 absent row is treated as a stale write and rejected). An
142 existing row updates only when ``expected_revision`` matches its
143 current revision, bumping the revision by one.
145 Args:
146 channel: The channel to persist.
147 expected_revision: Revision the caller observed; ``None``
148 for a blind write (on an existing name this still
149 verifies against the live revision).
151 Returns:
152 The new revision on success, ``None`` on a stale write.
153 """
154 await self._ensure_tables()
155 now = clock.now().isoformat()
156 payload = _payload(channel)
157 current = await self._db.execute_query(_SELECT_ROW, [channel.name])
158 if not current.rows:
159 if expected_revision is not None:
160 return None
161 await self._db.execute(_INSERT_ROW, [channel.name, payload, now, now])
162 return 1
163 revision = int(current.rows[0]["revision"])
164 if expected_revision is not None and expected_revision != revision:
165 return None
166 new_revision = revision + 1
167 updated = await self._db.execute(
168 _UPDATE_ROW,
169 [payload, now, new_revision, channel.name, revision],
170 )
171 if updated.row_count == 0:
172 return None
173 return new_revision
175 async def delete(self, name: str, *, expected_revision: int) -> bool:
176 """Delete the row for *name* under compare-and-set.
178 Args:
179 name: The channel name to delete.
180 expected_revision: Revision the caller observed.
182 Returns:
183 ``True`` when a row was removed, ``False`` when the channel
184 is absent or the revision is stale.
185 """
186 await self._ensure_tables()
187 result = await self._db.execute(_DELETE_ROW, [name, expected_revision])
188 return result.row_count > 0