nfo

nfo Function Reference

This document provides a comprehensive reference of all functions in the nfo project, organized by module and category.

Core API

Decorators (nfo.decorators)

@log_call

Automatically logs function calls with arguments, return values, exceptions, and duration.

@log_call
def my_function(arg1, arg2):
    return arg1 + arg2

Parameters:

• func (Optional[Callable]) - Function to decorate
• level (str) - Log level (default: “DEBUG”)
• logger (Optional[Logger]) - Custom logger instance
• max_repr_length (Optional[int]) - Truncate long representations

Returns: Decorated function

@catch

Like @log_call but suppresses exceptions and returns a default value.

@catch(default=None)
def risky_function():
    return 1 / 0  # Returns None instead of raising

Parameters:

• func (Optional[Callable]) - Function to decorate
• default (Any) - Value to return on exception (default: None)
• level (str) - Log level (default: “ERROR”)
• logger (Optional[Logger]) - Custom logger instance
• max_repr_length (Optional[int]) - Truncate long representations

Returns: Decorated function

@logged

Class decorator that automatically wraps all public methods with @log_call.

@logged
class MyService:
    def method1(self): pass  # Will be logged
    def _private(self): pass  # Won't be logged

Parameters:

• cls (Optional[Type]) - Class to decorate
• level (str) - Log level for all methods
• logger (Optional[Logger]) - Custom logger instance
• max_repr_length (Optional[int]) - Truncate long representations

Returns: Decorated class

@skip

Mark a public method to be excluded from @logged auto-wrapping.

@logged
class MyService:
    @skip
    def health_check(self): pass  # Excluded from logging

Auto-Logging (nfo.auto)

auto_log(*modules, **kwargs)

Automatically wrap all functions in specified modules with logging decorators.

import mymodule
auto_log(mymodule, level="INFO", catch_exceptions=True)

Parameters:

• *modules - Module objects to instrument
• level (str) - Log level for all functions
• catch_exceptions (bool) - Use @catch instead of @log_call
• default (Any) - Default value for @catch
• include_private (bool) - Also wrap private functions
• max_repr_length (Optional[int]) - Truncate long representations

Returns: Number of functions patched

auto_log_by_name(*module_names, **kwargs)

Like auto_log() but accepts module name strings.

auto_log_by_name("myapp.api", "myapp.core", level="INFO")

Parameters:

• *module_names - Module names to instrument
• Same kwargs as auto_log()

Returns: Number of functions patched

Configuration (nfo.configure)

configure(**kwargs)

One-liner project setup with automatic environment variable support.

configure(
    sinks=["sqlite:logs.db", "csv:logs.csv"],
    level="INFO",
    modules=["myapp.api", "myapp.core"]
)

Parameters:

• name (str) - Logger name (default: “nfo”)
• level (str) - Log level (default: “DEBUG”)
• sinks (List[Union[str, Sink]]) - Sink specifications or instances
• modules (List[str]) - Stdlib modules to bridge
• propagate_stdlib (bool) - Forward to stdlib loggers
• environment (str) - Environment tag
• version (str) - Application version
• llm_model (str) - LLM model for analysis
• detect_injection (bool) - Enable prompt injection detection
• force (bool) - Re-configure even if already configured

Returns: Configured Logger instance

Sinks

Built-in Sinks (nfo.sinks)

SQLiteSink(db_path, table)

Persist logs to SQLite database for querying.

sink = SQLiteSink("logs.db", table="function_calls")

Parameters:

• db_path (Any) - Database file path
• table (str) - Table name (default: “logs”)

CSVSink(file_path)

Append logs to CSV file.

sink = CSVSink("logs.csv")

Parameters:

• file_path (Any) - CSV file path

MarkdownSink(file_path)

Write human-readable Markdown logs.

sink = MarkdownSink("logs.md")

Parameters:

• file_path (Any) - Markdown file path

JSONSink(file_path, pretty)

Write structured JSON Lines output.

sink = JSONSink("logs.jsonl", pretty=False)

Parameters:

• file_path (Any) - JSON file path
• pretty (bool) - Pretty-print JSON (default: False)

Advanced Sinks

PrometheusSink (nfo.prometheus)

Export function call metrics to Prometheus.

sink = PrometheusSink(
    delegate=SQLiteSink("logs.db"),
    port=9090
)

Parameters:

• delegate (Optional[Sink]) - Downstream sink
• port (int) - Metrics server port
• prefix (str) - Metric name prefix

Methods:

• get_metrics() - Return current metrics in Prometheus format

WebhookSink (nfo.webhook)

Send HTTP alerts to Slack, Discord, or Teams.

sink = WebhookSink(
    url="https://hooks.slack.com/...",
    levels=["ERROR"],
    format="slack"
)

Parameters:

• url (str) - Webhook URL
• delegate (Optional[Sink]) - Downstream sink
• levels (List[str]) - Log levels to alert on
• format (str) - Payload format: “slack”, “discord”, “teams”, “raw”

LLMSink (nfo.llm)

AI-powered log analysis via litellm.

sink = LLMSink(
    model="gpt-4o-mini",
    delegate=SQLiteSink("logs.db"),
    detect_injection=True
)

Parameters:

• model (str) - LLM model name
• delegate (Optional[Sink]) - Downstream sink
• detect_injection (bool) - Scan for prompt injection
• analyze_levels (List[str]) - Levels to analyze (default: [“ERROR”])

Environment Sinks (nfo.env)

EnvTagger

Auto-tag logs with environment, trace ID, and version.

sink = EnvTagger(
    SQLiteSink("logs.db"),
    environment="prod",
    trace_id="abc123"
)

Parameters:

• delegate (Sink) - Downstream sink
• environment (Optional[str]) - Environment tag
• trace_id (Optional[str]) - Trace ID
• version (Optional[str]) - Application version

DynamicRouter

Route logs to different sinks based on rules.

router = DynamicRouter([
    (lambda e: e.level == "ERROR", SQLiteSink("errors.db")),
    (lambda e: e.environment == "prod", PrometheusSink())
])

Parameters:

• rules (List[tuple]) - (predicate, sink) pairs
• default (Optional[Sink]) - Default sink

DiffTracker

Detect when function output changes between versions.

sink = DiffTracker(SQLiteSink("logs.db"))

Parameters:

• delegate (Sink) - Downstream sink

Utilities

LLM Functions (nfo.llm)

detect_prompt_injection(text)

Scan text for common prompt injection patterns.

result = detect_prompt_injection("ignore previous instructions")

Parameters:

• text (str) - Text to scan

Returns: Optional[str] - Injection type if detected

scan_entry_for_injection(entry)

Scan a LogEntry’s arguments for prompt injection.

injection = scan_entry_for_injection(log_entry)

Parameters:

• entry (LogEntry) - Log entry to scan

Returns: Optional[str] - Injection type if detected

Environment Functions (nfo.env)

generate_trace_id()

Generate a new trace ID for distributed tracing.

trace_id = generate_trace_id()

Returns: str - UUID-based trace ID

Model Functions (nfo.models)

safe_repr(value, max_length)

Safe string representation with truncation.

repr_str = safe_repr(large_object, max_length=512)

Parameters:

• value (Any) - Value to represent
• max_length (Optional[int]) - Maximum length

Returns: str - Safe representation

CLI Interface

Main Commands (nfo.__main__)

nfo run -- <command>

Run any command with automatic logging.

nfo run -- python script.py
nfo run -- bash deploy.sh prod

nfo logs [options]

Query logs from SQLite database.

nfo logs --errors --last 24h
nfo logs --function deploy -n 50

nfo serve [--port]

Start centralized HTTP logging service.

nfo serve --port 8080

nfo version

Print nfo version.

nfo version

Data Models

LogEntry (nfo.models)

Core data structure representing a function call log.

Fields:

• timestamp (datetime) - UTC timestamp
• level (str) - Log level (DEBUG/ERROR)
• function_name (str) - Qualified function name
• module (str) - Python module
• args (tuple) - Positional arguments
• kwargs (dict) - Keyword arguments
• arg_types (tuple) - Argument type names
• kwarg_types (dict) - Keyword argument type names
• return_value (Any) - Function return value
• return_type (str) - Return value type
• exception (Optional[str]) - Exception message
• exception_type (Optional[str]) - Exception class name
• traceback (Optional[str]) - Full traceback
• duration_ms (float) - Execution time in milliseconds
• environment (Optional[str]) - Environment tag
• trace_id (Optional[str]) - Trace ID
• version (Optional[str]) - Application version
• llm_analysis (Optional[str]) - LLM analysis result
• extra (dict) - Additional metadata

Methods:

• now() - Create timestamp
• args_repr() - Get truncated args representation
• kwargs_repr() - Get truncated kwargs representation
• return_value_repr() - Get truncated return value representation
• as_dict() - Convert to flat dictionary

Logger (nfo.logger)

Central dispatcher for log entries.

Methods:

• add_sink(sink) - Register a sink
• remove_sink(sink) - Remove a sink
• emit(entry) - Send entry to all sinks
• close() - Close all sinks

Sink Interface

All sinks implement the same interface:

class Sink:
    def write(self, entry: LogEntry) -> None:
        """Write a log entry."""
        pass
    
    def close(self) -> None:
        """Close the sink and release resources."""
        pass

Environment Variables

nfo automatically reads these environment variables:

• NFO_LEVEL - Default log level
• NFO_SINKS - Comma-separated sink specifications
• NFO_ENV - Environment tag
• NFO_VERSION - Application version
• NFO_LLM_MODEL - LLM model name
• OPENAI_API_KEY - OpenAI API key (for LLM features)
• NFO_WEBHOOK_URL - Webhook URL for alerts
• NFO_PROMETHEUS_PORT - Prometheus metrics port
• NFO_LOG_DIR - Directory for log files
• NFO_PORT - HTTP service port

Sink Specifications

String format for configure() and CLI:

sqlite:path/to/db.db
csv:path/to/file.csv
md:path/to/file.md
json:path/to/file.jsonl
prometheus:9090

Error Handling

Common Exceptions

• ImportError - Optional dependencies not installed
• sqlite3.Error - Database connection issues
• FileNotFoundError - Invalid file paths
• ConnectionError - Webhook/network issues
• ValueError - Invalid configuration

Best Practices

1. Always wrap risky operations with @catch
2. Use appropriate log levels (DEBUG for success, ERROR for failures)
3. Configure multiple sinks for redundancy
4. Set max_repr_length for functions with large arguments
5. Use environment variables for deployment-specific configuration

This site is open source. Improve this page.