Skip to content

Extension Points

NLT provides two Protocol-based extension points that let you integrate the engine with your own data sources and template storage. Both are optional -- NLT works out of the box with just the data you pass to the constructor.

EventStore Protocol

When to Implement

Implement EventStore when you need to look up events that are not available in the additional_events dictionary at render time. This is useful when:

  • Your event history is large and pre-loading all events would be impractical
  • You want to fetch events lazily from a database or API
  • The template references event types that were not known at flow trigger time

Protocol Definition

python
from typing import Any, Protocol
 
class EventStore(Protocol):
async def get_latest(
self,
contact_id: str,
event_name: str,
) -> dict[str, Any] | None: ...

The method should return an event dict with an attributes key, or None if no matching event exists.

Note

The EventStore protocol is currently reserved for future async integration. The synchronous rendering pipeline does not invoke async methods. The protocol is defined for forward compatibility -- implement it now so your integration is ready when async rendering is available.

Example: Database EventStore

python
from pyrx_nlt import NLTRenderer
from pyrx_nlt.renderer import EventStore
from typing import Any
 
 
class DatabaseEventStore:
"""Look up events from a database."""
 
def __init__(self, db_session, contact_id: str):
self.db = db_session
self.contact_id = contact_id
 
async def get_latest(
self, contact_id: str, event_name: str
) -> dict[str, Any] | None:
row = await self.db.execute(
"""
SELECT attributes FROM events
WHERE contact_id = :cid AND event_name = :name
ORDER BY created_at DESC LIMIT 1
""",
{"cid": contact_id, "name": event_name},
)
result = row.first()
if result:
return {"attributes": result.attributes}
return None
 
 
# Usage
store = DatabaseEventStore(db_session, contact_id="abc-123")
renderer = NLTRenderer(
contact=contact_data,
trigger_event=trigger_data,
event_store=store,
)

TemplateLoader Protocol

When to Implement

Implement TemplateLoader to enable {extends} template inheritance. The loader is responsible for finding and returning the raw template string for a given template name.

Without a TemplateLoader, {extends} declarations are silently ignored and the child template is rendered as-is.

Protocol Definition

python
from typing import Protocol
 
class TemplateLoader(Protocol):
def load(self, template_name: str) -> str | None: ...

Returns the raw template string, or None if the template is not found.

Example: Filesystem Loader

python
from pyrx_nlt import NLTRenderer
from pyrx_nlt.renderer import TemplateLoader
import os
 
 
class FileTemplateLoader:
"""Load templates from the filesystem."""
 
def __init__(self, template_dir: str):
self.template_dir = template_dir
 
def load(self, template_name: str) -> str | None:
path = os.path.join(self.template_dir, template_name)
if os.path.exists(path):
with open(path) as f:
return f.read()
return None
 
 
# Usage
loader = FileTemplateLoader("/path/to/templates")
renderer = NLTRenderer(
contact=contact_data,
template_loader=loader,
)
 
result = renderer.render(
subject_template="Welcome",
body_template='`{extends "base.html"}`\n`{block content}`Hello!`{end block}`',
)

Example: Database Loader

python
class DatabaseTemplateLoader:
"""Load templates from a database."""
 
def __init__(self, db_session, tenant_id: str):
self.db = db_session
self.tenant_id = tenant_id
 
def load(self, template_name: str) -> str | None:
row = self.db.execute(
"SELECT body FROM templates WHERE tenant_id = %s AND name = %s",
(self.tenant_id, template_name),
).fetchone()
return row.body if row else None
Tip

For production use, wrap your TemplateLoader with an in-memory cache. Base templates rarely change, and caching avoids repeated database or filesystem lookups during high-volume rendering.

Integration Patterns

Minimal Integration

The simplest integration passes all data inline with no protocols:

python
renderer = NLTRenderer(
contact=contact_data,
trigger_event=trigger_data,
additional_events={"payment": payment_event},
)
result = renderer.render(subject_template=subject, body_template=body)

Full Integration

A production integration typically provides both protocols:

python
renderer = NLTRenderer(
contact=contact_data,
trigger_event=trigger_data,
additional_events=preloaded_events,
event_store=DatabaseEventStore(db, contact_id),
template_loader=CachedDatabaseTemplateLoader(db, tenant_id),
)

One Renderer Per Email

Create a new NLTRenderer instance for each email you render. The renderer tracks suppression state and loop/macro context internally, so reusing an instance across multiple renders can produce incorrect results.

python
# Correct: one renderer per email
for contact in contacts:
renderer = NLTRenderer(contact=contact, trigger_event=event)
result = renderer.render(subject_template=subject, body_template=body)
if not result.suppressed:
send_email(contact, result)

Next Steps