Implement Company & Product Management with JWT Authentication and PostgreSQL Integration

.gitignore

@@ -0,0 +1,7 @@
+# Ignore Python cache files
+__pycache__/
+*.pyc
+*.pyo
+
+# Ignore init files if you don’t want them tracked
+__init__.py

app/auth.py

@@ -0,0 +1,59 @@
+from datetime import datetime, timedelta
+from fastapi import Depends, HTTPException, status
+from fastapi.security import HTTPBearer
+from jose import JWTError, jwt
+from passlib.context import CryptContext
+from sqlalchemy.orm import Session
+from app.models.user import User
+from app.database import get_db
+from fastapi.security import HTTPAuthorizationCredentials
+
+
+SECRET_KEY = "af3287c8391bb9f4f7a72feb3b85f72e1d5bd07cbf4fa4ad9497c78412923312"
+ALGORITHM = "HS256"
+ACCESS_TOKEN_EXPIRE_MINUTES = 30
+
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+bearer_scheme = HTTPBearer()
+
+
+# ---------------- PASSWORD UTILS ---------------- #
+def verify_password(plain_password, hashed_password):
+    return pwd_context.verify(plain_password, hashed_password)
+
+
+def get_password_hash(password):
+    return pwd_context.hash(password)
+
+
+# ---------------- TOKEN CREATION ---------------- #
+def create_access_token(user_id: int):
+    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
+    to_encode = {"sub": str(user_id), "exp": expire}
+    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+
+
+# ---------------- VERIFY CURRENT USER ---------------- #
+def get_current_user(
+    token: HTTPAuthorizationCredentials = Depends(bearer_scheme),
+    db: Session = Depends(get_db)
+):
+    credential_exception = HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Could not validate credentials",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+
+    try:
+        payload = jwt.decode(token.credentials, SECRET_KEY,
+                             algorithms=[ALGORITHM])
+        user_id: str = payload.get("sub")
+        if user_id is None:
+            raise credential_exception
+    except JWTError:
+        raise credential_exception
+
+    user = db.query(User).filter(User.id == int(user_id)).first()
+    if user is None:
+        raise credential_exception
+    return user

app/controllers/auth_controller.py

@@ -0,0 +1,31 @@
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy.orm import Session
+from app.schemas.user import UserCreate, UserResponse
+from app.services.user_service import UserService
+from app.database import get_db
+from app.auth import create_access_token, verify_password
+from app.models.user import User  
+router = APIRouter()
+
+
+@router.post("/register", response_model=UserResponse)
+def register(user: UserCreate, db: Session = Depends(get_db)):
+    service = UserService(db)
+    try:
+        return service.create_user(user)
+    except ValueError as e:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
+
+
+@router.post("/token")
+def login(user: UserCreate, db: Session = Depends(get_db)):
+    # ✅ Query using database model
+    db_user = db.query(User).filter(User.username == user.username).first()
+
+    if not db_user or not verify_password(user.password, db_user.password):
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid credentials")
+
+    token = create_access_token(db_user.id)
+    return {"access_token": token, "token_type": "bearer"}

app/controllers/company_controller.py

@@ -0,0 +1,46 @@
+from fastapi import APIRouter, Depends
+from sqlalchemy.orm import Session
+from app.schemas.company import CompanyCreate, CompanyResponse
+from app.services.company_service import CompanyService
+from app.database import get_db
+from app.auth import get_current_user
+
+router = APIRouter()
+
+
+@router.post("/", response_model=CompanyResponse)
+def create_company(
+    company: CompanyCreate,
+    db: Session = Depends(get_db),
+    current_user=Depends(get_current_user)
+):
+    service = CompanyService(db)
+    return service.create_company(current_user.id, company)
+
+
+@router.get("/me", response_model=CompanyResponse)
+def get_my_company(
+    db: Session = Depends(get_db),
+    current_user=Depends(get_current_user)
+):
+    service = CompanyService(db)
+    return service.get_my_company(current_user.id)
+
+
+@router.put("/me", response_model=CompanyResponse)
+def edit_my_company(
+        company: CompanyCreate,
+        db: Session = Depends(get_db),
+        current_user=Depends(get_current_user)
+):
+    service = CompanyService(db)
+    return service.edit_company(current_user.id, company)
+
+
+@router.delete("/me", dependencies=[Depends(get_current_user)])
+def delete_my_company(
+    db: Session = Depends(get_db),
+    current_user=Depends(get_current_user)
+):
+    service = CompanyService(db)
+    return service.delete_company(current_user.id)

app/controllers/product_controller.py

@@ -0,0 +1,63 @@
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy.orm import Session
+from app.schemas.product import ProductCreate, ProductResponse
+from app.services.product_service import ProductService
+from app.services.company_service import CompanyService
+from app.database import get_db
+from app.auth import get_current_user
+
+
+router = APIRouter()
+
+
+@router.post("/", response_model=ProductResponse)
+def create_product(
+    product: ProductCreate,
+    db: Session = Depends(get_db),
+    current_user=Depends(get_current_user)
+):
+    company_service = CompanyService(db)
+    product_service = ProductService(db)
+    company = company_service.get_my_company(current_user.id)
+    if not company:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND, detail="no company found")
+
+    return product_service.create_product(company.id, product)
+
+
+@router.get("/", response_model=list[ProductResponse])
+def list_product(
+    db: Session = Depends(get_db),
+    current_user=Depends(get_current_user)
+):
+    product_service = ProductService(db)
+    return product_service.list_products()
+
+
+@router.get("/{product_id}", response_model=ProductResponse)
+def get_product_by_id(
+    product_id: int,
+    db: Session = Depends(get_db),
+    current_user=Depends(get_current_user)
+):
+    product_service = ProductService(db)
+    return product_service.get_product(product_id)
+
+
+@router.put("/{product_id}", response_model=ProductResponse)
+def update_product_by_id(
+        product_id: int,
+        product: ProductCreate,
+        db: Session = Depends(get_db),
+        current_user=Depends(get_current_user)):
+    product_service = ProductService(db)
+    return product_service.update_product(product_id,product)
+
+@router.delete("/{product_id}", response_model=ProductResponse)
+def delete_product_by_id(
+        product_id: int,
+        db: Session = Depends(get_db),
+        current_user=Depends(get_current_user)):
+    product_service = ProductService(db)
+    return product_service.delete_product(product_id)

database.py → app/database.py

@@ -1,7 +1,20 @@
 from sqlalchemy import create_engine
 from sqlalchemy.orm import sessionmaker
 
+from sqlalchemy.ext.declarative import declarative_base
+
+
 db_url = "postgresql://postgres:123@localhost:5432/inventory"
 
 engine = create_engine(db_url)
 SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+
+Base = declarative_base()
+
+def get_db():
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()

app/main.py

@@ -0,0 +1,52 @@
+from fastapi import FastAPI
+from app.database import Base, engine
+from app.controllers import auth_controller, company_controller, product_controller
+
+
+Base.metadata.create_all(bind=engine)
+version = "v1"
+app = FastAPI(title="Company & Product API",
+              description="in which user create their company",
+              version=version
+              )
+
+app.include_router(auth_controller.router,
+                   prefix=f"/api/{version}/auth", tags=["Auth"])
+app.include_router(company_controller.router,
+                   prefix=f"/api/{version}/company", tags=["Company"])
+app.include_router(product_controller.router,
+                   prefix=f"/api/{version}/product", tags=["Product"])
+
+
+@app.get("/")
+def root():
+    return {"message": "Welcome to Company API!"}
+
+
+# from fastapi import FastAPI
+# from fastapi.middleware.cors import CORSMiddleware
+# import db.database_model as database_model
+# from app.database import engine
+# from routes import file_routes, post_routes, products_routes, user_route
+
+# version = "v1"
+# app = FastAPI(title="Fastapi ",
+#               description="this is learning project.",
+#               version=version,)
+
+
+# database_model.Base.metadata.create_all(bind=engine)
+
+
+# app.include_router(products_routes.router,
+#                    prefix=f"/api/{version}/products", tags=['Products'])
+# app.include_router(file_routes.router,
+#                    prefix=f"/api/{version}/files", tags=['Files'])
+# app.include_router(user_route.router,
+#                    prefix=f"/api/{version}/users", tags=['Users'])
+# app.include_router(post_routes.router,
+#                    prefix=f"/api/{version}/posts", tags=["Posts"])
+
+# @app.get("/")
+# def greet():
+#     return {"message": "Hello, World!"}

app/models/company.py

@@ -0,0 +1,15 @@
+from sqlalchemy import Column, Integer, String, ForeignKey
+from sqlalchemy.orm import relationship
+from app.database import Base
+
+
+class Company(Base):
+    __tablename__ = "companies"
+
+    id =Column(Integer, primary_key=True,  index=True)
+    name= Column(String)
+    location = Column(String)
+    user_id = Column(Integer, ForeignKey("users.id"))
+
+    user= relationship("User", back_populates="company")
+    products= relationship("Product", back_populates="company")

app/models/model.py

@@ -0,0 +1,34 @@
+from pydantic import BaseModel
+from typing import Optional
+
+class Product(BaseModel):
+    id: int
+    name: str
+    description: str
+    price: float
+    quantity: int
+
+    class Config:
+        from_attributes = True
+
+
+class CreateUser(BaseModel):
+    username: str
+    password: str
+
+
+class User(BaseModel):
+    id: int
+    username: str
+    password: str
+
+
+class Post(BaseModel):
+    post_id: Optional [int] 
+    title: str
+    description: str
+
+
+class Token(BaseModel):
+    access_token: str
+    token_type: str

app/models/product.py

@@ -0,0 +1,15 @@
+from sqlalchemy import Column, Integer, String, Float, ForeignKey
+from sqlalchemy.orm import relationship
+from app.database import Base
+
+
+class Product(Base):
+    __tablename__ = "products"
+
+    id = Column(Integer, primary_key=True, index=True)
+    name = Column(String)
+    price = Column(Float)
+    description = Column(String,nullable=True)
+    company_id = Column(Integer, ForeignKey("companies.id"))
+
+    company = relationship("Company", back_populates="products")

app/models/user.py

@@ -0,0 +1,13 @@
+from sqlalchemy import Column, Integer, String
+from sqlalchemy.orm import relationship
+from app.database import Base
+
+
+class User(Base):
+    __tablename__ = "users"
+
+    id = Column(Integer, primary_key=True, index=True)
+    username = Column(String, unique=True)
+    password = Column(String)
+
+    company= relationship("Company",back_populates="user", uselist=False)

app/schemas/company.py

@@ -0,0 +1,18 @@
+from pydantic import BaseModel
+from typing import List, Optional
+from app.schemas.product import ProductResponse
+
+
+class CompanyCreate(BaseModel):
+    name: str
+    location: str
+
+
+class CompanyResponse(BaseModel):
+    id: int
+    name: str
+    location: str
+    products: List[ProductResponse] = []
+
+    class Config:
+        from_attributes = True