Skip to content

How To Use an Emitter

Jump to bottom
Fernando Koch edited this page Mar 29, 2026 · 3 revisions

How To Use An Emitter

Purpose

Use Emitter when you want GraphK to behave like a request/response component instead of a manually managed runner.

Emitter wraps a Runner and gives you a simple interface:

  • send an event with request(...)
  • read the output with response()

When Should You Use It?

Use Emitter when:

  • you want to expose a pipeline as a callable processing unit
  • you want one active session retained across requests unless reset
  • you want a public interface that feels simpler than managing Runner directly

Prefer Runner when you need lower-level execution control.

Core Methods

request(event=None)

Purpose:

  • ensure a session exists
  • apply the request payload to the session
  • execute the wrapped pipeline
  • return the resulting session

Behavior:

  • stores incoming payload under request
  • stores outgoing payload under response

response(default=None)

Purpose:

  • read the current response payload from the active session

reset(session=None)

Purpose:

  • replace the active session

Use this when:

  • you want a fresh state boundary
  • you want to preload a session before issuing a request

enable_log(level=graphk.Logger.DEBUG)

Purpose:

  • enable runner logging through the wrapped runtime

Smallest Useful Example

import graphk


pipeline = graphk.SequencePipe(nodes=[graphk.EchoNode()])
emitter = graphk.Emitter(pipeline)

emitter.request({"value": 5})

print(emitter.response())

Expected idea:

The request payload becomes session input, the node runs, and the response can be read directly from the emitter.

Example With Logging

import graphk


pipeline = graphk.SequencePipe(nodes=[graphk.EchoNode()])
emitter = graphk.Emitter(pipeline)

emitter.enable_log(graphk.Logger.DEBUG)
session = emitter.request({"value": 5})

print(session.to_dict())
print(emitter.response())

Use this form when you want to inspect lifecycle behavior.

Example With A Preloaded Session

import graphk


base = graphk.Session(prefix="base", counter=10)

pipeline = graphk.SequencePipe(
    nodes=[
        graphk.ApplyNode(
            "PrefixEcho",
            fn=lambda session: f"{session.get('prefix', 'ok')}:{session.get('value', 0)}",
            response_key="response",
        )
    ]
)

emitter = graphk.Emitter(pipeline, session=base)
session = emitter.request({"value": 3})

print(session is base)
print(emitter.response())

This is useful when the emitter should retain or start from an existing state baseline.

How Request And Response Are Stored

By default, the emitter uses:

  • request key: request
  • response key: response

Nodes in the pipeline should write their final payload into response if they are intended to feed the emitter response directly.

Good built-in options:

  • EchoNode
  • ApplyNode
  • IncrementNode(..., response=True)
  • ConcatNode(..., response=True)
  • RouteNode

Relationship To Runner

Internally, Emitter is a convenience wrapper over Runner.

Use Emitter when:

  • you want the package to feel like a reusable service
  • you want one method call per request

Use Runner when:

  • you need direct start/step/run control
  • you want to inspect execution at a lower level

Common Mistakes

  • Do not assume a response appears automatically unless nodes write to the response key.
  • Do not forget to reset the emitter if you need a clean session between calls.
  • Do not reimplement request/session plumbing when Emitter already matches the use case.

Learn More

Recommended references:

Clone this wiki locally