4aac37bc90
- Setup project structure with FastAPI application - Create database models with SQLAlchemy - Configure Alembic for database migrations - Implement CRUD operations for products, categories, suppliers - Add inventory transaction functionality - Implement user authentication with JWT - Add health check endpoint - Create comprehensive documentation
88 lines
3.4 KiB
Python
88 lines
3.4 KiB
Python
from typing import List
|
|
from sqlalchemy.orm import Session
|
|
from app.crud.base import CRUDBase
|
|
from app.models.inventory_transaction import InventoryTransaction, TransactionType
|
|
from app.schemas.inventory_transaction import InventoryTransactionCreate, InventoryTransactionUpdate
|
|
from datetime import datetime
|
|
|
|
|
|
class CRUDInventoryTransaction(CRUDBase[InventoryTransaction, InventoryTransactionCreate, InventoryTransactionUpdate]):
|
|
def create_with_product_update(
|
|
self, db: Session, *, obj_in: InventoryTransactionCreate, current_user_id: int
|
|
) -> InventoryTransaction:
|
|
# Create transaction with the current user
|
|
db_obj = InventoryTransaction(
|
|
transaction_type=obj_in.transaction_type,
|
|
quantity=obj_in.quantity,
|
|
unit_price=obj_in.unit_price,
|
|
transaction_date=obj_in.transaction_date or datetime.utcnow(),
|
|
notes=obj_in.notes,
|
|
reference_number=obj_in.reference_number,
|
|
product_id=obj_in.product_id,
|
|
user_id=current_user_id,
|
|
)
|
|
db.add(db_obj)
|
|
|
|
# Update product quantity
|
|
product = db.query(self.model).filter(
|
|
self.model.product_id == obj_in.product_id
|
|
).first().product
|
|
|
|
# Update quantity based on transaction type
|
|
if obj_in.transaction_type == TransactionType.PURCHASE:
|
|
product.quantity += obj_in.quantity
|
|
elif obj_in.transaction_type == TransactionType.SALE:
|
|
product.quantity -= obj_in.quantity
|
|
elif obj_in.transaction_type == TransactionType.ADJUSTMENT:
|
|
product.quantity += obj_in.quantity # Can be positive or negative
|
|
elif obj_in.transaction_type == TransactionType.RETURN:
|
|
if obj_in.notes and 'purchase' in obj_in.notes.lower():
|
|
# Return to supplier decreases quantity
|
|
product.quantity -= obj_in.quantity
|
|
else:
|
|
# Customer return increases quantity
|
|
product.quantity += obj_in.quantity
|
|
|
|
db.add(product)
|
|
db.commit()
|
|
db.refresh(db_obj)
|
|
return db_obj
|
|
|
|
def get_by_product(
|
|
self, db: Session, *, product_id: int, skip: int = 0, limit: int = 100
|
|
) -> List[InventoryTransaction]:
|
|
return (
|
|
db.query(InventoryTransaction)
|
|
.filter(InventoryTransaction.product_id == product_id)
|
|
.order_by(InventoryTransaction.transaction_date.desc())
|
|
.offset(skip)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
|
|
def get_by_user(
|
|
self, db: Session, *, user_id: int, skip: int = 0, limit: int = 100
|
|
) -> List[InventoryTransaction]:
|
|
return (
|
|
db.query(InventoryTransaction)
|
|
.filter(InventoryTransaction.user_id == user_id)
|
|
.order_by(InventoryTransaction.transaction_date.desc())
|
|
.offset(skip)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
|
|
def get_by_type(
|
|
self, db: Session, *, transaction_type: TransactionType, skip: int = 0, limit: int = 100
|
|
) -> List[InventoryTransaction]:
|
|
return (
|
|
db.query(InventoryTransaction)
|
|
.filter(InventoryTransaction.transaction_type == transaction_type)
|
|
.order_by(InventoryTransaction.transaction_date.desc())
|
|
.offset(skip)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
|
|
|
|
inventory_transaction = CRUDInventoryTransaction(InventoryTransaction) |