Schema evolution and migrations for versioned data. Register model versions, define migrations between them, and converge payloads to a target schema at runtime. The engine is provider-agnostic; Pydantic and JSON Schema adapters ship today.
pip install pyverge # core library
pip install "pyverge[cli]" # CLI
pip install "pyverge[fastmcp]" # MCP tool server
pip install "pyverge[faststream]" # messaging adapterDevelopment: uv sync --group dev.
from typing import Literal
import semver
from pydantic import BaseModel
from pyverge import Manager, MigrationSettings, PydanticModelAdapter
UserManager = Manager[semver.Version].configure(
MigrationSettings(),
PydanticModelAdapter(),
)
@UserManager.model()
class UserV1(BaseModel):
kind: Literal["User"] = "User"
version: Literal["1.0.0"] = "1.0.0"
name: str
email: str
@UserManager.model()
class UserV2(BaseModel):
kind: Literal["User"] = "User"
version: Literal["2.0.0"] = "2.0.0"
name: str
email: str
age: int | None = None
@UserManager.migration("User", "1.0.0", "2.0.0")
def add_age(data: dict, context=None) -> dict:
return {**data, "age": None}
migrated = UserManager().migrate(
{"kind": "User", "version": "1.0.0", "name": "Alice", "email": "a@b.com"}
)
# {"kind": "User", "version": "2.0.0", "name": "Alice", "email": "a@b.com", "age": None}result = await manager.amigrate(payload)
run = manager.delegate(payload)
first = run.run(context={"tenant_id": "tenant_a"})
second = run.run(context={"tenant_id": "tenant_b"})migrate/amigrate execute immediately. delegate and prepare return a
reusable MigrationRun. Execution policy comes from settings:
MigrationSettings(execution_mode="async", parallel_workers=4)| Format | Use when | Context |
|---|---|---|
| JSON Patch | structural changes | no |
| JSONata | declarative expressions | yes |
| Python callable | imperative logic | yes |
pyverge init # scaffold project
pyverge doctor --manager myapp.models:UserManager # registry health
pyverge validate data.json --manager myapp.models:UserManager \
--schema User --version 1.0.0
pyverge migrate data.json --manager myapp.models:UserManager --target latest
pyverge migrate data.json --manager myapp.models:UserManager --check
pyverge compat --manager myapp.models:UserManager --fail-on-breaking
pyverge manifest --manager myapp.models:UserManager > manifest.json
pyverge diff --manager myapp.models:UserManager --from 1.0.0 --to 2.0.0
pyverge export --manager myapp.models:UserManager
pyverge info --manager myapp.models:UserManagerManagers are module:object_path and may be an instance, configured subclass,
or zero-arg factory. Runtime schemas require a factory that builds the manager
from their source.
manifest is a CLI-only contract: it emits a JGF v2 graph envelope with node
schema hashes, migration hashes, and a content hash. Lifespan/startup does not
write it. CI commands (doctor, validate, migrate --check, compat)
return non-zero on failure. Config auto-discovery is not wired yet.
See CLI Reference.
- Versioned registry — semver or ISO date versions, decorator registration
- Convergent engine — graph-driven migration of nested versioned entries
- Declarative migrations — JSON Patch, JSONata expressions, or Python callables
- Runtime context — shared read-only bindings across every step
- Sync/async executors — sequential or level-parallel; prepared runs
- Reflection — reconstruct missing models or propose migrations from diffs
- Compatibility —
CompatibilityReport, breaking-change checks, JSON Patch rendering - Graph manifest — JGF v2 snapshot with stable hashes for CI divergence checks
- Adapters — FastMCP tools/prompts/resources and FastStream messages
- Hooks — before/after/error callbacks, async-aware, OpenTelemetry hook
Do not report security vulnerabilities through public issues. Use GitHub private vulnerability reporting. See SECURITY.md for scope, response targets, and disclosure timeline.