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.
| Engine | Location | Use Case |
|---|---|---|
memory | engines/storage/memory.py | Testing, ephemeral data |
sqlite | engines/storage/sqlite.py | Local 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.
| Engine | Style |
|---|---|
structlog | Structured, machine-readable |
loguru | Beautiful, 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/:
| Contract | Purpose |
|---|---|
SchemaEngine | validate(), serialize(), deserialize() |
StorageEngine | read(), write(), delete(), health() |
CacheEngine | get(), set(), delete(), exists(), health() |
SerializerEngine | encode(), decode() |
AdapterEngine | start(), 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
| Plugin | Package | Description |
|---|---|---|
| Base | evoid-base | Shared protocols (StorageEngine, CacheEngine, LoggerEngine) |
| DI | evoid-di | Dependency injection with fault tolerance |
| SQLite | evoid-sqlite | SQLite storage engine |
| Redis | evoid-redis | Redis cache with TTL |
| PostgreSQL | evoid-postgresql | PostgreSQL via SQLAlchemy + asyncpg |
| ScyllaDB | evoid-scylla | ScyllaDB/Cassandra storage |
| Smart Storage | evoid-smart-storage | Multi-DB routing, schema enforcement, multi-tenancy |
| Auth | evoid-auth | Bring your own auth provider |
| Tasks | evoid-tasks | Background tasks + structured logging |
| Dashboard | evoid-dashboard | Monitoring UI: service map, DB viewer, logs |
| Cluster | evoid-cluster | Multi-node clustering with failover |
| Godot | evoid-godot | Godot game integration adapter |
| Scheduler | evoid-scheduler | Priority-aware scheduler with adaptive concurrency |
| Transport | evoid-transport | Low-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")
Related
- Plugin Ecosystem: build your own plugins
- Plugin Collection: full plugin reference
- Plugin Standard: plugin packaging spec