Memory Manager

MemoryManager is a fixed-size in-memory list that automatically evicts the oldest entries when full. Nautica uses it internally for log memory, but you can use it in your own services too.

from nautica import MemoryManager

memory = MemoryManager(limit=50)  # holds up to 50 entries, default is 50

Adding and Recalling Entries

memory.Add("hello")
memory.Add({"user": "Martin", "action": "login"})

memory.Recall()       # returns all entries as a list
memory.Recall(10)     # returns last 10 entries
memory.Forget()       # clears all entries

Recall() always returns a deep copy, so modifying the result won't affect stored entries.

Validating Entries

Use setValidator() to enforce a schema on entries before they're added. Raises TypeError if an entry fails validation:

memory.setValidator(lambda entry: isinstance(entry, dict))

memory.Add({"user": "Martin"})   # ok
memory.Add("hello")              # raises TypeError

Mirrors

A mirror is a linked MemoryManager that receives all new entries added to the original. Existing entries can optionally be copied over:

mirror = memory.CreateMirror(copy_content=True)  # copy_content defaults to True

memory.Add("new entry")   # mirror also receives this
mirror.Recall()           # includes "new entry"

Mirrors are independent, you can Forget() a mirror without affecting the original.

Full Example

A service that stores the last 100 requests and exposes them:

from nautica import Service, Logger
from nautica.manager import MemoryManager

class RequestHistory(Service):
    def __init__(self):
        super().__init__()
        self.history = MemoryManager(limit=100)
        self.history.setValidator(lambda e: isinstance(e, dict))

    def onStart(self, registry):
        Logger.ok("RequestHistory ready")

    def record(self, ip: str, path: str, status: int):
        self.history.Add({
            "ip": ip,
            "path": path,
            "status": status
        })

    def recent(self, count: int = 10):
        return self.history.Recall(count)

Service.Export(RequestHistory)

Then from another service:

def onStart(self, registry):
    self.history = registry.Get("RequestHistory")

def after_request(self, ctx):
    self.history.record(ctx.clientIp, str(ctx.url), ctx.response.status_code)