In modern backend projects, especially with FastAPI and async SQLAlchemy, seeding initial data like (e.g.,roles) is an important part.
Here’s a practical and scalable approach we used to seed data smoothly:
✅ 1. Organized Seeders
Each seeder reads data from JSON files and checks if the entry already exists in the DB. If not, it creates it — avoiding duplicates.
[ { "role": "admin" }, { "role": "manager" }, { "role": "developer" }, { "role": "user" } ] # roles_seeder.py import json import os from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.future import select from app.models.roles import Role async def seed_roles(session: AsyncSession): try: json_path = os.path.join( os.path.dirname(__file__), 'data', 'roles.json' ) with open(json_path, 'r') as file: roles_to_seed = json.load(file) for role_data in roles_to_seed: role_name = role_data.get('role') if not role_name: print("[-] Skipping invalid role data: missing 'role'.") continue existing_role_query = await session.execute( select(Role).where(Role.role == role_name) ) existing_role = existing_role_query.scalars().first() if not existing_role: print(f"[+] Creating new role '{role_name}'.") new_role = Role(role=role_name) session.add(new_role) await session.commit() print('[+] Roles seeded or updated successfully.') except Exception as e: await session.rollback() print(f'[-] Error while seeding or updating roles: {e}') raise Note: Your code might be different based on your requirements.
✅ 2. Shared Async Context
We centralize DB session logic using sessionmanager to handle init/close properly with async SQLAlchemy.
# cli.py import asyncio import typer from app.seed.articles_seeder import seed_articles from app.seed.categories_seeder import seed_categories from app.seed.roles_seeder import seed_roles from app.services.config import config from app.services.connection import sessionmanager cli = typer.Typer() async def run_seed(func): sessionmanager.init(config.db_dsn) async with sessionmanager.session() as session: await func(session) await session.commit() await sessionmanager.close() @cli.command() def roles(): asyncio.run(run_seed(seed_roles)) @cli.command() def categories(): asyncio.run(run_seed(seed_categories)) @cli.command() def articles(): asyncio.run(run_seed(seed_articles)) if __name__ == '__main__': cli() ✅ 3. CLI with Typer
Typer gives us a clean CLI to run seed commands like:
python3 cli.py roles ✉️ Conclusion:
I didn’t go into too much detail here—just shared the core code for you to copy and use. Hopefully, you’re already familiar with Python and SQLAlchemy.
Top comments (0)