be0ae5f3b3
- Create project structure with FastAPI and SQLAlchemy - Implement database models for items, categories, suppliers, and stock movements - Add CRUD operations for all models - Configure Alembic for database migrations - Create RESTful API endpoints for inventory management - Add health endpoint for system monitoring - Update README with setup and usage instructions generated with BackendIM... (backend.im)
54 lines
2.1 KiB
Python
54 lines
2.1 KiB
Python
from sqlalchemy.orm import Session
|
|
from fastapi import HTTPException, status
|
|
from app.models.supplier import Supplier
|
|
from app.schemas.supplier import SupplierCreate, SupplierUpdate
|
|
|
|
def get_suppliers(db: Session, skip: int = 0, limit: int = 100, include_inactive: bool = False):
|
|
query = db.query(Supplier)
|
|
if not include_inactive:
|
|
query = query.filter(Supplier.is_active == True)
|
|
return query.offset(skip).limit(limit).all()
|
|
|
|
def get_supplier(db: Session, supplier_id: int):
|
|
return db.query(Supplier).filter(Supplier.id == supplier_id).first()
|
|
|
|
def get_supplier_by_name(db: Session, name: str):
|
|
return db.query(Supplier).filter(Supplier.name == name).first()
|
|
|
|
def create_supplier(db: Session, supplier: SupplierCreate):
|
|
existing_supplier = get_supplier_by_name(db, supplier.name)
|
|
if existing_supplier:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Supplier with this name already exists")
|
|
|
|
db_supplier = Supplier(**supplier.model_dump())
|
|
db.add(db_supplier)
|
|
db.commit()
|
|
db.refresh(db_supplier)
|
|
return db_supplier
|
|
|
|
def update_supplier(db: Session, supplier_id: int, supplier: SupplierUpdate):
|
|
db_supplier = get_supplier(db, supplier_id)
|
|
if not db_supplier:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Supplier not found")
|
|
|
|
if supplier.name and supplier.name != db_supplier.name:
|
|
existing_supplier = get_supplier_by_name(db, supplier.name)
|
|
if existing_supplier:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Supplier with this name already exists")
|
|
|
|
update_data = supplier.model_dump(exclude_unset=True)
|
|
for key, value in update_data.items():
|
|
setattr(db_supplier, key, value)
|
|
|
|
db.commit()
|
|
db.refresh(db_supplier)
|
|
return db_supplier
|
|
|
|
def delete_supplier(db: Session, supplier_id: int):
|
|
db_supplier = get_supplier(db, supplier_id)
|
|
if not db_supplier:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Supplier not found")
|
|
|
|
db_supplier.is_active = False
|
|
db.commit()
|
|
return {"message": "Supplier deleted successfully"} |