Checkpoint4

.python-version

@@ -1 +1 @@
-apivenv
+3.10.17

app/db/alembic/env.py

@@ -16,7 +16,6 @@
 from app.db.base import Base
 
 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)  
 

app/db/crud/orders.py

@@ -1,21 +1,44 @@
 from sqlalchemy.orm import Session
 from app.db.models import Order
+from app.validators.orders import OrderUpdate
 
 def get_order_by_id(db: Session, order_id: int) -> Order | None:
     return db.query(Order).filter(Order.id == order_id).first()
 
-def create_order(db: Session, user_id: int, total_amount, status: str = "pending") -> Order:
+
+def create_order(db: Session, user_id: int, total_amount: float, status: str = "pending") -> Order:
     order = Order(user_id=user_id, total_amount=total_amount, status=status)
     db.add(order)
     db.commit()
     db.refresh(order)
     return order
 
+
+def get_all_orders(db: Session) -> list[Order]:
+    return db.query(Order).all()
+
+
 def get_orders_by_user_id(db: Session, user_id: int) -> list[Order]:
     return db.query(Order).filter(Order.user_id == user_id).all()
 
+
+def update_order_db(db: Session, order_id: int, order_update: OrderUpdate) -> Order:
+    order = get_order_by_id(db, order_id)
+    if not order:
+        return None
+
+    # Update allowed fields only
+    for field, value in order_update.dict(exclude_unset=True).items():
+        setattr(order, field, value)
+
+    db.commit()
+    db.refresh(order)
+    return order
+
+
 def delete_order_by_id(db: Session, order_id: int):
     order = get_order_by_id(db, order_id)
     if order:
         db.delete(order)
         db.commit()
+    return 

app/db/crud/users.py

@@ -6,27 +6,33 @@
 def get_user_by_id(db: Session, user_id: int) -> User | None:
     return db.query(User).filter(User.id == user_id).first()
 
+
 def get_user_by_email(db: Session, email: str) -> User | None:
     return db.query(User).filter(User.email == email).first()
 
+
 def get_user_by_username(db: Session, username: str) -> User | None:
     return db.query(User).filter(User.username == username).first()
 
+
 def get_all_users(db: Session):
     return db.query(User).all()
 
+
 def create_user(db: Session, username: str, email: str, hashed_password: str, role: str = "customer") -> User:
     user = User(
         username=username,
         email=email,
         hashed_password=hashed_password,
-        role=role
+        role_name=role
     )
+
     db.add(user)
     db.commit()
     db.refresh(user)
     return user
 
+
 def update_user_db(db: Session, user_id: int, user_update: UserUpdate):
     user = get_user_by_id(db, user_id)
     if not user:
@@ -44,8 +50,9 @@ def update_user_db(db: Session, user_id: int, user_update: UserUpdate):
     db.refresh(user)
     return user
 
+
 def delete_user_by_id(db: Session, user_id: int):
     user = get_user_by_id(db, user_id)
     if user:
         db.delete(user)
-        db.commit()
+        db.commit()

app/db/seed.py

@@ -8,7 +8,7 @@
 from dotenv import load_dotenv
 from sqlalchemy import text
 
-load_dotenv() # Load environment variables from .env file
+load_dotenv()
 
 def clear_tables(db):
     # Truncate with restart identity to reset auto-increment IDs
@@ -37,43 +37,55 @@ def seed_permissions(db):
         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"),
+
+        # Order related permissions
+        Permission(name="can_create_order"),
+        Permission(name="can_list_orders"),
+        Permission(name="can_get_order"),
+        Permission(name="can_update_order"),
+        Permission(name="can_delete_order"),
+
+        Permission(name="can_get_own_orders"),
+        Permission(name="can_get_own_order"),
     ]
+
     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),
+        # Admin - User Management
+        Rule(role_name=admin_role.name, permission_name="can_create_user"),
+        Rule(role_name=admin_role.name, permission_name="can_list_users"),
+        Rule(role_name=admin_role.name, permission_name="can_get_user"),
+        Rule(role_name=admin_role.name, permission_name="can_update_user"),
+        Rule(role_name=admin_role.name, permission_name="can_delete_user"),
+
+        # Admin - Order Management
+        Rule(role_name=admin_role.name, permission_name="can_create_order"),
+        Rule(role_name=admin_role.name, permission_name="can_list_orders"),
+        Rule(role_name=admin_role.name, permission_name="can_get_order"),
+        Rule(role_name=admin_role.name, permission_name="can_update_order"),
+        Rule(role_name=admin_role.name, permission_name="can_delete_order"),
+
+        # Customer - Profile Management
+        Rule(role_name=customer_role.name, permission_name="can_get_own_profile"),
+        Rule(role_name=customer_role.name, permission_name="can_update_own_profile"),
+
+        # Customer - Order Management
+        Rule(role_name=customer_role.name, permission_name="can_create_order"),
+        Rule(role_name=customer_role.name, permission_name="can_get_own_orders"),
+        Rule(role_name=customer_role.name, permission_name="can_get_own_order"),
     ]
 
     for rule in rules:
@@ -109,6 +121,25 @@ def seed_admin_user(db):
     print("Admin user created.")
 
 
+def seed_customer_users(db, n=3):
+    customer_role = db.query(Role).filter(Role.name == "customer").first()
+
+    customer_usernames = [f"customer{x}" for x in range(1, n+1)]
+    customer_emails = [f"customer{x}@example.com" for x in range(1,n+1)]
+    customer_passwords = [f"password{x}" for x in range(1, n+1)]
+
+    for customer_username, customer_email, customer_password in zip(customer_usernames, customer_emails, customer_passwords):
+        customer_user = User(
+            username=customer_username,
+            email=customer_email,
+            hashed_password=hash_password(customer_password),
+            role_name=customer_role.name
+        )
+        db.add(customer_user)
+        db.commit()
+        print("Customer user created.")
+
+
 def seed_all():
     db = SessionLocal()
     try:
@@ -117,6 +148,8 @@ def seed_all():
         seed_permissions(db)
         seed_rules(db)
         seed_admin_user(db)
+        seed_customer_users(db, n=3)
+
     finally:
         db.close()
 

app/routers/auth.py

@@ -1,4 +1,4 @@
-from fastapi import APIRouter, Depends, HTTPException, status
+from fastapi import APIRouter, Depends, HTTPException
 from fastapi.security import OAuth2PasswordRequestForm
 from sqlalchemy.orm import Session
 from app.db.session import get_db
@@ -18,7 +18,6 @@ def login(
     if not user:
         raise HTTPException(status_code=401, detail="Invalid credentials")
 
-    # Fetch permissions for user's role from DB
     permissions = get_permissions_for_role(db, user.role_name)
 
     user_data = {