Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -36,38 +36,21 @@
from __future__ import annotations

import logging
from pathlib import Path
from urllib.parse import urlsplit

from omnimarket.projection.postgres_sync_database import PostgresSyncProjectionAdapter
from omnimarket.projection.protocol_database import DatabaseAdapter
from omnimarket.projection.runner import projection_runtime_binding_from_overlay_env
from omnimarket.projection.sqlite_database import (
SQLITE_SCHEMES,
SqliteDatabaseAdapter,
default_evidence_db_path,
sqlite_path_from_dsn,
)

logger = logging.getLogger(__name__)

_POSTGRES_SCHEMES = frozenset({"postgres", "postgresql"})
_SQLITE_SCHEMES = frozenset({"sqlite", "file"})


def _sqlite_path_from_dsn(dsn: str) -> Path:
"""Extract a filesystem path from a ``sqlite:``/``file:`` DSN or bare path.

Follows the SQLAlchemy-style slash convention: ``sqlite:///rel/path`` is a
relative path (``rel/path``) and ``sqlite:////abs/path`` is absolute
(``/abs/path``) — i.e. exactly one leading slash from the URL path component
is the scheme separator and is stripped.
"""
split = urlsplit(dsn)
if split.scheme in _SQLITE_SCHEMES:
raw = split.path or split.netloc
if raw.startswith("/"):
raw = raw[1:]
return Path(raw)
return Path(dsn)


def _adapter_for_dsn(
Expand All @@ -82,8 +65,8 @@ def _adapter_for_dsn(
scheme = urlsplit(dsn).scheme.lower()
if scheme in _POSTGRES_SCHEMES:
return PostgresSyncProjectionAdapter(dsn, schema=postgres_schema)
if scheme in _SQLITE_SCHEMES or not scheme:
return SqliteDatabaseAdapter(_sqlite_path_from_dsn(dsn))
if scheme in SQLITE_SCHEMES or not scheme:
return SqliteDatabaseAdapter(sqlite_path_from_dsn(dsn))
raise ValueError(
f"unsupported delegation evidence database_url scheme {scheme!r}; "
"use a postgres[ql]:// or sqlite:/file: DSN"
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,8 @@ description: >
Read-only EFFECT node that answers a typed read of one contract-declared
projection exposure (a node contract's projection_api block) from the table
its writer materializes, through the runtime's own projection database
binding. Pages with the same function as the standalone projection API's
binding, which names Postgres or the local SQLite store (local MVP mode 1,
OMN-20329). Pages with the same function as the standalone projection API's
GET /projection/{topic}, so both read paths answer a request identically.
Reachable through the runtime's Pattern B /skill edge (OMN-20159).
node_type: effect
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,9 @@
TenantConflictError,
resolve_request_tenant,
)
from omnimarket.nodes.node_projection_read_effect.ports.sqlite_row_source import (
SqliteTableRowSource,
)
from omnimarket.projection.discovery import build_projection_topic_map
from omnimarket.projection.models import ProjectionTableConfig
from omnimarket.projection.read_page import (
Expand Down Expand Up @@ -81,7 +84,7 @@ def __init__(
) -> None:
self._topic_map = topic_map
self._row_source = row_source
self._owned_source: TableRowSource | None = None
self._owned_source: TableRowSource | SqliteTableRowSource | None = None

def _topics(self) -> dict[str, ProjectionTableConfig]:
if self._topic_map is None:
Expand All @@ -98,7 +101,8 @@ def _source(self) -> ProtocolProjectionRowSource:
async def close(self) -> None:
"""Close the pools of a row source this handler opened itself."""
owned, self._owned_source = self._owned_source, None
if owned is not None:
# A SQLite source holds no connection between reads.
if isinstance(owned, TableRowSource):
await owned.close()

async def handle(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,11 @@
-- the same resolver ``node_delegate_skill_orchestrator`` uses, so this module
reads no environment variable itself.

The binding may name Postgres, where the deployed writers materialize their
tables, or the local SQLite store a local runtime's writers fill (local MVP
mode 1, OMN-20329). Either way the read goes to the database the writers were
bound to.

Unlike that node there is no fallback. A runtime with no binding has no
projection tables to read, and answering from some other database would
serve rows the writers never wrote, so the read is refused by name.
Expand All @@ -18,19 +23,28 @@

from urllib.parse import urlsplit

from omnimarket.nodes.node_projection_read_effect.ports.sqlite_row_source import (
SqliteTableRowSource,
)
from omnimarket.projection.runner import projection_runtime_binding_from_overlay_env
from omnimarket.projection.table_reader import ProjectionReadError, TableRowSource
from omnimarket.projection.sqlite_database import (
SQLITE_SCHEMES,
sqlite_path_from_dsn,
)
from omnimarket.projection.table_reader import (
ProjectionReadError,
TableRowSource,
)

_POSTGRES_SCHEMES = frozenset({"postgres", "postgresql"})


def resolve_projection_read_source() -> TableRowSource:
def resolve_projection_read_source() -> TableRowSource | SqliteTableRowSource:
"""A row source over the database the runtime binding names.

Raises :class:`ProjectionReadError` ``projection_binding_unconfigured`` when
the runtime carries no binding, and ``projection_binding_unsupported`` when
the bound database is not Postgres (the writer tables this node reads are
materialized in Postgres).
the bound database is neither Postgres nor a SQLite file.
"""
binding = projection_runtime_binding_from_overlay_env()
if binding is None:
Expand All @@ -46,12 +60,16 @@ def resolve_projection_read_source() -> TableRowSource:
"projection_binding_unconfigured",
"the projection runtime binding's database reference did not resolve",
) from exc
if urlsplit(database_url).scheme.lower() not in _POSTGRES_SCHEMES:
raise ProjectionReadError(
"projection_binding_unsupported",
"the projection runtime binding names a database that is not Postgres",
)
return TableRowSource.for_database_url(database_url)
scheme = urlsplit(database_url).scheme.lower()
if scheme in _POSTGRES_SCHEMES:
return TableRowSource.for_database_url(database_url)
if scheme in SQLITE_SCHEMES:
return SqliteTableRowSource(sqlite_path_from_dsn(database_url))
raise ProjectionReadError(
"projection_binding_unsupported",
"the projection runtime binding names a database that is neither "
"Postgres nor a SQLite file",
)


__all__ = ["resolve_projection_read_source"]
Loading
Loading