
Security News
Meet Socket at Black Hat and DEF CON 2025 in Las Vegas
Meet Socket at Black Hat & DEF CON 2025 for 1:1s, insider security talks at Allegiant Stadium, and a private dinner with top minds in software supply chain security.
db-migration-manager
Advanced tools
A comprehensive, ORM-agnostic database migration system with FastAPI integration, supporting PostgreSQL, MySQL, and SQLite
A comprehensive, ORM-agnostic database migration system with FastAPI integration and Pydantic model support, supporting PostgreSQL, MySQL, and SQLite.
pip install db-migration-manager
# PostgreSQL support
pip install db-migration-manager[postgresql]
# MySQL support
pip install db-migration-manager[mysql]
# SQLite support
pip install db-migration-manager[sqlite]
# FastAPI integration
pip install db-migration-manager[fastapi]
# All dependencies
pip install db-migration-manager[all]
import asyncio
from db_migration_manager import PostgreSQLAdapter, MigrationManager
async def main():
# Initialize database adapter
db_adapter = PostgreSQLAdapter("postgresql://user:pass@localhost/db")
# Create migration manager
manager = MigrationManager(db_adapter)
await manager.initialize()
# Create a migration
await manager.create_migration(
"create_users_table",
up_sql="""
CREATE TABLE users (
id SERIAL PRIMARY KEY,
email VARCHAR(255) UNIQUE NOT NULL,
name VARCHAR(255) NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""",
down_sql="DROP TABLE users"
)
# Apply migrations
results = await manager.migrate()
print(f"Applied {len(results)} migrations")
# Get status
status = await manager.get_migration_status()
print(f"Applied: {status['applied_count']}, Pending: {status['pending_count']}")
asyncio.run(main())
Define your database schema using Pydantic models:
from datetime import datetime
from typing import Optional
from pydantic import Field
from db_migration_manager import DatabaseModel, primary_key, unique_field, indexed_field
class User(DatabaseModel):
# Primary key with auto-increment
id: int = primary_key(default=None)
# Unique email field
email: str = unique_field(max_length=255)
# Username with unique constraint
username: str = db_field(unique=True, max_length=50)
# Full name
full_name: str = Field(..., max_length=255)
# Timestamps
created_at: datetime = Field(default_factory=datetime.now)
is_active: bool = Field(default=True)
class Config:
__table_name__ = "users"
# Create migration from models
import asyncio
from db_migration_manager import PostgreSQLAdapter, MigrationManager
async def create_migration_from_models():
db_adapter = PostgreSQLAdapter("postgresql://user:pass@localhost/db")
manager = MigrationManager(db_adapter)
await manager.initialize()
# Create migration from Pydantic models
filepath = await manager.create_migration_from_models(
name="create_user_table",
models=[User],
auto_diff=True # Automatically compare with previous schema
)
print(f"Created migration: {filepath}")
# Apply the migration
results = await manager.migrate()
print(f"Applied {len(results)} migrations")
asyncio.run(create_migration_from_models())
from fastapi import FastAPI
from db_migration_manager import PostgreSQLAdapter, MigrationManager
from db_migration_manager.api import add_migration_routes
app = FastAPI()
# Initialize migration manager
db_adapter = PostgreSQLAdapter("postgresql://user:pass@localhost/db")
manager = MigrationManager(db_adapter)
# Add migration routes
add_migration_routes(app, manager)
# Your API endpoints...
@app.get("/")
async def root():
return {"message": "Hello World"}
# Available migration endpoints:
# GET /health - Health check
# GET /migrations/status - Migration status
# GET /migrations/pending - Pending migrations
# POST /migrations/migrate - Apply migrations
# POST /migrations/rollback - Rollback migrations
# POST /migrations/create - Create new migration
# POST /migrations/create-from-models - Create migration from Pydantic models
# POST /migrations/validate-models - Validate Pydantic models
# POST /migrations/show-sql - Show SQL for Pydantic model
# Set database URL
export DATABASE_URL="postgresql://user:pass@localhost/db"
# Check migration status
db-migrate status
# Create a new migration
db-migrate create add_user_profile --up-sql "ALTER TABLE users ADD COLUMN profile TEXT"
# Apply pending migrations
db-migrate migrate
# Rollback to specific version
db-migrate rollback 20240101_120000
# Create migration from Pydantic models
db-migrate create-from-models create_users my_app.models
# Validate Pydantic models
db-migrate validate-models my_app.models
# Show SQL for a specific model
db-migrate show-sql User my_app.models --dialect postgresql
# Help
db-migrate --help
from db_migration_manager import PostgreSQLAdapter
adapter = PostgreSQLAdapter("postgresql://user:pass@localhost:5432/dbname")
from db_migration_manager import MySQLAdapter
adapter = MySQLAdapter({
'host': 'localhost',
'user': 'user',
'password': 'password',
'db': 'dbname',
'port': 3306
})
from db_migration_manager import SQLiteAdapter
adapter = SQLiteAdapter("path/to/database.db")
The library provides special field annotations for database-specific features:
from db_migration_manager import (
DatabaseModel,
primary_key,
unique_field,
indexed_field,
db_field
)
class User(DatabaseModel):
# Primary key with auto-increment
id: int = primary_key(default=None)
# Unique field with length constraint
email: str = unique_field(max_length=255)
# Indexed field
username: str = indexed_field(max_length=50)
# Custom field with multiple constraints
slug: str = db_field(
unique=True,
index=True,
max_length=100
)
# Regular Pydantic field (stored as TEXT/VARCHAR)
bio: Optional[str] = None
# JSON field (stored as JSONB in PostgreSQL, JSON in MySQL, TEXT in SQLite)
metadata: dict = Field(default_factory=dict)
# Enum field (stored as VARCHAR)
status: UserStatus = Field(default=UserStatus.ACTIVE)
primary_key(**kwargs)
- Creates a primary key field with auto-incrementunique_field(**kwargs)
- Creates a unique fieldindexed_field(**kwargs)
- Creates an indexed fielddb_field(**kwargs)
- Custom field with database-specific options:
primary_key: bool
- Primary key constraintunique: bool
- Unique constraintindex: bool
- Create indexunique_index: bool
- Create unique indexauto_increment: bool
- Auto-increment for integersmax_length: int
- Maximum length for stringsPython Type | PostgreSQL | MySQL | SQLite |
---|---|---|---|
str | VARCHAR(255) | VARCHAR(255) | TEXT |
int | INTEGER | INT | INTEGER |
float | DOUBLE PRECISION | DOUBLE | REAL |
bool | BOOLEAN | TINYINT(1) | INTEGER |
datetime | TIMESTAMP | DATETIME | TIMESTAMP |
date | DATE | DATE | DATE |
Decimal | DECIMAL | DECIMAL | DECIMAL |
list /dict | JSONB | JSON | TEXT |
Enum | VARCHAR(50) | VARCHAR(50) | TEXT |
git checkout -b feature/amazing-feature
)git commit -m 'Add amazing feature'
)git push origin feature/amazing-feature
)This project is licensed under the MIT License - see the LICENSE file for details.
FAQs
A comprehensive, ORM-agnostic database migration system with FastAPI integration, supporting PostgreSQL, MySQL, and SQLite
We found that db-migration-manager demonstrated a healthy version release cadence and project activity because the last version was released less than a year ago.ย It has 1 open source maintainer collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Security News
Meet Socket at Black Hat & DEF CON 2025 for 1:1s, insider security talks at Allegiant Stadium, and a private dinner with top minds in software supply chain security.
Security News
CAI is a new open source AI framework that automates penetration testing tasks like scanning and exploitation up to 3,600ร faster than humans.
Security News
Deno 2.4 brings back bundling, improves dependency updates and telemetry, and makes the runtime more practical for real-world JavaScript projects.