Checkpoint3

app/db/alembic/env.py

@@ -18,8 +18,8 @@
 def import_all_models():
     # Correct the path to the models directory
     models_dir = os.path.join(os.path.dirname(__file__), "../models")
-    models_dir = os.path.abspath(models_dir)  # Ensure it's an absolute path
-
+    models_dir = os.path.abspath(models_dir)  
+    
     for filename in os.listdir(models_dir):
         if filename.endswith(".py") and filename != "__init__.py":
             module_name = f"app.db.models.{filename[:-3]}"

app/db/alembic/versions/5cfc72e753ba_transforming_schema.py

@@ -0,0 +1,54 @@
+"""transforming schema
+
+Revision ID: 5cfc72e753ba
+Revises: 075b701a1a93
+Create Date: 2025-05-27 16:16:34.166953
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = '5cfc72e753ba'
+down_revision: Union[str, None] = '075b701a1a93'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('permissions',
+    sa.Column('name', sa.String(length=100), nullable=False),
+    sa.PrimaryKeyConstraint('name')
+    )
+    op.create_table('roles',
+    sa.Column('name', sa.String(length=50), nullable=False),
+    sa.PrimaryKeyConstraint('name')
+    )
+    op.create_table('rules',
+    sa.Column('role_name', sa.String(length=50), nullable=False),
+    sa.Column('permission_name', sa.String(length=100), nullable=False),
+    sa.ForeignKeyConstraint(['permission_name'], ['permissions.name'], ),
+    sa.ForeignKeyConstraint(['role_name'], ['roles.name'], ),
+    sa.PrimaryKeyConstraint('role_name', 'permission_name')
+    )
+    op.add_column('users', sa.Column('role_name', sa.String(length=50), nullable=False))
+    op.create_foreign_key(None, 'users', 'roles', ['role_name'], ['name'])
+    op.drop_column('users', 'role')
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('users', sa.Column('role', sa.VARCHAR(length=50), autoincrement=False, nullable=False))
+    op.drop_constraint(None, 'users', type_='foreignkey')
+    op.drop_column('users', 'role_name')
+    op.drop_table('rules')
+    op.drop_table('roles')
+    op.drop_table('permissions')
+    # ### end Alembic commands ###

app/db/alembic/versions/9951448af167_testing_schema_change_potentially.py

@@ -0,0 +1,32 @@
+"""testing schema change potentially
+
+Revision ID: 9951448af167
+Revises: 5cfc72e753ba
+Create Date: 2025-05-27 19:07:18.419756
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = '9951448af167'
+down_revision: Union[str, None] = '5cfc72e753ba'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    pass
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    pass
+    # ### end Alembic commands ###

app/db/crud/auth.py

@@ -0,0 +1,25 @@
+from sqlalchemy.orm import Session
+from app.db.models import User, Rule
+
+def get_user_by_username(db: Session, username: str) -> User | None:
+    print("Debug: Listing all users in the database:")
+    all_users = db.query(User).all()
+    for user in all_users:
+        print(f" - id={user.id}, username={user.username}, email={user.email}, role={user.role_name}")
+
+    print(f"Debug: Searching for user with username='{username}'")
+    user = db.query(User).filter(User.username == username).first()
+    if user:
+        print(f"Debug: Found user with id={user.id}")
+    else:
+        print("Debug: No user found")
+    return user
+
+
+def get_permissions_for_role(db: Session, role_name: str) -> list[str]:
+    permissions = (
+        db.query(Rule.permission_name)
+        .filter(Rule.role_name == role_name)
+        .all()
+    )
+    return [perm.permission_name for perm in permissions]

app/db/crud/orders.py

@@ -1,5 +1,5 @@
 from sqlalchemy.orm import Session
-from app.db.models.orders import Order
+from app.db.models import Order
 
 def get_order_by_id(db: Session, order_id: int) -> Order | None:
     return db.query(Order).filter(Order.id == order_id).first()

app/db/crud/users.py

@@ -1,6 +1,6 @@
 from sqlalchemy.orm import Session
 from fastapi import HTTPException, status
-from app.db.models.users import User
+from app.db.models import User
 from app.validators.users import UserUpdate
 
 def get_user_by_id(db: Session, user_id: int) -> User | None:

app/db/models/__init__.py

@@ -0,0 +1,9 @@
+# Import your declarative base from app/db/base.py
+from app.db.base import Base
+
+# Import all your model classes from their respective files within this directory
+from .orders import Order
+from .permissions import Permission
+from .roles import Role
+from .rules import Rule
+from .users import User

app/db/models/permissions.py

@@ -0,0 +1,9 @@
+from sqlalchemy import Column, String
+from sqlalchemy.orm import relationship
+from app.db.base import Base
+
+class Permission(Base):
+    __tablename__ = "permissions"
+
+    name = Column(String(100), primary_key=True)
+    rules = relationship("Rule", back_populates="permission", cascade="all, delete-orphan")

app/db/models/roles.py

@@ -0,0 +1,10 @@
+from sqlalchemy import Column, String
+from sqlalchemy.orm import relationship
+from app.db.base import Base
+
+class Role(Base):
+    __tablename__ = "roles"
+
+    name = Column(String(50), primary_key=True)
+    users = relationship("User", back_populates="role")
+    rules = relationship("Rule", back_populates="role", cascade="all, delete-orphan")

app/db/models/rules.py

@@ -0,0 +1,12 @@
+from sqlalchemy import Column, String, ForeignKey
+from sqlalchemy.orm import relationship
+from app.db.base import Base
+
+class Rule(Base):
+    __tablename__ = "rules"
+
+    role_name = Column(String(50), ForeignKey("roles.name"), primary_key=True)
+    permission_name = Column(String(100), ForeignKey("permissions.name"), primary_key=True)
+
+    role = relationship("Role", back_populates="rules")
+    permission = relationship("Permission", back_populates="rules")

app/db/models/users.py

@@ -1,7 +1,6 @@
-from sqlalchemy import Column, Integer, String, DateTime, func
+from sqlalchemy import Column, Integer, String, DateTime, ForeignKey, func
 from sqlalchemy.orm import relationship
 from app.db.base import Base
-from app.db.models.orders import Order
 
 class User(Base):
     __tablename__ = "users"
@@ -10,8 +9,9 @@ class User(Base):
     username = Column(String(255), unique=True, nullable=False)
     email = Column(String(255), unique=True, nullable=False)
     hashed_password = Column(String(255), nullable=False)
-    role = Column(String(50), nullable=False, default="customer")
+    role_name = Column(String(50), ForeignKey("roles.name"), nullable=False, default="customer")
     created_at = Column(DateTime, server_default=func.now(), nullable=False)
     updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now(), nullable=False)
 
-    orders = relationship(Order, back_populates="user", cascade="all, delete-orphan")
+    role = relationship("Role", back_populates="users")
+    orders = relationship("Order", back_populates="user", cascade="all, delete-orphan")  # Use string reference

app/db/seed.py

@@ -0,0 +1,125 @@
+from app.db.session import SessionLocal
+from app.db.models.roles import Role
+from app.db.models.permissions import Permission
+from app.db.models import Rule, User
+from app.utils.security import hash_password
+
+import os
+from dotenv import load_dotenv
+from sqlalchemy import text
+
+load_dotenv() # Load environment variables from .env file
+
+def clear_tables(db):
+    # Truncate with restart identity to reset auto-increment IDs
+    db.execute(text("TRUNCATE TABLE rules, permissions, roles, users RESTART IDENTITY CASCADE"))
+    db.commit()
+
+
+def seed_roles(db):
+    roles = [
+        Role(name="admin"),
+        Role(name="customer"),
+        # add more roles if needed
+    ]
+    for role in roles:
+        existing = db.query(Role).filter(Role.name == role.name).first()
+        if not existing:
+            db.add(role)
+    db.commit()
+
+
+def seed_permissions(db):
+    permissions = [
+        # User related permissions
+        Permission(name="can_create_user"),
+        Permission(name="can_list_users"),
+        Permission(name="can_get_user"),
+        Permission(name="can_update_user"),
+        Permission(name="can_delete_user"),
+
+        Permission(name="can_get_own_profile"),
+        Permission(name="can_update_own_profile"),
+    ]
+    for perm in permissions:
+        existing = db.query(Permission).filter(Permission.name == perm.name).first()
+        if not existing:
+            db.add(perm)
+    db.commit()
+
+
+def seed_rules(db):
+    # Fetch roles
+    admin_role = db.query(Role).filter(Role.name == "admin").first()
+    customer_role = db.query(Role).filter(Role.name == "customer").first()
+
+    # Fetch permissions
+    can_create_user = db.query(Permission).filter(Permission.name == "can_create_user").first()
+    can_list_users = db.query(Permission).filter(Permission.name == "can_list_users").first()
+    can_get_user = db.query(Permission).filter(Permission.name == "can_get_user").first()
+    can_update_user = db.query(Permission).filter(Permission.name == "can_update_user").first()
+    can_delete_user = db.query(Permission).filter(Permission.name == "can_delete_user").first()
+
+    can_get_own_profile = db.query(Permission).filter(Permission.name == "can_get_own_profile").first()
+    can_update_own_profile = db.query(Permission).filter(Permission.name == "can_update_own_profile").first()
+
+    rules = [
+        # Admin permissions - full user management
+        Rule(role_name=admin_role.name, permission_name=can_create_user.name),
+        Rule(role_name=admin_role.name, permission_name=can_list_users.name),
+        Rule(role_name=admin_role.name, permission_name=can_get_user.name),
+        Rule(role_name=admin_role.name, permission_name=can_update_user.name),
+        Rule(role_name=admin_role.name, permission_name=can_delete_user.name),
+
+        # Customer permissions - maybe limited or none here
+        Rule(role_name=customer_role.name, permission_name=can_get_own_profile.name),
+        Rule(role_name=customer_role.name, permission_name=can_update_own_profile.name),
+    ]
+
+    for rule in rules:
+        existing = db.query(Rule).filter(
+            Rule.role_name == rule.role_name,
+            Rule.permission_name == rule.permission_name
+        ).first()
+        if not existing:
+            db.add(rule)
+
+    db.commit()
+
+
+def seed_admin_user(db):
+    admin_role = db.query(Role).filter(Role.name == "admin").first()
+    existing_admin = db.query(User).filter(User.role_name == admin_role.name).first()
+    if existing_admin:
+        print("Admin user already exists, skipping creation.")
+        return
+
+    admin_username = os.getenv("ADMIN_USERNAME", "admin")
+    admin_email = os.getenv("ADMIN_EMAIL", "admin@example.com")
+    admin_password = os.getenv("ADMIN_PASSWORD", "adminpassword123")
+
+    admin_user = User(
+        username=admin_username,
+        email=admin_email,
+        hashed_password=hash_password(admin_password),
+        role_name=admin_role.name
+    )
+    db.add(admin_user)
+    db.commit()
+    print("Admin user created.")
+
+
+def seed_all():
+    db = SessionLocal()
+    try:
+        clear_tables(db)  
+        seed_roles(db)
+        seed_permissions(db)
+        seed_rules(db)
+        seed_admin_user(db)
+    finally:
+        db.close()
+
+
+if __name__ == "__main__":
+    seed_all()

app/db/session.py

@@ -1,10 +1,11 @@
-# app/database/session.py
-
 import os
+from dotenv import load_dotenv
 from sqlalchemy import create_engine
 from sqlalchemy.orm import sessionmaker, Session
 from typing import Generator
 
+load_dotenv()
+
 DB_HOST = os.getenv("DB_HOST", "localhost")
 DB_PORT = os.getenv("DB_PORT", "5432")
 DB_USER = os.getenv("POSTGRES_USER", "postgres")