pyrmute-registry 0.0.1

Pyrmute Schema Registry

Pyrmute Registry

A centralized schema registry for pyrmute, enabling teams to discover, version, and manage Pydantic model schemas across services.

Overview

Pyrmute Registry provides both a client and plugin for integrating your pyrmute-based applications with a central schema registry server. Think of it as a "Confluence for schemas" - a single source of truth for all your data models.

Key Features

• Automatic Schema Registration - Schemas are registered as you define models
• Schema Discovery - Find and compare schemas across services
• Version Management - Track schema evolution and breaking changes
• Multi-Tenant Support - Namespace schemas by service or use globally
• Multi-Service Coordination - Share schemas between services
• Zero-Config Integration - Drop-in plugin for existing pyrmute projects
• Robust Error Handling - Graceful degradation when registry is unavailable
• Authentication Support - API key-based security

Installation

pip install pyrmute-registry

Or with extras:

pip install pyrmute-registry[server]  # Include FastAPI server components

Quick Start

Basic Usage (Namespaced)

from pyrmute import ModelManager
from pyrmute_registry import RegistryPlugin
from pydantic import BaseModel

# Create your ModelManager
manager = ModelManager()

# Wrap it with the registry plugin
with RegistryPlugin(
    manager,
    registry_url="http://registry:8000",
    namespace="user-service",  # Service-specific namespace
):

    # Define models as usual - they're automatically registered!
    @manager.model("User", "1.0.0")
    class UserV1(BaseModel):
        name: str
        email: str

    # That's it! The schema is now in the registry under user-service namespace.

Global Schemas

For schemas that should be shared across all services:

# Register as global schemas (no namespace)
plugin = RegistryPlugin(
    manager,
    registry_url="http://registry:8000",
    namespace=None,  # Global schemas
)

@manager.model("CommonModel", "1.0.0")
class CommonModel(BaseModel):
    id: str
    created_at: str

Environment Variables

Set configuration via environment variables:

export PYRMUTE_REGISTRY_URL="http://registry:8000"
export PYRMUTE_REGISTRY_NAMESPACE="user-service"  # Optional, None for global
export PYRMUTE_REGISTRY_API_KEY="your-api-key"    # Optional

Then use without explicit configuration:

from pyrmute import ModelManager
from pyrmute_registry import create_plugin

manager = ModelManager()
plugin = create_plugin(manager)  # Reads from environment

@manager.model("User", "1.0.0")
class UserV1(BaseModel):
    name: str

Configuration

Plugin Options

plugin = RegistryPlugin(
    manager=manager,                      # Your ModelManager instance
    registry_url="http://registry:8000",  # Registry server URL
    namespace="my-service",               # Optional namespace (None for global)
    auto_register=True,                   # Auto-register on model definition
    fail_on_error=False,                  # Raise exceptions on registry errors
    verify_ssl=True,                      # Verify SSL certificates
    api_key=None,                         # Optional API key for auth
    allow_overwrite=False,                # Allow overwriting existing schemas
    metadata={"team": "platform"},        # Default metadata for all schemas
)

Client Options

from pyrmute_registry import RegistryClient

client = RegistryClient(
    base_url="http://registry:8000",
    timeout=30.0,           # Request timeout in seconds
    max_retries=3,          # Retry attempts for transient failures
    verify_ssl=True,        # Verify SSL certificates
    api_key=None,           # Optional API key
)

Namespaces

Pyrmute Registry supports multi-tenant schema organization through namespaces:

When to Use Namespaces

• Namespaced schemas: Service-specific schemas that may differ between services
  • Example: user-service::User@1.0.0 vs admin-service::User@1.0.0
• Global schemas: Shared schemas used across all services
  • Example: CommonTypes@1.0.0 (no namespace)

Working with Namespaces

# Register namespaced schema
client.register_schema(
    "User",
    "1.0.0",
    schema,
    "user-service",
    namespace="auth-service",  # Scoped to auth-service
)

# Register global schema
client.register_schema(
    "CommonModel",
    "1.0.0",
    schema,
    "platform-team",
    namespace=None,  # Available to all services
)

# Get namespaced schema
schema = client.get_schema("User", "1.0.0", namespace="auth-service")

# Get global schema
schema = client.get_schema("CommonModel", "1.0.0", namespace=None)

# List schemas in a namespace
schemas = client.list_schemas(namespace="auth-service")

# List all global schemas
schemas = client.list_schemas(namespace="")

# List schemas across all namespaces
schemas = client.list_schemas()  # namespace=None means all

Usage Patterns

Manual Registration Control

Disable auto-registration and register selectively:

plugin = RegistryPlugin(
    manager,
    registry_url="http://registry:8000",
    namespace="user-service",
    auto_register=False,  # Don't auto-register
)

# Define your models
@manager.model("User", "1.0.0")
class UserV1(BaseModel):
    name: str

@manager.model("InternalModel", "1.0.0")
class InternalModelV1(BaseModel):
    secret: str

# Only register public models
plugin.register_existing_models([
    ("User", "1.0.0"),
    # InternalModel not registered
])

Registering Existing Models

If you have models defined before creating the plugin:

# Models already defined
@manager.model("User", "1.0.0")
class UserV1(BaseModel):
    name: str

@manager.model("Product", "1.0.0")
class ProductV1(BaseModel):
    title: str

# Create plugin and register all existing
plugin = RegistryPlugin(
    manager,
    registry_url="...",
    namespace="catalog-service",
    auto_register=False
)
results = plugin.register_existing_models()

print(results)
# {"User@1.0.0": True, "Product@1.0.0": True}

Schema Discovery

Find schemas from other services:

from pyrmute_registry import RegistryClient

client = RegistryClient("http://registry:8000")

# List all schemas (across all namespaces)
schemas = client.list_schemas()
for schema_info in schemas["schemas"]:
    ns = schema_info["namespace"] or "global"
    print(f"[{ns}] {schema_info['model_name']}: {schema_info['versions']}")

# List schemas in specific namespace
user_schemas = client.list_schemas(namespace="user-service")

# List only global schemas
global_schemas = client.list_schemas(namespace="")

# Get specific schema
user_schema = client.get_schema("User", "1.0.0", namespace="user-service")
print(user_schema)

# Get latest version
latest_user = client.get_latest_schema("User", namespace="user-service")
print(f"Latest: {latest_user['version']}")

Schema Comparison

Compare schemas across versions:

# Compare two versions in same namespace
diff = client.compare_schemas(
    "User",
    "1.0.0",
    "2.0.0",
    namespace="user-service"
)
print(diff["changes"])

# Using the plugin
comparison = plugin.compare_with_registry("User", "1.0.0")
if not comparison["matches"]:
    print("Schema drift detected!")
    print(f"Added: {comparison['differences']['properties_added']}")
    print(f"Removed: {comparison['differences']['properties_removed']}")

Schema Deprecation

Mark schemas as deprecated:

# Deprecate a schema version
client.deprecate_schema(
    "User",
    "1.0.0",
    namespace="user-service",
    message="Security vulnerability. Please upgrade to 2.0.0"
)

# List including deprecated schemas
schemas = client.list_schemas(
    namespace="user-service",
    include_deprecated=True
)

for schema in schemas["schemas"]:
    if schema["deprecated_versions"]:
        print(f"Deprecated: {schema['deprecated_versions']}")

Synchronization

Check sync status between local and registry:

status = plugin.sync_with_registry()

if not status["in_sync"]:
    if status["local_only"]:
        print("Models only in local:")
        for model, versions in status["local_only"].items():
            print(f"  {model}: {versions}")

    if status["registry_only"]:
        print("Models only in registry:")
        for model, versions in status["registry_only"].items():
            print(f"  {model}: {versions}")

    if status["version_mismatches"]:
        print("Version mismatches:")
        for model, diff in status["version_mismatches"].items():
            print(f"  {model}:")
            print(f"    Local only: {diff['local_only']}")
            print(f"    Registry only: {diff['registry_only']}")

Validation

Ensure your local schema matches the registry:

# Check if schemas match
is_valid = plugin.validate_against_registry("User", "1.0.0")
if not is_valid:
    print("Warning: Local schema differs from registry!")

# Or raise on mismatch
try:
    plugin.validate_against_registry(
        "User", "1.0.0",
        raise_on_mismatch=True
    )
except RegistryPluginError as e:
    print(f"Schema validation failed: {e}")

Error Handling Modes

Fail-Fast Mode

Raise exceptions immediately on any registry error:

plugin = RegistryPlugin(
    manager,
    registry_url="http://registry:8000",
    namespace="user-service",
    fail_on_error=True,  # Raise on errors
)

try:
    @manager.model("User", "1.0.0")
    class UserV1(BaseModel):
        name: str
except RegistryPluginError as e:
    print(f"Registration failed: {e}")
    # Handle the error

Graceful Degradation Mode (Default)

Continue operation even if registry is unavailable:

plugin = RegistryPlugin(
    manager,
    registry_url="http://registry:8000",
    namespace="user-service",
    fail_on_error=False,  # Default: warn but continue
)

# Even if registry is down, models are still usable
@manager.model("User", "1.0.0")
class UserV1(BaseModel):
    name: str

# User model works normally, registry registration just failed silently
user = UserV1(name="Alice")

Schema Metadata

Add custom metadata to schemas:

# Default metadata for all schemas
plugin = RegistryPlugin(
    manager,
    registry_url="http://registry:8000",
    namespace="user-service",
    metadata={
        "team": "platform",
        "owner": "[e-mail]",
        "criticality": "high",
    }
)

# Update metadata later
plugin.set_metadata({"last_reviewed": "2025-01-15"})

@manager.model("User", "1.0.0")
class UserV1(BaseModel):
    name: str
# Schema includes all metadata

Health Checks

Monitor plugin and registry health:

health = plugin.health_check()

print(f"Plugin active: {health['plugin_active']}")
print(f"Registry healthy: {health['registry_healthy']}")
print(f"Registered models: {health['registered_models']}")
print(f"Namespace: {health['namespace']}")

if not health["registry_healthy"]:
    print(f"Registry error: {health.get('registry_error')}")

Using the Client Directly

For advanced use cases, use the client without the plugin:

from pyrmute_registry import RegistryClient

with RegistryClient("http://registry:8000") as client:
    # Register a namespaced schema
    client.register_schema(
        model_name="User",
        version="1.0.0",
        schema={
            "type": "object",
            "properties": {
                "name": {"type": "string"},
            }
        },
        registered_by="user-service",
        namespace="user-service",
        metadata={"author": "Alice"},
    )

    # Register a global schema
    client.register_schema(
        model_name="CommonType",
        version="1.0.0",
        schema={"type": "object"},
        registered_by="platform-team",
        namespace=None,  # Global
    )

    # List all versions in namespace
    versions = client.list_versions("User", namespace="user-service")
    print(f"Available versions: {versions}")

    # Delete a schema (use with caution!)
    client.delete_schema(
        "User",
        "0.9.0",
        namespace="user-service",
        force=True
    )

Best Practices

Namespace Strategy

Choose the right namespace approach for your use case:

# Service-specific schemas (most common)
plugin = RegistryPlugin(
    manager,
    registry_url="http://registry:8000",
    namespace="user-service",  # Scoped to this service
)

# Global schemas (shared models)
plugin = RegistryPlugin(
    manager,
    registry_url="http://registry:8000",
    namespace=None,  # Available to all services
)

When to use namespaces:

• Different services need different versions of the same model
• Service-specific schemas that shouldn't be shared
• Multi-tenant deployments

When to use global schemas:

• Common types shared across all services
• Standard contracts between services
• Platform-wide data models

Namespace Naming

Use consistent, descriptive namespace names:

# Good
namespace = "user-api"
namespace = "payment-service"
namespace = "analytics-worker"

# Avoid
namespace = "service1"
namespace = "test"
namespace = "my-namespace"

Version Management

Follow semantic versioning principles:

# Breaking change: Increment major version
@manager.model("User", "2.0.0")  # Was 1.x.x
class UserV2(BaseModel):
    id: str  # Changed from int to str (breaking)

# New field: Increment minor version
@manager.model("User", "1.1.0")  # Was 1.0.x
class UserV1_1(BaseModel):
    name: str
    email: str  # New optional field

# Bug fix: Increment patch version
@manager.model("User", "1.0.1")  # Was 1.0.0
class UserV1_0_1(BaseModel):
    name: str  # Fixed validation logic

Metadata Usage

Add meaningful metadata for discoverability:

plugin = RegistryPlugin(
    manager,
    registry_url="http://registry:8000",
    namespace="user-service",
    metadata={
        "team": "platform-team",
        "owner": "[e-mail]",
        "repo": "github.com/company/user-service",
        "docs": "https://docs.company.com/schemas/user",
        "environment": "production",
    }
)

Error Handling Strategy

Choose the right mode for your use case:

# Production services: Graceful degradation
plugin = RegistryPlugin(
    manager,
    registry_url="http://registry:8000",
    namespace="user-service",
    fail_on_error=False,  # Don't break service if registry is down
)

# CI/CD pipelines: Fail-fast
plugin = RegistryPlugin(
    manager,
    registry_url="http://registry:8000",
    namespace="user-service",
    fail_on_error=True,  # Catch schema issues early
)

Validation in CI

Validate schemas in your CI pipeline:

# In your CI script
plugin = RegistryPlugin(
    manager,
    registry_url="...",
    namespace="user-service",
    auto_register=False
)

# Define all models
# ...

# Validate against registry
all_valid = True
for model_name in manager.list_models():
    for version in manager.list_versions(model_name):
        if not plugin.validate_against_registry(str(model_name), str(version)):
            print(f"Schema drift detected: {model_name} {version}")
            all_valid = False

if not all_valid:
    sys.exit(1)

Troubleshooting

Registry Connection Issues

# Check connectivity
plugin = RegistryPlugin(
    manager,
    registry_url="http://registry:8000",
    namespace="user-service"
)
health = plugin.health_check()

if not health["registry_healthy"]:
    print(f"Registry is down: {health.get('registry_error')}")
    # Check network, DNS, firewall rules

Schema Conflicts

# Enable overwrite mode
plugin = RegistryPlugin(
    manager,
    registry_url="http://registry:8000",
    namespace="user-service",
    allow_overwrite=True,  # Allow replacing schemas
)

# Or handle conflicts manually
from pyrmute_registry.exceptions import SchemaConflictError

try:
    plugin._register_schema_safe("User", "1.0.0", schema)
except SchemaConflictError:
    # Delete old schema first
    client.delete_schema("User", "1.0.0", namespace="user-service", force=True)
    # Then register new one
    plugin._register_schema_safe("User", "1.0.0", schema)

Schema Drift

# Detect drift
comparison = plugin.compare_with_registry("User", "1.0.0")
if not comparison["matches"]:
    print("Schemas differ:")
    print(f"  Local: {comparison['local_schema']}")
    print(f"  Registry: {comparison['registry_schema']}")

    # Decide: update registry or fix local?

Namespace Issues

# Wrong namespace
schema = client.get_schema("User", "1.0.0", namespace="wrong-service")
# SchemaNotFoundError

# Check which namespaces have this model
namespaces = client.list_schemas(model_name="User")
for schema in namespaces["schemas"]:
    print(f"Found in: {schema['namespace']}")

Contributing

Contributions are welcome! Please see CONTRIBUTING.md for guidelines.

Reporting a Security Vulnerability

See our security policy.

License

MIT License - see LICENSE for details.