ESC

Type to search...

Plugins and Engines

EVOID is infrastructure-agnostic. Every infrastructure component (database, cache, serializer, schema engine, DI) is a plugin. Plugins communicate through contracts, never concrete implementations.

Built-in Engines

EVOID ships with built-in engines. No plugins required.

Schema Engine

Validates and serializes data.

# Default: native (stdlib dataclasses + TypedDict)
# Located at: evoid/engines/schema/native.py

Storage Engine

Persists data.

EngineLocationUse Case
memoryengines/storage/memory.pyTesting, ephemeral data
sqliteengines/storage/sqlite.pyLocal persistence

Cache Engine

In-memory LRU cache with optional TTL.

from evoid.engines.cache import memory as cache

await cache.configure(max_size=1000)
await cache.set("user:123", user_data, ttl=300)
value = await cache.get("user:123")

Serializer Engine

JSON serialization with datetime support.

from evoid.engines.serializer import json_engine

encoded = json_engine.encode(data)
decoded = json_engine.decode(encoded)

DI Engine

Dependency injection: resolves dependencies by name.

from evoid.engines.di import native as di

di.register("db", database_factory)
db = di.resolve("db")

Logger Engine

Structured logging.

EngineStyle
structlogStructured, machine-readable
loguruBeautiful, colored output

Metrics Engine

Simple metrics collection.

Auth Engine

Authentication and authorization.

Plugin Engine

Meta-plugin for registering other plugins.

Contracts

Contracts define the interface that engines must satisfy. They live in evoid/contracts/:

ContractPurpose
SchemaEnginevalidate(), serialize(), deserialize()
StorageEngineread(), write(), delete(), health()
CacheEngineget(), set(), delete(), exists(), health()
SerializerEngineencode(), decode()
AdapterEnginestart(), stop(), handle()

Engines implement these contracts as classes that satisfy the Protocol.

Configuration

Engines are configured in evoid.toml:

[engines]
schema = "native"
storage = "memory"
cache = "memory"
serializer = "json"
di = "native"
logger = "loguru"
metrics = "simple"
auth = "simple"

Change infrastructure by changing config. Business logic stays untouched.

Writing Custom Engines

Implement the contract as functions:

# my_storage.py

async def read(key: str) -> Any:
    """Read data by key."""
    ...

async def write(key: str, value: Any) -> bool:
    """Write data by key."""
    ...

async def delete(key: str) -> bool:
    """Delete data by key."""
    ...

async def health() -> bool:
    """Check engine health."""
    return True

Register it:

from evoid.engines.plugin import register

register(
    name="postgres",
    type="storage",
    factory=lambda: postgres_storage,
    version="1.0.0",
    description="PostgreSQL storage engine",
)

Plugin Registry

The plugin system is a dict-based registry. Register a plugin, resolve it by name.

from evoid.engines.plugin import register, resolve

# Register a plugin
register(
    name="sqlite",
    type="storage",
    factory=sqlite_storage_factory,
    version="1.0.0",
    description="SQLite storage engine",
)

# Resolve it later
factory = resolve("sqlite", "storage")
engine = factory()

Dependency Map

EVOID maps engine names to Python packages for evo sync:

ENGINE_PACKAGES = {
    "sqlite": "aiosqlite",
    "redis": "redis.asyncio",
    "structlog": "structlog",
    "loguru": "loguru",
    # ...
}

evo sync reads evoid.toml, resolves package names, and installs them via uv.

Official Plugin Collection

Need more than built-in engines? The official plugin repository at EvolveBeyond/evoid-plugins has you covered.

Install

# Short names via evo CLI
evo install di            # → evoid-di
evo install redis         # → evoid-redis
evo install smart-storage # → evoid-smart-storage

# Or directly with uv
uv add evoid-di
uv add evoid-redis
uv add evoid-smart-storage

Available Plugins

PluginPackageDescription
Baseevoid-baseShared protocols (StorageEngine, CacheEngine, LoggerEngine)
DIevoid-diDependency injection with fault tolerance
SQLiteevoid-sqliteSQLite storage engine
Redisevoid-redisRedis cache with TTL
PostgreSQLevoid-postgresqlPostgreSQL via SQLAlchemy + asyncpg
ScyllaDBevoid-scyllaScyllaDB/Cassandra storage
Smart Storageevoid-smart-storageMulti-DB routing, schema enforcement, multi-tenancy
Authevoid-authBring your own auth provider
Tasksevoid-tasksBackground tasks + structured logging
Dashboardevoid-dashboardMonitoring UI: service map, DB viewer, logs
Clusterevoid-clusterMulti-node clustering with failover
Godotevoid-godotGodot game integration adapter
Schedulerevoid-schedulerPriority-aware scheduler with adaptive concurrency
Transportevoid-transportLow-latency UDP transport (Rust core)

DI Integration

All plugins integrate with evoid-di for automatic service discovery and fault tolerance:

from evoid_di import di

# Plugins register themselves with DI
di.register("storage.sqlite", SQLiteStorage, scope="singleton")
di.register("cache.redis", RedisCache, scope="singleton")

# Resolve with automatic fallback
storage = di.resolve_with_fallback("storage.postgresql")
# Tries: postgresql → sqlite → redis → cluster peers → None

Smart Storage Example

Routes data to different backends based on type, level, or user:

[engines]
storage = "smart_storage"

[engines.smart_storage.mapping]
credentials = "cache.redis"   # Sensitive data → Redis
session = "storage.sqlite"    # Sessions → SQLite
logs = "storage.postgresql"   # Logs → PostgreSQL

[engines.smart_storage.level_routing]
CRITICAL = "storage.postgresql"  # Critical intents → PostgreSQL
STANDARD = "storage.sqlite"      # Standard → SQLite

DI Engine Example

Three levels of complexity with fault tolerance:

from evoid_di import di

# Level 1: Simple
di.register("db", create_db)
db = di.resolve("db")

# Level 2: Scoped
di.register("db", create_db, scope="singleton")
di.register("session", create_session, scope="per_user")

# Level 3: Context-aware routing
di = DIEngine(rules_config=rules, implementations=impls)
instance = await di.resolve("notifier", ctx)

# Fault tolerance
di.set_fallback("storage.postgresql", ["storage.sqlite", "cache.redis"])
storage = di.resolve_with_fallback("storage.postgresql")

Cluster Integration

Cluster nodes share services automatically via DI:

from evoid_cluster import ClusterBridge

bridge = ClusterBridge(config)
await bridge.start()

# Cluster connects its registry to DI
# Remote services become available as fallbacks
storage = di.resolve("storage.postgresql")
# If not local, checks cluster peers automatically

Auth Example

Bring your own provider, no forced JWT:

from evoid_auth import register_provider

# Your auth logic
async def my_auth(token: str) -> dict:
    user = await db.find_by_token(token)
    return {"user": user.name, "role": user.role}

register_provider("my_auth", my_auth)

# Wire to pipeline
from evoid.core.extend import before
before("GET:/users", "authenticate")