This guide shows how to extend IdentityPlanKit with custom models for your application.
Use IdentityPlanKit's BaseModel which provides UUID primary keys and timestamps:
# models/profile.py
from uuid import UUID
from sqlalchemy import ForeignKey, String, Text
from sqlalchemy.orm import Mapped, mapped_column
from identity_plan_kit.shared.models import BaseModel
class UserProfile(BaseModel):
"""Extended user profile - separate table linked to IPK users."""
__tablename__ = "user_profiles"
user_id: Mapped[UUID] = mapped_column(
ForeignKey("users.id", ondelete="CASCADE"),
unique=True,
index=True,
)
phone: Mapped[str | None] = mapped_column(String(20))
bio: Mapped[str | None] = mapped_column(Text)
company: Mapped[str | None] = mapped_column(String(255))
alembic/env.py:
from identity_plan_kit.shared.database import Base
# Import IPK models
from identity_plan_kit.auth.models import user, user_provider, refresh_token
from identity_plan_kit.rbac.models import role, permission, role_permission
from identity_plan_kit.plans.models import plan, feature, plan_limit, user_plan, feature_usage
# Import YOUR models
from models.profile import UserProfile
target_metadata = Base.metadata
# Apply IPK base tables first
ipk db upgrade
# Generate migration for your models
alembic revision --autogenerate -m "Add user profiles"
# Apply your migration
alembic upgrade head
from fastapi import FastAPI
from sqlalchemy import select
from identity_plan_kit import IdentityPlanKit, IdentityPlanKitConfig, CurrentUser
from models.profile import UserProfile
config = IdentityPlanKitConfig()
kit = IdentityPlanKit(config)
app = FastAPI(lifespan=kit.lifespan)
kit.setup(app)
@app.post("/profile")
async def create_profile(user: CurrentUser):
async with kit.session_factory() as session:
profile = UserProfile(user_id=user.id, bio="Hello!")
session.add(profile)
await session.commit()
return {"id": str(profile.id)}
@app.get("/profile")
async def get_profile(user: CurrentUser):
async with kit.session_factory() as session:
result = await session.execute(
select(UserProfile).where(UserProfile.user_id == user.id)
)
profile = result.scalar_one_or_none()
return profile or {"error": "No profile"}
Create a link table to connect IPK users to your legacy users:
class UserLegacyLink(BaseModel):
__tablename__ = "user_legacy_links"
ipk_user_id: Mapped[UUID] = mapped_column(
ForeignKey("users.id", ondelete="CASCADE"),
unique=True,
)
legacy_user_id: Mapped[int] = mapped_column(
ForeignKey("legacy_users.id", ondelete="CASCADE"),
unique=True,
)
Then migrate existing users with a data migration.
Just install IPK and reference its users table:
class Project(BaseModel):
__tablename__ = "projects"
name: Mapped[str] = mapped_column(String(255))
owner_id: Mapped[UUID] = mapped_column(ForeignKey("users.id"))
users, roles, plans tablesipk db upgrade before your own migrationsSee extension_example.py for a full working example with: