Official plugin collection for the EVOID Intent-Oriented Programming runtime.
Each plugin is an independent Python package on PyPI. Install only what you need.
uv add evoid-sqlite
uv add evoid-redis
uv add evoid-di
uv add evoid-smart-storageOr via EVOID CLI:
evo install di
evo install redis
evo install smart-storageAll plugins integrate with evoid-di for dependency injection, service discovery, and fault tolerance.
┌─────────────────────────────────────────────────────┐
│ Your App │
└──────────────────────┬──────────────────────────────┘
│
┌──────────────────────▼──────────────────────────────┐
│ evoid-di (Global DI) │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ register │ │ resolve │ │ fallback │ │
│ │ health │ │ cluster │ │ load-balance│ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
└──────────────────────┬──────────────────────────────┘
│
┌──────────────────────▼──────────────────────────────┐
│ Plugins │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ sqlite │ │ postgres │ │ redis │ │
│ │ storage │ │ storage │ │ cache │ │
│ └────┬─────┘ └────┬─────┘ └────┬─────┘ │
│ └─────────────┼───────────┘ │
│ ▼ │
│ evoid-smart-storage │
│ (routes by data type/level) │
└─────────────────────────────────────────────────────┘
Plugins that use DI get automatic service discovery, fault tolerance, and cluster integration.
from evoid_di import di
def register_handlers(config=None):
# 1. Register with DI
di.register("storage.mydb", lambda: MyStorage(config), scope="singleton")
# 2. Define fallback chain
di.set_fallback("storage.mydb", ["storage.sqlite", "cache.redis"])
# 3. Optional: health check
di.set_health_check("storage.mydb", lambda: my_storage.ping())
# 4. Wire to EVOID intents
from evoid.core import register as register_intent, register_processor
async def handle_read(ctx):
storage = di.resolve("storage.mydb") # resolve via DI
return await storage.read(ctx.intent.metadata.get("key"))
register_processor("storage.read", handle_read)Benefits:
- Automatic fallback when service fails
- Load balancing across cluster nodes
- Health checking and auto-reconnect
- Smart-storage integration
For quick prototypes or standalone use.
from evoid_sqlite import create_storage
storage = create_storage("app.db")
await storage.write("user:1", {"name": "Alice"})
user = await storage.read("user:1")from evoid_di import di
# Register services
di.register("storage.sqlite", lambda: SQLiteStorage("app.db"), scope="singleton")
di.register("cache.redis", lambda: RedisCache("redis://localhost"), scope="singleton")
# Batch registration
di.register_many({
"db": create_db,
"cache": create_cache,
})# Simple resolve
storage = di.resolve("storage.sqlite")
# Safe resolve (returns None if not found)
storage = di.resolve_or_none("nonexistent")
# Resolve with automatic fallback
storage = di.resolve_with_fallback("storage.postgresql")
# Resolve first available from list
cache = di.resolve_any("cache.redis", "cache.memory", "storage.sqlite")# Define fallback chain
di.set_fallback("storage.postgresql", ["storage.sqlite", "cache.redis"])
# Health checking
di.set_health_check("cache.redis", lambda: redis.ping())
# Mark services unhealthy
di.mark_unhealthy("cache.redis")
# Auto-fallback on failure
storage = di.resolve_with_fallback("storage.postgresql")
# Tries: postgresql → sqlite → redis → None (no crash)# Connect cluster registry
from evoid_cluster import ServiceRegistry
di.set_cluster_registry(cluster._registry)
# Remote resolution (automatic)
storage = di.resolve("storage.postgresql")
# If not local, checks cluster peers automatically# Singleton (default) - one instance
di.register("db", create_db, scope="singleton")
# Transient - new instance each time
di.register("request", create_request, scope="transient")
# Per-user - isolated per user
di.register("session", create_session, scope="per_user")Routes data to different backends based on type, level, or metadata.
from evoid_smart_storage import register_handlers
register_handlers(config={
"mapping": {
"credentials": "cache.redis", # sensitive data → Redis
"session": "storage.sqlite", # session → SQLite
"logs": "storage.postgresql", # logs → PostgreSQL
},
"level_routing": {
"CRITICAL": "storage.postgresql", # critical → PostgreSQL
},
"schemas": {
"credentials": ["email", "password_hash"],
"session": ["username", "uuid", "cookie"],
},
})Multi-node clustering with automatic failover.
from evoid_cluster import ClusterBridge, ClusterConfig
config = ClusterConfig(
node_id="node-1",
host="0.0.0.0",
port=9100,
services=["storage.sqlite", "cache.redis"],
peers={"node-2": {"host": "192.168.1.11", "port": 9100}},
)
bridge = ClusterBridge(config)
await bridge.start()Features:
- Automatic failover between nodes
- Load balancing across healthy nodes
- Auto-reconnect to offline nodes
- Heartbeat-based health checking
- Message bus integration
MIT