Writing a service¶

The Quick Start shows the smallest service. This guide adds configuration, data and readiness. Its example, notes, keeps notes in a database.

Messages and the interface¶

Put them in a module of their own, which other services can import without pulling in the implementation:

from typing import Protocol

import msgspec


class NewNote(msgspec.Struct, frozen=True):
    text: str


class NoteRef(msgspec.Struct, frozen=True):
    id: int


class Note(msgspec.Struct, frozen=True):
    id: int
    text: str
    words: int | None


class NotesAPI(Protocol):
    async def add(self, note: NewNote) -> NoteRef: ...

    async def get(self, ref: NoteRef) -> Note: ...

Each operation takes one message and returns one, or None. Use a tuple where you would use a list, since a list can change. An operation with no input takes an empty struct.

Configuration¶

The class attribute Config names a struct filled from the layout's [services.notes], here NotesConfig. A parameter takes the struct's own name as its type, since mypy refuses a class attribute there. With forbid_unknown_fields, a mistyped setting stops the process, with a message that names the section and the field:

class NotesConfig(msgspec.Struct, forbid_unknown_fields=True):
    database: str

A value that reads the environment arrives as a string, and fills a number or a boolean all the same.

Your own data¶

A service owns its tables and creates them in migrate, which must be safe to run again. It holds its engine in run, and answers for its database in check:

from collections.abc import Callable

from sqlalchemy import Column, Integer, MetaData, String, Table, select

from myxo import NotFound, Service, Unavailable
from myxo.db import DB_ERRORS, make_engine

metadata = MetaData()
NOTES = Table(
    "notes",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("text", String, nullable=False),
    Column("words", Integer),
)


class Notes(Service):
    name = "notes"
    api = NotesAPI
    Config = NotesConfig

    @classmethod
    async def migrate(cls, config: NotesConfig) -> None:
        engine = make_engine(config.database)
        try:
            async with engine.begin() as conn:
                await conn.run_sync(metadata.create_all)
        finally:
            await engine.dispose()

    async def run(self, ready: Callable[[], None]) -> None:
        self.engine = make_engine(self.config.database)
        try:
            await super().run(ready)
        finally:
            await self.engine.dispose()

    async def check(self) -> None:
        try:
            async with self.engine.connect() as conn:
                await conn.execute(select(1))
        except DB_ERRORS as exc:
            raise Unavailable(f"database: {exc}") from exc

    async def get(self, ref: NoteRef) -> Note:
        query = select(NOTES).where(NOTES.c.id == ref.id)
        async with self.engine.connect() as conn:
            row = (await conn.execute(query)).first()
        if row is None:
            raise NotFound(f"note {ref.id}")
        return Note(*row)

super().run(ready) calls ready() and waits to be cancelled, which suits a service that only answers calls. A service with a loop of its own calls ready() once it can take calls, and serves until cancelled. make_engine takes a postgresql:// or sqlite:// URL, so the same code runs on both.

A service raises NotFound and Invalid for answers the caller should handle. Let anything else propagate: the caller sees a RemoteError, while the log has the traceback.

Declaring it¶

[project.entry-points."myxo.services"]
notes = "hello.notes:Notes"

[project.entry-points."myxo.apis"]
notes = "hello.api:NotesAPI"

The entry point's name must match the service's name. Install the package again after changing these tables.

Stages¶

Within a process, services start by stage, each once the previous stage is ready. A service that calls another in its own process at startup takes a higher stage. Processes start independently, so a call to another process may raise Unavailable at first. Wrap it in retry where the operation allows.

An optional dependency¶

runtime.provides(API) says whether a service in the layout implements an interface, for a feature that a layout may leave out:

self.reports = runtime.client(ReportsAPI) if runtime.provides(ReportsAPI) else None

Finding code that depends on its placement¶

myxo check catches mutable messages, and imports that bypass an interface. Two process settings find the rest, without moving anything:

[processes.all]
services = ["web", "notes"]
url = "unix:run/all.sock"
loopback = true        # calls go through the listener, as if remote

[processes.all]
services = ["web", "notes"]
copy_messages = true   # calls stay in the process, with their messages copied

Run your tests against both, as well as against a layout with several processes.