Services

class walnats.Services

A description of services that your system has.

It doesn’t serve any direct practical purpose. Walnats doesn’t care what services you have or where events are emitted from. The purpose of Services is to describe your system for producing architectural overview in a form of diagrams and specs.

Currently, can produce the following specifications:

• Event storming diagram as d2 definition.

• AsyncAPI spec either for each service or all services combined.

services = walnats.Services()

add(name: str, *, version: str = '0.0.0', emits: walnats.Events | None = None, defines: walnats.Actors | None = None)

Add a service.

The service is described by the name it has, events it emits, and actors it defines.

services.add(
    name='users',
    emits=walnats.Events(USER_CREATED, USER_UPDATED),
)
services.add(
    name='notifications',
    defines=walnats.Actors(SEND_EMAIL),
)

get_d2(direction: Literal['left', 'right', 'down', 'up'] = 'right', colors: bool = True) → str

Generate a d2 diagram definition.

D2 is a language for generating diagrams, similar to PlantUML or GraphViz (see comparison). This method produces a diagram definition that you can pipe into d2 CLI to generate an actual image.

Parameters:

• direction – direction of drawing the diagram, in the order as services are defined. See d2 docs.

• colors –

  whatever to explicitly specify colors. If enabled (default), event storming colors will be used for elements. If disabled, the colors are controlled by the d2 theme.

In the script (arch.py):

print(services.get_d2())

And then run in shell:

python3 arch.py | d2 - > arch.svg

get_async_apis() → list[dict[str, Any]]

Generate an AsyncAPI schema for every service.

Returns a list of valid AsyncAPI schemas for each service.

specs = services.get_async_api()
for spec in specs:
    service_name = spec['info']['title']
    with open(f'{service_name}.json') as f:
        json.dump(spec, f)

get_async_api() → dict[str, Any]

Generate a combined AsyncAPI schema for all services.

It’s simply a merge of all events that all services emit plus a few required field on top of that to produce a valid schema. It’s generally a good idea to add more info into it before saving the result into a JSON or YAML file.

spec = services.get_async_api()
spec['info']['description'] = 'Here comes the sun'
with open('spec.json') as f:
    json.dump(spec, f)

D2 example

Here is a big working example of generating diagrams and its output.

"""Example of drawing d2 diagrams of walnats-based architecture.

Usage:
    python3 ./examples/diagram.py | d2 - > arch.svg
    chromium ./arch.svg
"""

from __future__ import annotations

from dataclasses import dataclass

import walnats


@dataclass
class User:
    id: int
    name: str


class Email:
    id: int
    receiver: str


def noop(_: object) -> None:
    pass


USER_CREATED = walnats.Event('user-created', User)
USER_UPDATED = walnats.Event('user-updated', User)
EMAIL_SENT = walnats.Event('email-sent', Email)

services = walnats.Services()
services.add(
    name='users',
    emits=walnats.Events(USER_CREATED, USER_UPDATED),
)
services.add(
    name='notifications',
    emits=walnats.Events(EMAIL_SENT),
    defines=walnats.Actors(
        walnats.Actor('send-email', USER_CREATED, noop),
        walnats.Actor('send-email', USER_UPDATED, noop),
        walnats.Actor('send-sms', USER_CREATED, noop),
        walnats.Actor('send-sms', USER_UPDATED, noop),
    ),
)
services.add(
    name='audit-log',
    defines=walnats.Actors(
        walnats.Actor('record', USER_CREATED, noop),
        walnats.Actor('record', EMAIL_SENT, noop),
    ),
)
print(services.get_d2())

The produced diagram: