---

name: database-manager description: Manage PostgreSQL database for NutriProfile. Use this skill when working with SQLAlchemy models, Alembic migrations, query optimization, or database schema changes. Covers async operations and Fly Postgres deployment. allowed-tools: Read,Write,Edit,Grep,Glob,Bash

NutriProfile Database Manager Skill

You are a database management expert for the NutriProfile application. This skill helps you work with PostgreSQL, SQLAlchemy async, and Alembic migrations.

Database Stack

• Database: PostgreSQL (Fly Postgres in production)
• ORM: SQLAlchemy 2.0 with async support
• Migrations: Alembic
• Connection: asyncpg driver

Connection Configuration

# backend/app/config.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    DATABASE_URL: str = "postgresql+asyncpg://user:pass@localhost:5432/nutriprofile"

    class Config:
        env_file = ".env"

settings = Settings()

# backend/app/database.py
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase

engine = create_async_engine(
    settings.DATABASE_URL,
    echo=False,  # Set True for SQL logging
    pool_size=5,
    max_overflow=10
)

AsyncSessionLocal = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False
)

class Base(DeclarativeBase):
    pass

async def get_db():
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

SQLAlchemy Models

Model Structure

backend/app/models/
├── __init__.py
├── user.py           # User, authentication
├── profile.py        # Nutritional profile
├── food_log.py       # FoodLog, FoodItem, DailyNutrition
├── recipe.py         # Recipe, FavoriteRecipe, RecipeHistory
├── activity.py       # ActivityLog, WeightLog, Goal
├── gamification.py   # Achievement, Streak, UserStats, Notification
└── subscription.py   # Subscription

Model Example

# backend/app/models/user.py
from sqlalchemy import Column, Integer, String, Boolean, DateTime, ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from app.database import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String(255), unique=True, index=True, nullable=False)
    hashed_password = Column(String(255), nullable=False)
    name = Column(String(100))
    is_active = Column(Boolean, default=True)
    preferred_language = Column(String(5), default="en")
    subscription_tier = Column(String(20), default="free")
    trial_ends_at = Column(DateTime(timezone=True))
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    updated_at = Column(DateTime(timezone=True), onupdate=func.now())

    # Relationships
    profile = relationship("Profile", back_populates="user", uselist=False)
    food_logs = relationship("FoodLog", back_populates="user")
    recipes = relationship("Recipe", back_populates="user")
    subscription = relationship("Subscription", back_populates="user", uselist=False)

    def __repr__(self):
        return f"<User(id={self.id}, email={self.email})>"

Relationship Patterns

# One-to-One
class User(Base):
    profile = relationship("Profile", back_populates="user", uselist=False)

class Profile(Base):
    user_id = Column(Integer, ForeignKey("users.id"), unique=True)
    user = relationship("User", back_populates="profile")

# One-to-Many
class User(Base):
    food_logs = relationship("FoodLog", back_populates="user")

class FoodLog(Base):
    user_id = Column(Integer, ForeignKey("users.id"))
    user = relationship("User", back_populates="food_logs")

# Many-to-Many
class Recipe(Base):
    tags = relationship("Tag", secondary="recipe_tags", back_populates="recipes")

class Tag(Base):
    recipes = relationship("Recipe", secondary="recipe_tags", back_populates="tags")

recipe_tags = Table(
    "recipe_tags",
    Base.metadata,
    Column("recipe_id", ForeignKey("recipes.id")),
    Column("tag_id", ForeignKey("tags.id"))
)

Async Database Operations

CRUD Operations

# backend/app/services/recipe.py
from sqlalchemy import select, update, delete
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.recipe import Recipe

class RecipeService:
    def __init__(self, db: AsyncSession):
        self.db = db

    # CREATE
    async def create(self, user_id: int, data: RecipeCreate) -> Recipe:
        recipe = Recipe(
            user_id=user_id,
            **data.model_dump()
        )
        self.db.add(recipe)
        await self.db.commit()
        await self.db.refresh(recipe)
        return recipe

    # READ (single)
    async def get(self, recipe_id: int, user_id: int) -> Recipe | None:
        result = await self.db.execute(
            select(Recipe).where(
                Recipe.id == recipe_id,
                Recipe.user_id == user_id
            )
        )
        return result.scalar_one_or_none()

    # READ (list)
    async def get_all(
        self,
        user_id: int,
        skip: int = 0,
        limit: int = 20
    ) -> list[Recipe]:
        result = await self.db.execute(
            select(Recipe)
            .where(Recipe.user_id == user_id)
            .order_by(Recipe.created_at.desc())
            .offset(skip)
            .limit(limit)
        )
        return result.scalars().all()

    # UPDATE
    async def update(
        self,
        recipe_id: int,
        user_id: int,
        data: RecipeUpdate
    ) -> Recipe | None:
        recipe = await self.get(recipe_id, user_id)
        if not recipe:
            return None

        for key, value in data.model_dump(exclude_unset=True).items():
            setattr(recipe, key, value)

        await self.db.commit()
        await self.db.refresh(recipe)
        return recipe

    # DELETE
    async def delete(self, recipe_id: int, user_id: int) -> bool:
        result = await self.db.execute(
            delete(Recipe).where(
                Recipe.id == recipe_id,
                Recipe.user_id == user_id
            )
        )
        await self.db.commit()
        return result.rowcount > 0

Complex Queries

# Aggregation
from sqlalchemy import func

async def get_daily_calories(user_id: int, date: date) -> float:
    result = await self.db.execute(
        select(func.sum(FoodLog.total_calories))
        .where(
            FoodLog.user_id == user_id,
            func.date(FoodLog.created_at) == date
        )
    )
    return result.scalar() or 0.0

# Join with eager loading
from sqlalchemy.orm import selectinload

async def get_user_with_profile(user_id: int) -> User:
    result = await self.db.execute(
        select(User)
        .options(selectinload(User.profile))
        .where(User.id == user_id)
    )
    return result.scalar_one_or_none()

# Subquery
from sqlalchemy import select

subquery = (
    select(FoodLog.user_id, func.count(FoodLog.id).label("log_count"))
    .group_by(FoodLog.user_id)
    .subquery()
)

result = await self.db.execute(
    select(User, subquery.c.log_count)
    .outerjoin(subquery, User.id == subquery.c.user_id)
)

Alembic Migrations

Configuration

# backend/alembic/env.py
from app.database import Base
from app.models import user, profile, food_log, recipe, activity, gamification, subscription

target_metadata = Base.metadata

Creating Migrations

# Auto-generate migration from model changes
cd backend
alembic revision --autogenerate -m "add trial_ends_at to users"

# Create empty migration for manual changes
alembic revision -m "add custom index"

Migration Examples

# alembic/versions/xxxx_add_trial_ends_at.py
"""add trial_ends_at to users

Revision ID: xxxx
"""
from alembic import op
import sqlalchemy as sa

def upgrade():
    op.add_column(
        'users',
        sa.Column('trial_ends_at', sa.DateTime(timezone=True), nullable=True)
    )

def downgrade():
    op.drop_column('users', 'trial_ends_at')

# Adding index
def upgrade():
    op.create_index(
        'ix_food_logs_user_date',
        'food_logs',
        ['user_id', sa.text('DATE(created_at)')]
    )

def downgrade():
    op.drop_index('ix_food_logs_user_date')

Running Migrations

# Apply all pending migrations
alembic upgrade head

# Apply specific migration
alembic upgrade xxxx

# Rollback one migration
alembic downgrade -1

# Show current revision
alembic current

# Show migration history
alembic history

Fly Postgres

Connection in Production

# backend/fly.toml
[env]
  DATABASE_URL = "postgres://nutriprofile:password@nutriprofile-db.internal:5432/nutriprofile?sslmode=disable"

Connecting to Production DB

# Via Fly proxy
fly proxy 5432 -a nutriprofile-db

# Then connect with psql
psql postgres://nutriprofile:password@localhost:5432/nutriprofile

# Or run migrations remotely
fly ssh console -a nutriprofile-api
cd /app
alembic upgrade head

Database Backups

# Create backup
fly postgres backup create -a nutriprofile-db

# List backups
fly postgres backup list -a nutriprofile-db

# Restore backup
fly postgres backup restore <backup-id> -a nutriprofile-db

Performance Optimization

Indexes

# In model definition
class FoodLog(Base):
    __tablename__ = "food_logs"

    id = Column(Integer, primary_key=True, index=True)
    user_id = Column(Integer, ForeignKey("users.id"), index=True)
    created_at = Column(DateTime(timezone=True), index=True)

    __table_args__ = (
        Index('ix_food_logs_user_date', 'user_id', text('DATE(created_at)')),
    )

Query Optimization

# Avoid N+1 queries - use eager loading
result = await self.db.execute(
    select(FoodLog)
    .options(selectinload(FoodLog.items))
    .where(FoodLog.user_id == user_id)
)

# Use pagination
.offset(skip).limit(limit)

# Select only needed columns
select(User.id, User.email, User.name)

# Use exists() for existence checks
from sqlalchemy import exists
await self.db.execute(
    select(exists().where(User.email == email))
)

Data Models Reference

Users & Profiles

users (id, email, hashed_password, name, is_active, preferred_language, subscription_tier, trial_ends_at, created_at, updated_at)
profiles (id, user_id, age, gender, height, weight, activity_level, fitness_goal, diet_type, allergies, bmr, tdee, daily_calories, protein_target, carbs_target, fat_target, health_conditions, medications, created_at, updated_at)

Food & Nutrition

food_logs (id, user_id, meal_type, image_url, detected_items, user_corrections, total_calories, total_protein, total_carbs, total_fat, confidence_score, source, created_at)
daily_nutrition (id, user_id, date, target_calories, actual_calories, target_protein, actual_protein, target_carbs, actual_carbs, target_fat, actual_fat, water_intake, created_at)

Recipes

recipes (id, user_id, name, description, ingredients, instructions, prep_time, cook_time, servings, calories_per_serving, protein_per_serving, carbs_per_serving, fat_per_serving, difficulty, cuisine_type, tags, image_url, confidence_score, created_at)
favorite_recipes (id, user_id, recipe_id, created_at)
recipe_history (id, user_id, recipe_id, generated_at, ingredients_used, preferences_applied)

Activity & Goals

activity_logs (id, user_id, activity_type, duration, intensity, distance, calories_burned, calories_source, heart_rate_avg, steps, notes, created_at)
weight_logs (id, user_id, weight, body_fat_percentage, muscle_mass, notes, created_at)
goals (id, user_id, goal_type, target_value, current_value, period, is_completed, created_at, completed_at)

Gamification

achievements (id, user_id, achievement_type, unlocked_at)
streaks (id, user_id, streak_type, current_count, longest_count, last_activity_date)
user_stats (id, user_id, xp, level, total_meals_logged, total_recipes_generated, total_activities_logged, created_at, updated_at)
notifications (id, user_id, type, title, message, is_read, created_at)

Subscriptions

subscriptions (id, user_id, tier, status, current_period_start, current_period_end, cancel_at_period_end, ls_subscription_id, ls_customer_id, ls_variant_id, ls_order_id, created_at, updated_at)
usage_tracking (id, user_id, date, vision_analyses, recipe_generations, coach_messages, created_at, updated_at)

Best Practices

1. Always use async - All database operations should be async
2. Commit/rollback properly - Use the get_db dependency
3. Eager load relationships - Avoid N+1 queries
4. Index frequently queried columns - user_id, created_at
5. Use migrations - Never modify schema directly
6. Test migrations - Apply locally before production
7. Backup before migrations - Especially for destructive changes